diff --git a/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQuery.java b/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQuery.java index 66e3dcbfa957..58f8872aec4f 100644 --- a/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQuery.java +++ b/core/trino-main/src/main/java/io/trino/dispatcher/FailedDispatchQuery.java @@ -219,6 +219,7 @@ private static QueryInfo immediateFailureQueryInfo( query, preparedQuery, immediateFailureQueryStats(), + immediateFailureQueryStats(), Optional.empty(), Optional.empty(), Optional.empty(), diff --git a/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQuery.java b/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQuery.java index 77cbce851d06..69237df657de 100644 --- a/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQuery.java +++ b/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQuery.java @@ -181,7 +181,7 @@ public DispatchInfo getDispatchInfo() { // observe submitted before getting the state, to ensure a failed query stat is visible boolean dispatched = submitted.isDone(); - BasicQueryInfo queryInfo = stateMachine.getBasicQueryInfo(Optional.empty()); + BasicQueryInfo queryInfo = stateMachine.getBasicQueryInfo(Optional.empty(), Optional.empty()); if (queryInfo.getState() == QueryState.FAILED) { ExecutionFailureInfo failureInfo = stateMachine.getFailureInfo() @@ -259,7 +259,7 @@ public BasicQueryInfo getBasicQueryInfo() { return tryGetQueryExecution() .map(QueryExecution::getBasicQueryInfo) - .orElseGet(() -> stateMachine.getBasicQueryInfo(Optional.empty())); + .orElseGet(() -> stateMachine.getBasicQueryInfo(Optional.empty(), Optional.empty())); } @Override diff --git a/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQueryFactory.java b/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQueryFactory.java index 083117bc8e49..12ecfce8d2d8 100644 --- a/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQueryFactory.java +++ b/core/trino-main/src/main/java/io/trino/dispatcher/LocalDispatchQueryFactory.java @@ -123,7 +123,7 @@ public DispatchQuery createDispatchQuery( // // Note that for immediate and in-order delivery of query events we depend on synchronous nature of // QueryMonitor and EventListenerManager. - queryMonitor.queryCreatedEvent(stateMachine.getBasicQueryInfo(Optional.empty())); + queryMonitor.queryCreatedEvent(stateMachine.getBasicQueryInfo(Optional.empty(), Optional.empty())); ListenableFuture queryExecutionFuture = executor.submit(() -> { QueryExecutionFactory queryExecutionFactory = executionFactories.get(preparedQuery.getStatement().getClass()); diff --git a/core/trino-main/src/main/java/io/trino/execution/DataDefinitionExecution.java b/core/trino-main/src/main/java/io/trino/execution/DataDefinitionExecution.java index dd12a8913050..37368928767f 100644 --- a/core/trino-main/src/main/java/io/trino/execution/DataDefinitionExecution.java +++ b/core/trino-main/src/main/java/io/trino/execution/DataDefinitionExecution.java @@ -129,7 +129,7 @@ public BasicQueryInfo getBasicQueryInfo() { return stateMachine.getFinalQueryInfo() .map(BasicQueryInfo::new) - .orElseGet(() -> stateMachine.getBasicQueryInfo(Optional.empty())); + .orElseGet(() -> stateMachine.getBasicQueryInfo(Optional.empty(), Optional.empty())); } @Override diff --git a/core/trino-main/src/main/java/io/trino/execution/QueryInfo.java b/core/trino-main/src/main/java/io/trino/execution/QueryInfo.java index 91e37e95fc65..98cdef768009 100644 --- a/core/trino-main/src/main/java/io/trino/execution/QueryInfo.java +++ b/core/trino-main/src/main/java/io/trino/execution/QueryInfo.java @@ -56,6 +56,7 @@ public class QueryInfo private final String query; private final Optional preparedQuery; private final QueryStats queryStats; + private final QueryStats queryStatsNoFailedTasks; private final Optional setCatalog; private final Optional setSchema; private final Optional setPath; @@ -91,6 +92,7 @@ public QueryInfo( @JsonProperty("query") String query, @JsonProperty("preparedQuery") Optional preparedQuery, @JsonProperty("queryStats") QueryStats queryStats, + @JsonProperty("queryStatsNoFailedTasks") QueryStats queryStatsNoFailedTasks, @JsonProperty("setCatalog") Optional setCatalog, @JsonProperty("setSchema") Optional setSchema, @JsonProperty("setPath") Optional setPath, @@ -120,6 +122,7 @@ public QueryInfo( requireNonNull(self, "self is null"); requireNonNull(fieldNames, "fieldNames is null"); requireNonNull(queryStats, "queryStats is null"); + requireNonNull(queryStatsNoFailedTasks, "queryStatsNoFailedTasks is null"); requireNonNull(setCatalog, "setCatalog is null"); requireNonNull(setSchema, "setSchema is null"); requireNonNull(setPath, "setPath is null"); @@ -148,6 +151,7 @@ public QueryInfo( this.query = query; this.preparedQuery = preparedQuery; this.queryStats = queryStats; + this.queryStatsNoFailedTasks = queryStatsNoFailedTasks; this.setCatalog = setCatalog; this.setSchema = setSchema; this.setPath = setPath; @@ -227,6 +231,12 @@ public QueryStats getQueryStats() return queryStats; } + @JsonProperty + public QueryStats getQueryStatsNoFailedTasks() + { + return queryStatsNoFailedTasks; + } + @JsonProperty public Optional getSetCatalog() { diff --git a/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java b/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java index 12edc2096b2c..e0abe84fb2af 100644 --- a/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java +++ b/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java @@ -348,7 +348,7 @@ public void updateMemoryUsage( peakTaskTotalMemory.accumulateAndGet(taskTotalMemoryInBytes, Math::max); } - public BasicQueryInfo getBasicQueryInfo(Optional rootStage) + public BasicQueryInfo getBasicQueryInfo(Optional rootStageStatsOptional, Optional rootStageStatsNoFailedTasksOptional) { // Query state must be captured first in order to provide a // correct view of the query. For example, building this @@ -364,7 +364,30 @@ public BasicQueryInfo getBasicQueryInfo(Optional rootStage) } } - BasicStageStats stageStats = rootStage.orElse(EMPTY_STAGE_STATS); + BasicStageStats rootStageStats = rootStageStatsOptional.orElse(EMPTY_STAGE_STATS); + BasicStageStats rootStageStatsNoFailedTasks = rootStageStatsNoFailedTasksOptional.orElse(EMPTY_STAGE_STATS); + BasicQueryStats queryStats = getBasicQueryStats(rootStageStats); + BasicQueryStats queryStatsNoFailedTasks = getBasicQueryStats(rootStageStatsNoFailedTasks); + + return new BasicQueryInfo( + queryId, + session.toSessionRepresentation(), + Optional.of(resourceGroup), + state, + rootStageStats.isScheduled(), + self, + query, + Optional.ofNullable(updateType.get()), + preparedQuery, + queryStats, + queryStatsNoFailedTasks, + errorCode == null ? null : errorCode.getType(), + errorCode, + queryType); + } + + private BasicQueryStats getBasicQueryStats(BasicStageStats stageStats) + { BasicQueryStats queryStats = new BasicQueryStats( queryStateTimer.getCreateTime(), getEndTime().orElse(null), @@ -393,21 +416,7 @@ public BasicQueryInfo getBasicQueryInfo(Optional rootStage) stageStats.isFullyBlocked(), stageStats.getBlockedReasons(), stageStats.getProgressPercentage()); - - return new BasicQueryInfo( - queryId, - session.toSessionRepresentation(), - Optional.of(resourceGroup), - state, - stageStats.isScheduled(), - self, - query, - Optional.ofNullable(updateType.get()), - preparedQuery, - queryStats, - errorCode == null ? null : errorCode.getType(), - errorCode, - queryType); + return queryStats; } @VisibleForTesting @@ -441,6 +450,7 @@ QueryInfo getQueryInfo(Optional rootStage) query, preparedQuery, getQueryStats(rootStage), + getQueryStatsNoFailedTasks(rootStage), Optional.ofNullable(setCatalog.get()), Optional.ofNullable(setSchema.get()), Optional.ofNullable(setPath.get()), @@ -466,6 +476,16 @@ QueryInfo getQueryInfo(Optional rootStage) } private QueryStats getQueryStats(Optional rootStage) + { + return getQueryStats(rootStage, true); + } + + private QueryStats getQueryStatsNoFailedTasks(Optional rootStage) + { + return getQueryStats(rootStage, false); + } + + private QueryStats getQueryStats(Optional rootStage, boolean includeFailedTasks) { int totalTasks = 0; int runningTasks = 0; @@ -512,7 +532,7 @@ private QueryStats getQueryStats(Optional rootStage) ImmutableList.Builder operatorStatsSummary = ImmutableList.builder(); boolean completeInfo = true; for (StageInfo stageInfo : getAllStages(rootStage)) { - StageStats stageStats = stageInfo.getStageStats(); + StageStats stageStats = includeFailedTasks ? stageInfo.getStageStats() : stageInfo.getStageStatsNoFailedTasks(); totalTasks += stageStats.getTotalTasks(); runningTasks += stageStats.getRunningTasks(); completedTasks += stageStats.getCompletedTasks(); @@ -556,11 +576,11 @@ private QueryStats getQueryStats(Optional rootStage) stageGcStatistics.add(stageStats.getGcInfo()); completeInfo = completeInfo && stageInfo.isCompleteInfo(); - operatorStatsSummary.addAll(stageInfo.getStageStats().getOperatorSummaries()); + operatorStatsSummary.addAll(stageStats.getOperatorSummaries()); } if (rootStage.isPresent()) { - StageStats outputStageStats = rootStage.get().getStageStats(); + StageStats outputStageStats = includeFailedTasks ? rootStage.get().getStageStats() : rootStage.get().getStageStatsNoFailedTasks(); outputDataSize += outputStageStats.getOutputDataSize().toBytes(); outputPositions += outputStageStats.getOutputPositions(); } @@ -1080,6 +1100,7 @@ public void pruneQueryInfo() outputStage.isCoordinatorOnly(), outputStage.getTypes(), outputStage.getStageStats(), + outputStage.getStageStatsNoFailedTasks(), ImmutableList.of(), // Remove the tasks ImmutableList.of(), // Remove the substages ImmutableMap.of(), // Remove tables @@ -1095,6 +1116,7 @@ public void pruneQueryInfo() queryInfo.getQuery(), queryInfo.getPreparedQuery(), pruneQueryStats(queryInfo.getQueryStats()), + pruneQueryStats(queryInfo.getQueryStatsNoFailedTasks()), queryInfo.getSetCatalog(), queryInfo.getSetSchema(), queryInfo.getSetPath(), diff --git a/core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java b/core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java index a0fff71f1121..bbe9c379d10c 100644 --- a/core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java +++ b/core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java @@ -358,7 +358,12 @@ public BasicQueryInfo getBasicQueryInfo() { return stateMachine.getFinalQueryInfo() .map(BasicQueryInfo::new) - .orElseGet(() -> stateMachine.getBasicQueryInfo(Optional.ofNullable(queryScheduler.get()).map(SqlQueryScheduler::getBasicStageStats))); + .orElseGet(() -> { + Optional sqlQueryScheduler = Optional.ofNullable(queryScheduler.get()); + Optional basicStageStats = sqlQueryScheduler.map(SqlQueryScheduler::getBasicStageStats); + Optional basicStageStatsNoFailedTasks = sqlQueryScheduler.map(SqlQueryScheduler::getBasicStageStatsNoFailedTasks); + return stateMachine.getBasicQueryInfo(basicStageStats, basicStageStatsNoFailedTasks); + }); } @Override diff --git a/core/trino-main/src/main/java/io/trino/execution/SqlStage.java b/core/trino-main/src/main/java/io/trino/execution/SqlStage.java index 82ab33861f81..5afcb98b8877 100644 --- a/core/trino-main/src/main/java/io/trino/execution/SqlStage.java +++ b/core/trino-main/src/main/java/io/trino/execution/SqlStage.java @@ -193,6 +193,11 @@ public BasicStageStats getBasicStageStats() return stateMachine.getBasicStageStats(this::getAllTaskInfo); } + public BasicStageStats getBasicStageStatsNoFailedTasks() + { + return stateMachine.getBasicStageStats(this::getNoFailedTaskInfo); + } + public StageInfo getStageInfo() { return stateMachine.getStageInfo(this::getAllTaskInfo); @@ -205,6 +210,14 @@ private Iterable getAllTaskInfo() .collect(toImmutableList()); } + private Iterable getNoFailedTaskInfo() + { + return tasks.values().stream() + .map(RemoteTask::getTaskInfo) + .filter(task -> task.getTaskStatus().getState() != TaskState.FAILED) // todo: should we filter out tasks which were cancelled as a reason of other tasks failing? + .collect(toImmutableList()); + } + public synchronized Optional createTask( InternalNode node, int partition, diff --git a/core/trino-main/src/main/java/io/trino/execution/StageInfo.java b/core/trino-main/src/main/java/io/trino/execution/StageInfo.java index ffd7172ab6bf..659f32d2781f 100644 --- a/core/trino-main/src/main/java/io/trino/execution/StageInfo.java +++ b/core/trino-main/src/main/java/io/trino/execution/StageInfo.java @@ -40,6 +40,7 @@ public class StageInfo private final boolean coordinatorOnly; private final List types; private final StageStats stageStats; + private final StageStats stageStatsNoFailedTasks; private final List tasks; private final List subStages; private final ExecutionFailureInfo failureCause; @@ -53,6 +54,7 @@ public StageInfo( @JsonProperty("coordinatorOnly") boolean coordinatorOnly, @JsonProperty("types") List types, @JsonProperty("stageStats") StageStats stageStats, + @JsonProperty("stageStatsNoFailedTasks") StageStats stageStatsNoFailedTasks, @JsonProperty("tasks") List tasks, @JsonProperty("subStages") List subStages, @JsonProperty("tables") Map tables, @@ -61,6 +63,7 @@ public StageInfo( requireNonNull(stageId, "stageId is null"); requireNonNull(state, "state is null"); requireNonNull(stageStats, "stageStats is null"); + requireNonNull(stageStatsNoFailedTasks, "stageStatsNoFailedTasks is null"); requireNonNull(tasks, "tasks is null"); requireNonNull(subStages, "subStages is null"); requireNonNull(tables, "tables is null"); @@ -71,6 +74,7 @@ public StageInfo( this.coordinatorOnly = coordinatorOnly; this.types = types; this.stageStats = stageStats; + this.stageStatsNoFailedTasks = stageStatsNoFailedTasks; this.tasks = ImmutableList.copyOf(tasks); this.subStages = subStages; this.failureCause = failureCause; @@ -114,6 +118,12 @@ public StageStats getStageStats() return stageStats; } + @JsonProperty + public StageStats getStageStatsNoFailedTasks() + { + return stageStatsNoFailedTasks; + } + @JsonProperty public List getTasks() { diff --git a/core/trino-main/src/main/java/io/trino/execution/StageStateMachine.java b/core/trino-main/src/main/java/io/trino/execution/StageStateMachine.java index 950b0fd8708d..20d9c473a372 100644 --- a/core/trino-main/src/main/java/io/trino/execution/StageStateMachine.java +++ b/core/trino-main/src/main/java/io/trino/execution/StageStateMachine.java @@ -348,6 +348,26 @@ public StageInfo getStageInfo(Supplier> taskInfosSupplier) List taskInfos = ImmutableList.copyOf(taskInfosSupplier.get()); + ExecutionFailureInfo failureInfo = null; + if (state == FAILED) { + failureInfo = failureCause.get(); + } + return new StageInfo( + stageId, + state, + fragment, + fragment.getPartitioning().isCoordinatorOnly(), + fragment.getTypes(), + getStageStats(taskInfos, true), + getStageStats(taskInfos, false), + taskInfos, + ImmutableList.of(), + tables, + failureInfo); + } + + private StageStats getStageStats(List taskInfos, boolean includeFailedTasks) + { int totalTasks = taskInfos.size(); int runningTasks = 0; int completedTasks = 0; @@ -400,6 +420,10 @@ public StageInfo getStageInfo(Supplier> taskInfosSupplier) Map operatorToStats = new HashMap<>(); for (TaskInfo taskInfo : taskInfos) { + if (!includeFailedTasks && taskInfo.getTaskStatus().getState() == TaskState.FAILED) { + continue; + } + TaskState taskState = taskInfo.getTaskStatus().getState(); if (taskState.isDone()) { completedTasks++; @@ -471,7 +495,7 @@ public StageInfo getStageInfo(Supplier> taskInfosSupplier) } } - StageStats stageStats = new StageStats( + return new StageStats( schedulingComplete.get(), getSplitDistribution.snapshot(), @@ -525,22 +549,6 @@ public StageInfo getStageInfo(Supplier> taskInfosSupplier) (int) (1.0 * totalFullGcSec / fullGcCount)), ImmutableList.copyOf(operatorToStats.values())); - - ExecutionFailureInfo failureInfo = null; - if (state == FAILED) { - failureInfo = failureCause.get(); - } - return new StageInfo( - stageId, - state, - fragment, - fragment.getPartitioning().isCoordinatorOnly(), - fragment.getTypes(), - stageStats, - taskInfos, - ImmutableList.of(), - tables, - failureInfo); } public void recordGetSplitTime(long startNanos) diff --git a/core/trino-main/src/main/java/io/trino/execution/scheduler/SqlQueryScheduler.java b/core/trino-main/src/main/java/io/trino/execution/scheduler/SqlQueryScheduler.java index 1f036318cf8b..800142809391 100644 --- a/core/trino-main/src/main/java/io/trino/execution/scheduler/SqlQueryScheduler.java +++ b/core/trino-main/src/main/java/io/trino/execution/scheduler/SqlQueryScheduler.java @@ -480,6 +480,11 @@ public BasicStageStats getBasicStageStats() return stageManager.getBasicStageStats(); } + public BasicStageStats getBasicStageStatsNoFailedTasks() + { + return stageManager.getBasicStageStatsNoFailedTasks(); + } + public StageInfo getStageInfo() { return stageManager.getStageInfo(); @@ -675,9 +680,19 @@ public Optional getParent(StageId stageId) } public BasicStageStats getBasicStageStats() + { + return getBasicStageStats(true); + } + + public BasicStageStats getBasicStageStatsNoFailedTasks() + { + return getBasicStageStats(false); + } + + private BasicStageStats getBasicStageStats(boolean includeFailedTasks) { List stageStats = stages.values().stream() - .map(SqlStage::getBasicStageStats) + .map(includeFailedTasks ? SqlStage::getBasicStageStats : SqlStage::getBasicStageStatsNoFailedTasks) .collect(toImmutableList()); return aggregateBasicStageStats(stageStats); @@ -709,6 +724,7 @@ private StageInfo buildStageInfo(StageId stageId, Map stageI parent.isCoordinatorOnly(), parent.getTypes(), parent.getStageStats(), + parent.getStageStatsNoFailedTasks(), parent.getTasks(), childStages, parent.getTables(), diff --git a/core/trino-main/src/main/java/io/trino/server/BasicQueryInfo.java b/core/trino-main/src/main/java/io/trino/server/BasicQueryInfo.java index 1bd034e26a97..19908132a713 100644 --- a/core/trino-main/src/main/java/io/trino/server/BasicQueryInfo.java +++ b/core/trino-main/src/main/java/io/trino/server/BasicQueryInfo.java @@ -50,6 +50,7 @@ public class BasicQueryInfo private final Optional updateType; private final Optional preparedQuery; private final BasicQueryStats queryStats; + private final BasicQueryStats queryStatsNoFailedTasks; private final ErrorType errorType; private final ErrorCode errorCode; private final Optional queryType; @@ -66,6 +67,7 @@ public BasicQueryInfo( @JsonProperty("updateType") Optional updateType, @JsonProperty("preparedQuery") Optional preparedQuery, @JsonProperty("queryStats") BasicQueryStats queryStats, + @JsonProperty("queryStatsNoFailedTasks") BasicQueryStats queryStatsNoFailedTasks, @JsonProperty("errorType") ErrorType errorType, @JsonProperty("errorCode") ErrorCode errorCode, @JsonProperty("queryType") Optional queryType) @@ -82,6 +84,7 @@ public BasicQueryInfo( this.updateType = requireNonNull(updateType, "updateType is null"); this.preparedQuery = requireNonNull(preparedQuery, "preparedQuery is null"); this.queryStats = requireNonNull(queryStats, "queryStats is null"); + this.queryStatsNoFailedTasks = requireNonNull(queryStatsNoFailedTasks, "queryStatsNoFailedTasks is null"); this.queryType = requireNonNull(queryType, "queryType is null"); } @@ -97,6 +100,7 @@ public BasicQueryInfo(QueryInfo queryInfo) Optional.ofNullable(queryInfo.getUpdateType()), queryInfo.getPreparedQuery(), new BasicQueryStats(queryInfo.getQueryStats()), + new BasicQueryStats(queryInfo.getQueryStatsNoFailedTasks()), queryInfo.getErrorType(), queryInfo.getErrorCode(), queryInfo.getQueryType()); @@ -162,6 +166,12 @@ public BasicQueryStats getQueryStats() return queryStats; } + @JsonProperty + public BasicQueryStats getQueryStatsNoFailedTasks() + { + return queryStatsNoFailedTasks; + } + @Nullable @JsonProperty public ErrorType getErrorType() diff --git a/core/trino-main/src/test/java/io/trino/execution/MockManagedQueryExecution.java b/core/trino-main/src/test/java/io/trino/execution/MockManagedQueryExecution.java index f25bf98c422e..75422fda862f 100644 --- a/core/trino-main/src/test/java/io/trino/execution/MockManagedQueryExecution.java +++ b/core/trino-main/src/test/java/io/trino/execution/MockManagedQueryExecution.java @@ -107,6 +107,29 @@ public Optional getErrorCode() @Override public BasicQueryInfo getBasicQueryInfo() { + BasicQueryStats queryStats = new BasicQueryStats( + new DateTime(1), + new DateTime(2), + new Duration(3, NANOSECONDS), + new Duration(4, NANOSECONDS), + new Duration(5, NANOSECONDS), + 6, + 7, + 8, + 9, + DataSize.ofBytes(14), + 15, + DataSize.ofBytes(13), + 16.0, + memoryUsage, + memoryUsage, + DataSize.ofBytes(19), + DataSize.ofBytes(20), + cpuUsage, + new Duration(22, NANOSECONDS), + false, + ImmutableSet.of(), + OptionalDouble.empty()); return new BasicQueryInfo( new QueryId("test"), session.toSessionRepresentation(), @@ -117,29 +140,8 @@ public BasicQueryInfo getBasicQueryInfo() "SELECT 1", Optional.empty(), Optional.empty(), - new BasicQueryStats( - new DateTime(1), - new DateTime(2), - new Duration(3, NANOSECONDS), - new Duration(4, NANOSECONDS), - new Duration(5, NANOSECONDS), - 6, - 7, - 8, - 9, - DataSize.ofBytes(14), - 15, - DataSize.ofBytes(13), - 16.0, - memoryUsage, - memoryUsage, - DataSize.ofBytes(19), - DataSize.ofBytes(20), - cpuUsage, - new Duration(22, NANOSECONDS), - false, - ImmutableSet.of(), - OptionalDouble.empty()), + queryStats, + queryStats, null, null, Optional.empty()); @@ -148,6 +150,70 @@ public BasicQueryInfo getBasicQueryInfo() @Override public QueryInfo getFullQueryInfo() { + QueryStats queryStats = new QueryStats( + new DateTime(1), + new DateTime(2), + new DateTime(3), + new DateTime(4), + new Duration(6, NANOSECONDS), + new Duration(5, NANOSECONDS), + new Duration(31, NANOSECONDS), + new Duration(41, NANOSECONDS), + new Duration(7, NANOSECONDS), + new Duration(8, NANOSECONDS), + + new Duration(100, NANOSECONDS), + new Duration(200, NANOSECONDS), + + 9, + 10, + 11, + + 12, + 13, + 15, + 30, + 16, + + 17.0, + DataSize.ofBytes(18), + DataSize.ofBytes(19), + DataSize.ofBytes(20), + DataSize.ofBytes(21), + DataSize.ofBytes(22), + DataSize.ofBytes(23), + DataSize.ofBytes(24), + DataSize.ofBytes(25), + DataSize.ofBytes(26), + + true, + new Duration(20, NANOSECONDS), + new Duration(21, NANOSECONDS), + new Duration(23, NANOSECONDS), + false, + ImmutableSet.of(), + + DataSize.ofBytes(241), + 251, + new Duration(24, NANOSECONDS), + + DataSize.ofBytes(242), + 252, + + DataSize.ofBytes(25), + 26, + + DataSize.ofBytes(27), + 28, + + DataSize.ofBytes(29), + 30, + + DataSize.ofBytes(31), + + ImmutableList.of(), + DynamicFiltersStats.EMPTY, + ImmutableList.of()); return new QueryInfo( new QueryId("test"), session.toSessionRepresentation(), @@ -157,70 +223,8 @@ public QueryInfo getFullQueryInfo() ImmutableList.of(), "SELECT 1", Optional.empty(), - new QueryStats( - new DateTime(1), - new DateTime(2), - new DateTime(3), - new DateTime(4), - new Duration(6, NANOSECONDS), - new Duration(5, NANOSECONDS), - new Duration(31, NANOSECONDS), - new Duration(41, NANOSECONDS), - new Duration(7, NANOSECONDS), - new Duration(8, NANOSECONDS), - - new Duration(100, NANOSECONDS), - new Duration(200, NANOSECONDS), - - 9, - 10, - 11, - - 12, - 13, - 15, - 30, - 16, - - 17.0, - DataSize.ofBytes(18), - DataSize.ofBytes(19), - DataSize.ofBytes(20), - DataSize.ofBytes(21), - DataSize.ofBytes(22), - DataSize.ofBytes(23), - DataSize.ofBytes(24), - DataSize.ofBytes(25), - DataSize.ofBytes(26), - - true, - new Duration(20, NANOSECONDS), - new Duration(21, NANOSECONDS), - new Duration(23, NANOSECONDS), - false, - ImmutableSet.of(), - - DataSize.ofBytes(241), - 251, - new Duration(24, NANOSECONDS), - - DataSize.ofBytes(242), - 252, - - DataSize.ofBytes(25), - 26, - - DataSize.ofBytes(27), - 28, - - DataSize.ofBytes(29), - 30, - - DataSize.ofBytes(31), - - ImmutableList.of(), - DynamicFiltersStats.EMPTY, - ImmutableList.of()), + queryStats, + queryStats, Optional.empty(), Optional.empty(), Optional.empty(), diff --git a/core/trino-main/src/test/java/io/trino/server/TestBasicQueryInfo.java b/core/trino-main/src/test/java/io/trino/server/TestBasicQueryInfo.java index 72eee365a042..5cb2e2a59f3d 100644 --- a/core/trino-main/src/test/java/io/trino/server/TestBasicQueryInfo.java +++ b/core/trino-main/src/test/java/io/trino/server/TestBasicQueryInfo.java @@ -54,6 +54,65 @@ public void testConstructor() ImmutableList.of("2", "3"), "SELECT 4", Optional.empty(), + new QueryStats( + DateTime.parse("2991-09-06T05:00-05:30"), + DateTime.parse("2991-09-06T05:01-05:30"), + DateTime.parse("2991-09-06T05:02-05:30"), + DateTime.parse("2991-09-06T06:00-05:30"), + new Duration(18, MINUTES), + new Duration(17, MINUTES), + new Duration(135, MINUTES), + new Duration(135, MINUTES), + new Duration(144, MINUTES), + new Duration(19, MINUTES), + new Duration(199, SECONDS), + new Duration(112, MINUTES), + 113, + 114, + 115, + 116, + 117, + 118, + 134, + 119, + 120.0, + DataSize.valueOf("121GB"), + DataSize.valueOf("122GB"), + DataSize.valueOf("123GB"), + DataSize.valueOf("124GB"), + DataSize.valueOf("125GB"), + DataSize.valueOf("126GB"), + DataSize.valueOf("127GB"), + DataSize.valueOf("128GB"), + DataSize.valueOf("129GB"), + true, + new Duration(123, MINUTES), + new Duration(124, MINUTES), + new Duration(126, MINUTES), + true, + ImmutableSet.of(BlockedReason.WAITING_FOR_MEMORY), + DataSize.valueOf("1271GB"), + 1281, + new Duration(120, MINUTES), + DataSize.valueOf("1272GB"), + 1282, + DataSize.valueOf("127GB"), + 128, + DataSize.valueOf("129GB"), + 130, + DataSize.valueOf("131GB"), + 132, + DataSize.valueOf("132GB"), + ImmutableList.of(new StageGcStatistics( + 1101, + 1102, + 1103, + 1104, + 1105, + 1106, + 1107)), + DynamicFiltersStats.EMPTY, + ImmutableList.of()), new QueryStats( DateTime.parse("1991-09-06T05:00-05:30"), DateTime.parse("1991-09-06T05:01-05:30"), @@ -142,27 +201,48 @@ public void testConstructor() assertEquals(basicInfo.getQuery(), "SELECT 4"); assertEquals(basicInfo.getQueryType().get(), QueryType.SELECT); - assertEquals(basicInfo.getQueryStats().getCreateTime(), DateTime.parse("1991-09-06T05:00-05:30")); - assertEquals(basicInfo.getQueryStats().getEndTime(), DateTime.parse("1991-09-06T06:00-05:30")); - assertEquals(basicInfo.getQueryStats().getElapsedTime(), new Duration(8, MINUTES)); - assertEquals(basicInfo.getQueryStats().getExecutionTime(), new Duration(44, MINUTES)); + assertEquals(basicInfo.getQueryStats().getCreateTime(), DateTime.parse("2991-09-06T05:00-05:30")); + assertEquals(basicInfo.getQueryStats().getEndTime(), DateTime.parse("2991-09-06T06:00-05:30")); + assertEquals(basicInfo.getQueryStats().getElapsedTime(), new Duration(18, MINUTES)); + assertEquals(basicInfo.getQueryStats().getExecutionTime(), new Duration(144, MINUTES)); - assertEquals(basicInfo.getQueryStats().getTotalDrivers(), 16); - assertEquals(basicInfo.getQueryStats().getQueuedDrivers(), 17); - assertEquals(basicInfo.getQueryStats().getRunningDrivers(), 18); - assertEquals(basicInfo.getQueryStats().getCompletedDrivers(), 19); + assertEquals(basicInfo.getQueryStats().getTotalDrivers(), 116); + assertEquals(basicInfo.getQueryStats().getQueuedDrivers(), 117); + assertEquals(basicInfo.getQueryStats().getRunningDrivers(), 118); + assertEquals(basicInfo.getQueryStats().getCompletedDrivers(), 119); - assertEquals(basicInfo.getQueryStats().getCumulativeUserMemory(), 20.0); - assertEquals(basicInfo.getQueryStats().getUserMemoryReservation(), DataSize.valueOf("21GB")); - assertEquals(basicInfo.getQueryStats().getTotalMemoryReservation(), DataSize.valueOf("23GB")); - assertEquals(basicInfo.getQueryStats().getPeakUserMemoryReservation(), DataSize.valueOf("24GB")); - assertEquals(basicInfo.getQueryStats().getTotalCpuTime(), new Duration(24, MINUTES)); + assertEquals(basicInfo.getQueryStats().getCumulativeUserMemory(), 120.0); + assertEquals(basicInfo.getQueryStats().getUserMemoryReservation(), DataSize.valueOf("121GB")); + assertEquals(basicInfo.getQueryStats().getTotalMemoryReservation(), DataSize.valueOf("123GB")); + assertEquals(basicInfo.getQueryStats().getPeakUserMemoryReservation(), DataSize.valueOf("124GB")); + assertEquals(basicInfo.getQueryStats().getTotalCpuTime(), new Duration(124, MINUTES)); assertEquals(basicInfo.getQueryStats().isFullyBlocked(), true); assertEquals(basicInfo.getQueryStats().getBlockedReasons(), ImmutableSet.of(BlockedReason.WAITING_FOR_MEMORY)); assertEquals(basicInfo.getQueryStats().getProgressPercentage(), OptionalDouble.of(100)); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getCreateTime(), DateTime.parse("1991-09-06T05:00-05:30")); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getEndTime(), DateTime.parse("1991-09-06T06:00-05:30")); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getElapsedTime(), new Duration(8, MINUTES)); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getExecutionTime(), new Duration(44, MINUTES)); + + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getTotalDrivers(), 16); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getQueuedDrivers(), 17); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getRunningDrivers(), 18); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getCompletedDrivers(), 19); + + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getCumulativeUserMemory(), 20.0); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getUserMemoryReservation(), DataSize.valueOf("21GB")); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getTotalMemoryReservation(), DataSize.valueOf("23GB")); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getPeakUserMemoryReservation(), DataSize.valueOf("24GB")); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getTotalCpuTime(), new Duration(24, MINUTES)); + + assertEquals(basicInfo.getQueryStatsNoFailedTasks().isFullyBlocked(), true); + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getBlockedReasons(), ImmutableSet.of(BlockedReason.WAITING_FOR_MEMORY)); + + assertEquals(basicInfo.getQueryStatsNoFailedTasks().getProgressPercentage(), OptionalDouble.of(100)); + assertEquals(basicInfo.getErrorCode(), StandardErrorCode.ABANDONED_QUERY.toErrorCode()); assertEquals(basicInfo.getErrorType(), StandardErrorCode.ABANDONED_QUERY.toErrorCode().getType()); } diff --git a/core/trino-main/src/test/java/io/trino/server/TestQueryStateInfo.java b/core/trino-main/src/test/java/io/trino/server/TestQueryStateInfo.java index 9d83b5791875..e5e692de29b7 100644 --- a/core/trino-main/src/test/java/io/trino/server/TestQueryStateInfo.java +++ b/core/trino-main/src/test/java/io/trino/server/TestQueryStateInfo.java @@ -96,6 +96,58 @@ public void testQueryStateInfo() private QueryInfo createQueryInfo(String queryId, QueryState state, String query) { + QueryStats queryStats = new QueryStats( + DateTime.parse("1991-09-06T05:00-05:30"), + DateTime.parse("1991-09-06T05:01-05:30"), + DateTime.parse("1991-09-06T05:02-05:30"), + DateTime.parse("1991-09-06T06:00-05:30"), + new Duration(10, SECONDS), + new Duration(8, MINUTES), + new Duration(7, MINUTES), + new Duration(34, MINUTES), + new Duration(9, MINUTES), + new Duration(10, MINUTES), + new Duration(11, MINUTES), + new Duration(12, MINUTES), + 13, + 14, + 15, + 100, + 17, + 18, + 34, + 19, + 20.0, + DataSize.valueOf("21GB"), + DataSize.valueOf("22GB"), + DataSize.valueOf("23GB"), + DataSize.valueOf("24GB"), + DataSize.valueOf("25GB"), + DataSize.valueOf("26GB"), + DataSize.valueOf("27GB"), + DataSize.valueOf("28GB"), + DataSize.valueOf("29GB"), + true, + new Duration(23, MINUTES), + new Duration(24, MINUTES), + new Duration(26, MINUTES), + true, + ImmutableSet.of(WAITING_FOR_MEMORY), + DataSize.valueOf("271GB"), + 281, + new Duration(26, MINUTES), + DataSize.valueOf("272GB"), + 282, + DataSize.valueOf("27GB"), + 28, + DataSize.valueOf("29GB"), + 30, + DataSize.valueOf("31GB"), + 32, + DataSize.valueOf("33GB"), + ImmutableList.of(), + DynamicFiltersStats.EMPTY, + ImmutableList.of()); return new QueryInfo( new QueryId(queryId), TEST_SESSION.toSessionRepresentation(), @@ -105,58 +157,8 @@ private QueryInfo createQueryInfo(String queryId, QueryState state, String query ImmutableList.of("2", "3"), query, Optional.empty(), - new QueryStats( - DateTime.parse("1991-09-06T05:00-05:30"), - DateTime.parse("1991-09-06T05:01-05:30"), - DateTime.parse("1991-09-06T05:02-05:30"), - DateTime.parse("1991-09-06T06:00-05:30"), - new Duration(10, SECONDS), - new Duration(8, MINUTES), - new Duration(7, MINUTES), - new Duration(34, MINUTES), - new Duration(9, MINUTES), - new Duration(10, MINUTES), - new Duration(11, MINUTES), - new Duration(12, MINUTES), - 13, - 14, - 15, - 100, - 17, - 18, - 34, - 19, - 20.0, - DataSize.valueOf("21GB"), - DataSize.valueOf("22GB"), - DataSize.valueOf("23GB"), - DataSize.valueOf("24GB"), - DataSize.valueOf("25GB"), - DataSize.valueOf("26GB"), - DataSize.valueOf("27GB"), - DataSize.valueOf("28GB"), - DataSize.valueOf("29GB"), - true, - new Duration(23, MINUTES), - new Duration(24, MINUTES), - new Duration(26, MINUTES), - true, - ImmutableSet.of(WAITING_FOR_MEMORY), - DataSize.valueOf("271GB"), - 281, - new Duration(26, MINUTES), - DataSize.valueOf("272GB"), - 282, - DataSize.valueOf("27GB"), - 28, - DataSize.valueOf("29GB"), - 30, - DataSize.valueOf("31GB"), - 32, - DataSize.valueOf("33GB"), - ImmutableList.of(), - DynamicFiltersStats.EMPTY, - ImmutableList.of()), + queryStats, + queryStats, Optional.empty(), Optional.empty(), Optional.empty(), diff --git a/testing/trino-tests/src/test/java/io/trino/memory/TestClusterMemoryLeakDetector.java b/testing/trino-tests/src/test/java/io/trino/memory/TestClusterMemoryLeakDetector.java index 4602b89cdae7..e33020a550be 100644 --- a/testing/trino-tests/src/test/java/io/trino/memory/TestClusterMemoryLeakDetector.java +++ b/testing/trino-tests/src/test/java/io/trino/memory/TestClusterMemoryLeakDetector.java @@ -67,6 +67,29 @@ public void testLeakDetector() private static BasicQueryInfo createQueryInfo(String queryId, QueryState state) { + BasicQueryStats queryStats = new BasicQueryStats( + DateTime.parse("1991-09-06T05:00-05:30"), + DateTime.parse("1991-09-06T05:01-05:30"), + new Duration(8, MINUTES), + new Duration(7, MINUTES), + new Duration(34, MINUTES), + 13, + 14, + 15, + 100, + DataSize.valueOf("21GB"), + 22, + DataSize.valueOf("20GB"), + 23, + DataSize.valueOf("23GB"), + DataSize.valueOf("24GB"), + DataSize.valueOf("25GB"), + DataSize.valueOf("26GB"), + new Duration(23, MINUTES), + new Duration(24, MINUTES), + true, + ImmutableSet.of(WAITING_FOR_MEMORY), + OptionalDouble.of(20)); return new BasicQueryInfo( new QueryId(queryId), TEST_SESSION.toSessionRepresentation(), @@ -77,29 +100,8 @@ private static BasicQueryInfo createQueryInfo(String queryId, QueryState state) "", Optional.empty(), Optional.empty(), - new BasicQueryStats( - DateTime.parse("1991-09-06T05:00-05:30"), - DateTime.parse("1991-09-06T05:01-05:30"), - new Duration(8, MINUTES), - new Duration(7, MINUTES), - new Duration(34, MINUTES), - 13, - 14, - 15, - 100, - DataSize.valueOf("21GB"), - 22, - DataSize.valueOf("20GB"), - 23, - DataSize.valueOf("23GB"), - DataSize.valueOf("24GB"), - DataSize.valueOf("25GB"), - DataSize.valueOf("26GB"), - new Duration(23, MINUTES), - new Duration(24, MINUTES), - true, - ImmutableSet.of(WAITING_FOR_MEMORY), - OptionalDouble.of(20)), + queryStats, + queryStats, null, null, Optional.empty());