add new field updatedRows to QueryStatistics.java where it's availabl…

…e in EventListener callbacks. It comes from TableMutationOperator.java where update/delete queries are issued.
trinodb · Jan 21, 2025 · b45d6dc · b45d6dc
1 parent af08b56
commit b45d6dc
Show file tree

Hide file tree

Showing 36 changed files with 154 additions and 0 deletions.
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
@@ -332,6 +332,7 @@ private static QueryStats immediateFailureQueryStats()
                 DataSize.ofBytes(0),
                 0,
                 0,
+                0,
                 new Duration(0, MILLISECONDS),
                 new Duration(0, MILLISECONDS),
                 DataSize.ofBytes(0),

diff --git a/core/trino-main/src/main/java/io/trino/event/QueryMonitor.java b/core/trino-main/src/main/java/io/trino/event/QueryMonitor.java
@@ -231,6 +231,7 @@ public void queryImmediateFailureEvent(BasicQueryInfo queryInfo, ExecutionFailur
                         0,
                         0,
                         0,
+                        0,
                         ImmutableList.of(),
                         0,
                         true,
@@ -343,6 +344,7 @@ private QueryStatistics createQueryStatistics(QueryInfo queryInfo)
                 queryStats.getRawInputPositions(),
                 queryStats.getOutputDataSize().toBytes(),
                 queryStats.getOutputPositions(),
+                queryStats.getUpdatedPositions(),
                 queryStats.getLogicalWrittenDataSize().toBytes(),
                 queryStats.getWrittenPositions(),
                 queryStats.getSpilledDataSize().toBytes(),

diff --git a/core/trino-main/src/main/java/io/trino/execution/DistributionSnapshot.java b/core/trino-main/src/main/java/io/trino/execution/DistributionSnapshot.java
@@ -71,6 +71,7 @@ public static OperatorStats pruneOperatorStats(OperatorStats operatorStats)
                 operatorStats.getGetOutputCpu(),
                 operatorStats.getOutputDataSize(),
                 operatorStats.getOutputPositions(),
+                operatorStats.getUpdatedPositions(),
                 operatorStats.getDynamicFilterSplitsProcessed(),
                 pruneMetrics(operatorStats.getMetrics()),
                 pruneMetrics(operatorStats.getConnectorMetrics()),

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
@@ -686,6 +686,7 @@ private QueryStats getQueryStats(Optional<StageInfo> rootStage, List<StageInfo>
         long outputDataSize = 0;
         long failedOutputDataSize = 0;
         long outputPositions = 0;
+        long updatedPositions = 0;
         long failedOutputPositions = 0;
 
         long outputBlockedTime = 0;
@@ -772,6 +773,7 @@ private QueryStats getQueryStats(Optional<StageInfo> rootStage, List<StageInfo>
             outputDataSize += outputStageStats.getOutputDataSize().toBytes();
             failedOutputDataSize += outputStageStats.getFailedOutputDataSize().toBytes();
             outputPositions += outputStageStats.getOutputPositions();
+            updatedPositions += outputStageStats.getUpdatedPositions();
             failedOutputPositions += outputStageStats.getFailedOutputPositions();
         }
 
@@ -898,6 +900,7 @@ private QueryStats getQueryStats(Optional<StageInfo> rootStage, List<StageInfo>
                 succinctBytes(outputDataSize),
                 succinctBytes(failedOutputDataSize),
                 outputPositions,
+                updatedPositions,
                 failedOutputPositions,
 
                 new Duration(outputBlockedTime, NANOSECONDS).convertToMostSuccinctTimeUnit(),
@@ -1502,6 +1505,7 @@ private static QueryStats pruneQueryStats(QueryStats queryStats)
                 queryStats.getOutputDataSize(),
                 queryStats.getFailedOutputDataSize(),
                 queryStats.getOutputPositions(),
+                queryStats.getUpdatedPositions(),
                 queryStats.getFailedOutputPositions(),
                 queryStats.getOutputBlockedTime(),
                 queryStats.getFailedOutputBlockedTime(),

diff --git a/core/trino-main/src/main/java/io/trino/execution/QueryStats.java b/core/trino-main/src/main/java/io/trino/execution/QueryStats.java
@@ -118,6 +118,7 @@ public class QueryStats
     private final DataSize outputDataSize;
     private final DataSize failedOutputDataSize;
     private final long outputPositions;
+    private final long updatedPositions;
     private final long failedOutputPositions;
 
     private final Duration outputBlockedTime;
@@ -213,6 +214,7 @@ public QueryStats(
             @JsonProperty("outputDataSize") DataSize outputDataSize,
             @JsonProperty("failedOutputDataSize") DataSize failedOutputDataSize,
             @JsonProperty("outputPositions") long outputPositions,
+            @JsonProperty("updatedPositions") long updatedPositions,
             @JsonProperty("failedOutputPositions") long failedOutputPositions,
 
             @JsonProperty("outputBlockedTime") Duration outputBlockedTime,
@@ -323,6 +325,8 @@ public QueryStats(
         this.failedOutputDataSize = requireNonNull(failedOutputDataSize, "failedOutputDataSize is null");
         checkArgument(outputPositions >= 0, "outputPositions is negative");
         this.outputPositions = outputPositions;
+        checkArgument(updatedPositions >= 0, "updatedPositions is negative");
+        this.updatedPositions = updatedPositions;
         checkArgument(failedOutputPositions >= 0, "failedOutputPositions is negative");
         this.failedOutputPositions = failedOutputPositions;
 
@@ -743,6 +747,12 @@ public long getOutputPositions()
         return outputPositions;
     }
 
+    @JsonProperty
+    public long getUpdatedPositions()
+    {
+        return updatedPositions;
+    }
+
     @JsonProperty
     public long getFailedOutputPositions()
     {

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
@@ -473,6 +473,7 @@ public StageInfo getStageInfo(Supplier<Iterable<TaskInfo>> taskInfosSupplier)
         long outputDataSize = 0;
         long failedOutputDataSize = 0;
         long outputPositions = 0;
+        long updatedPositions = 0;
         long failedOutputPositions = 0;
         Metrics.Accumulator outputBufferMetrics = Metrics.accumulator();
 
@@ -553,6 +554,7 @@ public StageInfo getStageInfo(Supplier<Iterable<TaskInfo>> taskInfosSupplier)
             taskInfo.outputBuffers().getUtilization().ifPresent(bufferUtilizationHistograms::add);
             outputDataSize += taskStats.getOutputDataSize().toBytes();
             outputPositions += taskStats.getOutputPositions();
+            updatedPositions += taskStats.getUpdatedPositions();
             bufferMetrics.ifPresent(outputBufferMetrics::add);
 
             outputBlockedTime += taskStats.getOutputBlockedTime().roundTo(NANOSECONDS);
@@ -660,6 +662,7 @@ public StageInfo getStageInfo(Supplier<Iterable<TaskInfo>> taskInfosSupplier)
                 succinctBytes(outputDataSize),
                 succinctBytes(failedOutputDataSize),
                 outputPositions,
+                updatedPositions,
                 failedOutputPositions,
                 outputBufferMetrics.get(),
                 succinctDuration(outputBlockedTime, NANOSECONDS),

diff --git a/core/trino-main/src/main/java/io/trino/execution/StageStats.java b/core/trino-main/src/main/java/io/trino/execution/StageStats.java
@@ -106,6 +106,7 @@ public class StageStats
     private final DataSize outputDataSize;
     private final DataSize failedOutputDataSize;
     private final long outputPositions;
+    private final long updatedPositions;
     private final long failedOutputPositions;
     private final Metrics outputBufferMetrics;
 
@@ -182,6 +183,7 @@ public StageStats(
             @JsonProperty("outputDataSize") DataSize outputDataSize,
             @JsonProperty("failedOutputDataSize") DataSize failedOutputDataSize,
             @JsonProperty("outputPositions") long outputPositions,
+            @JsonProperty("updatedPositions") long updatedPositions,
             @JsonProperty("failedOutputPositions") long failedOutputPositions,
             @JsonProperty("outputBufferMetrics") Metrics outputBufferMetrics,
 
@@ -273,6 +275,8 @@ public StageStats(
         this.failedOutputDataSize = requireNonNull(failedOutputDataSize, "failedOutputDataSize is null");
         checkArgument(outputPositions >= 0, "outputPositions is negative");
         this.outputPositions = outputPositions;
+        checkArgument(updatedPositions >= 0, "updatedPositions is negative");
+        this.updatedPositions = updatedPositions;
         checkArgument(failedOutputPositions >= 0, "failedOutputPositions is negative");
         this.failedOutputPositions = failedOutputPositions;
         this.outputBufferMetrics = requireNonNull(outputBufferMetrics, "outputBufferMetrics is null");
@@ -588,6 +592,12 @@ public long getOutputPositions()
         return outputPositions;
     }
 
+    @JsonProperty
+    public long getUpdatedPositions()
+    {
+        return updatedPositions;
+    }
+
     @JsonProperty
     public long getFailedOutputPositions()
     {
@@ -736,6 +746,7 @@ public static StageStats createInitial()
                 zeroBytes,
                 0,
                 0,
+                0,
                 Metrics.EMPTY,
                 zeroSeconds,
                 zeroSeconds,

diff --git a/core/trino-main/src/main/java/io/trino/operator/DriverContext.java b/core/trino-main/src/main/java/io/trino/operator/DriverContext.java
@@ -346,6 +346,7 @@ public DriverStats getDriverStats()
         Duration inputBlockedTime;
         DataSize outputDataSize;
         long outputPositions;
+        long updatedPositions;
         Duration outputBlockedTime;
         if (inputOperator != null) {
             physicalInputDataSize = inputOperator.getPhysicalInputDataSize();
@@ -364,6 +365,8 @@ public DriverStats getDriverStats()
 
             inputBlockedTime = inputOperator.getBlockedWall();
 
+            updatedPositions = inputOperator.getUpdatedPositions();
+
             OperatorStats outputOperator = requireNonNull(getLast(operators, null));
             outputDataSize = outputOperator.getOutputDataSize();
             outputPositions = outputOperator.getOutputPositions();
@@ -389,6 +392,7 @@ public DriverStats getDriverStats()
 
             outputDataSize = DataSize.ofBytes(0);
             outputPositions = 0;
+            updatedPositions = 0;
 
             outputBlockedTime = new Duration(0, MILLISECONDS);
         }
@@ -428,6 +432,7 @@ public DriverStats getDriverStats()
                 inputBlockedTime,
                 outputDataSize.succinct(),
                 outputPositions,
+                updatedPositions,
                 outputBlockedTime,
                 succinctBytes(physicalWrittenDataSize),
                 operators);

diff --git a/core/trino-main/src/main/java/io/trino/operator/DriverStats.java b/core/trino-main/src/main/java/io/trino/operator/DriverStats.java
@@ -67,6 +67,7 @@ public class DriverStats
 
     private final DataSize outputDataSize;
     private final long outputPositions;
+    private final long updatedPositions;
 
     private final Duration outputBlockedTime;
 
@@ -109,6 +110,7 @@ public DriverStats()
 
         this.outputDataSize = DataSize.ofBytes(0);
         this.outputPositions = 0;
+        this.updatedPositions = 0;
 
         this.outputBlockedTime = new Duration(0, MILLISECONDS);
 
@@ -152,6 +154,7 @@ public DriverStats(
 
             @JsonProperty("outputDataSize") DataSize outputDataSize,
             @JsonProperty("outputPositions") long outputPositions,
+            @JsonProperty("updatedPositions") long updatedPositions,
 
             @JsonProperty("outputBlockedTime") Duration outputBlockedTime,
 
@@ -197,6 +200,8 @@ public DriverStats(
         this.outputDataSize = requireNonNull(outputDataSize, "outputDataSize is null");
         checkArgument(outputPositions >= 0, "outputPositions is negative");
         this.outputPositions = outputPositions;
+        checkArgument(updatedPositions >= 0, "updatedPositions is negative");
+        this.updatedPositions = updatedPositions;
 
         this.outputBlockedTime = requireNonNull(outputBlockedTime, "outputBlockedTime is null");
 
@@ -357,6 +362,12 @@ public long getOutputPositions()
         return outputPositions;
     }
 
+    @JsonProperty
+    public long getUpdatedPositions()
+    {
+        return updatedPositions;
+    }
+
     @JsonProperty
     public Duration getOutputBlockedTime()
     {

diff --git a/core/trino-main/src/main/java/io/trino/operator/OperatorContext.java b/core/trino-main/src/main/java/io/trino/operator/OperatorContext.java
@@ -82,6 +82,7 @@ public class OperatorContext
     private final OperationTiming getOutputTiming = new OperationTiming();
     private final CounterStat outputDataSize = new CounterStat();
     private final CounterStat outputPositions = new CounterStat();
+    private final CounterStat updatedPositions = new CounterStat();
 
     private final AtomicLong dynamicFilterSplitsProcessed = new AtomicLong();
     private final AtomicReference<Metrics> metrics = new AtomicReference<>(Metrics.EMPTY);  // this is not incremental, but gets overwritten by the latest value.
@@ -490,6 +491,11 @@ public CounterStat getOutputPositions()
         return outputPositions;
     }
 
+    public CounterStat getUpdatedPositions()
+    {
+        return updatedPositions;
+    }
+
     public long getWriterInputDataSize()
     {
         return writerInputDataSize.get();
@@ -554,6 +560,7 @@ public OperatorStats getOperatorStats()
                 new Duration(getOutputTiming.getCpuNanos(), NANOSECONDS).convertToMostSuccinctTimeUnit(),
                 DataSize.ofBytes(outputDataSize.getTotalCount()),
                 outputPositions.getTotalCount(),
+                updatedPositions.getTotalCount(),
 
                 dynamicFilterSplitsProcessed.get(),
                 getOperatorMetrics(

diff --git a/core/trino-main/src/main/java/io/trino/operator/OperatorStats.java b/core/trino-main/src/main/java/io/trino/operator/OperatorStats.java
@@ -62,6 +62,7 @@ public class OperatorStats
     private final Duration getOutputCpu;
     private final DataSize outputDataSize;
     private final long outputPositions;
+    private final long updatedPositions;
 
     private final long dynamicFilterSplitsProcessed;
     private final Metrics metrics;
@@ -117,6 +118,7 @@ public OperatorStats(
             @JsonProperty("getOutputCpu") Duration getOutputCpu,
             @JsonProperty("outputDataSize") DataSize outputDataSize,
             @JsonProperty("outputPositions") long outputPositions,
+            @JsonProperty("updatedPositions") long updatedPositions,
 
             @JsonProperty("dynamicFilterSplitsProcessed") long dynamicFilterSplitsProcessed,
             @JsonProperty("metrics") Metrics metrics,
@@ -174,6 +176,8 @@ public OperatorStats(
         this.outputDataSize = requireNonNull(outputDataSize, "outputDataSize is null");
         checkArgument(outputPositions >= 0, "outputPositions is negative");
         this.outputPositions = outputPositions;
+        checkArgument(updatedPositions >= 0, "updatedPositions is negative");
+        this.updatedPositions = updatedPositions;
 
         this.dynamicFilterSplitsProcessed = dynamicFilterSplitsProcessed;
         this.metrics = requireNonNull(metrics, "metrics is null");
@@ -340,6 +344,12 @@ public long getOutputPositions()
         return outputPositions;
     }
 
+    @JsonProperty
+    public long getUpdatedPositions()
+    {
+        return updatedPositions;
+    }
+
     @JsonProperty
     public long getDynamicFilterSplitsProcessed()
     {
@@ -485,6 +495,7 @@ private OperatorStats add(Iterable<OperatorStats> operators, Optional<Metrics> p
         long getOutputCpu = this.getOutputCpu.roundTo(NANOSECONDS);
         long outputDataSize = this.outputDataSize.toBytes();
         long outputPositions = this.outputPositions;
+        long updatedPositions = this.updatedPositions;
 
         long dynamicFilterSplitsProcessed = this.dynamicFilterSplitsProcessed;
         Metrics.Accumulator metricsAccumulator = Metrics.accumulator().add(this.getMetrics());
@@ -535,6 +546,7 @@ private OperatorStats add(Iterable<OperatorStats> operators, Optional<Metrics> p
             getOutputCpu += operator.getGetOutputCpu().roundTo(NANOSECONDS);
             outputDataSize += operator.getOutputDataSize().toBytes();
             outputPositions += operator.getOutputPositions();
+            updatedPositions += operator.getUpdatedPositions();
 
             dynamicFilterSplitsProcessed += operator.getDynamicFilterSplitsProcessed();
             metricsAccumulator.add(operator.getMetrics());
@@ -596,6 +608,7 @@ private OperatorStats add(Iterable<OperatorStats> operators, Optional<Metrics> p
                 new Duration(getOutputCpu, NANOSECONDS).convertToMostSuccinctTimeUnit(),
                 DataSize.ofBytes(outputDataSize),
                 outputPositions,
+                updatedPositions,
 
                 dynamicFilterSplitsProcessed,
                 metricsAccumulator.get(),
@@ -672,6 +685,7 @@ public OperatorStats summarize()
                 getOutputCpu,
                 outputDataSize,
                 outputPositions,
+                updatedPositions,
                 dynamicFilterSplitsProcessed,
                 metrics,
                 connectorMetrics,
@@ -717,6 +731,7 @@ public OperatorStats withPipelineMetrics(Metrics pipelineMetrics)
                 getOutputCpu,
                 outputDataSize,
                 outputPositions,
+                updatedPositions,
                 dynamicFilterSplitsProcessed,
                 metrics,
                 connectorMetrics,

diff --git a/core/trino-main/src/main/java/io/trino/operator/PipelineContext.java b/core/trino-main/src/main/java/io/trino/operator/PipelineContext.java
@@ -98,6 +98,7 @@ public class PipelineContext
 
     private final CounterStat outputDataSize = new CounterStat();
     private final CounterStat outputPositions = new CounterStat();
+    private final CounterStat updatedPositions = new CounterStat();
 
     private final AtomicLong outputBlockedTime = new AtomicLong();
 
@@ -237,6 +238,7 @@ public void driverFinished(DriverContext driverContext)
 
         outputDataSize.update(driverStats.getOutputDataSize().toBytes());
         outputPositions.update(driverStats.getOutputPositions());
+        updatedPositions.update(driverStats.getUpdatedPositions());
 
         outputBlockedTime.getAndAdd(driverStats.getOutputBlockedTime().roundTo(NANOSECONDS));
 
@@ -403,6 +405,7 @@ public PipelineStats getPipelineStats()
 
         long outputDataSize = this.outputDataSize.getTotalCount();
         long outputPositions = this.outputPositions.getTotalCount();
+        long updatedPositions = this.updatedPositions.getTotalCount();
 
         long outputBlockedTime = this.outputBlockedTime.get();
 
@@ -455,6 +458,7 @@ public PipelineStats getPipelineStats()
 
             outputDataSize += driverStats.getOutputDataSize().toBytes();
             outputPositions += driverStats.getOutputPositions();
+            updatedPositions += driverStats.getUpdatedPositions();
 
             outputBlockedTime += driverStats.getOutputBlockedTime().roundTo(NANOSECONDS);
 
@@ -538,6 +542,7 @@ else if (pipelineLevelMetrics != Metrics.EMPTY) {
 
                 succinctBytes(outputDataSize),
                 outputPositions,
+                updatedPositions,
 
                 new Duration(outputBlockedTime, NANOSECONDS).convertToMostSuccinctTimeUnit(),