Add processed input bytes and rows to event listener QueryStatistics

core/trino-main/src/main/java/io/trino/event/QueryMonitor.java

@@ -195,6 +195,8 @@ public void queryImmediateFailureEvent(BasicQueryInfo queryInfo, ExecutionFailur
                         0,
                         0,
                         0,
+                        0,
+                        0,
                         ImmutableList.of(),
                         0,
                         true,
@@ -273,6 +275,8 @@ private QueryStatistics createQueryStatistics(QueryInfo queryInfo)
                 queryStats.getPeakTaskTotalMemory().toBytes(),
                 queryStats.getPhysicalInputDataSize().toBytes(),
                 queryStats.getPhysicalInputPositions(),
+                queryStats.getProcessedInputDataSize().toBytes(),
+                queryStats.getProcessedInputPositions(),
                 queryStats.getInternalNetworkInputDataSize().toBytes(),
                 queryStats.getInternalNetworkInputPositions(),
                 queryStats.getRawInputDataSize().toBytes(),

core/trino-spi/src/main/java/io/trino/spi/eventlistener/QueryStatistics.java

@@ -43,6 +43,8 @@ public class QueryStatistics
     private final long peakTaskTotalMemory;
     private final long physicalInputBytes;
     private final long physicalInputRows;
+    private final long processedInputBytes;
+    private final long processedInputRows;
     private final long internalNetworkBytes;
     private final long internalNetworkRows;
     private final long totalBytes;
@@ -90,6 +92,8 @@ public QueryStatistics(
             long peakTaskTotalMemory,
             long physicalInputBytes,
             long physicalInputRows,
+            long processedInputBytes,
+            long processedInputRows,
             long internalNetworkBytes,
             long internalNetworkRows,
             long totalBytes,
@@ -122,6 +126,8 @@ public QueryStatistics(
         this.peakTaskTotalMemory = peakTaskTotalMemory;
         this.physicalInputBytes = physicalInputBytes;
         this.physicalInputRows = physicalInputRows;
+        this.processedInputBytes = processedInputBytes;
+        this.processedInputRows = processedInputRows;
         this.internalNetworkBytes = internalNetworkBytes;
         this.internalNetworkRows = internalNetworkRows;
         this.totalBytes = totalBytes;
@@ -230,6 +236,18 @@ public long getPhysicalInputRows()
         return physicalInputRows;
     }
 
+    @JsonProperty
+    public long getProcessedInputBytes()
+    {
+        return processedInputBytes;
+    }
+
+    @JsonProperty
+    public long getProcessedInputRows()
+    {
+        return processedInputRows;
+    }
+
     @JsonProperty
     public long getInternalNetworkBytes()
     {

plugin/trino-http-event-listener/src/test/java/io/trino/plugin/httpquery/TestHttpEventListener.java

@@ -164,6 +164,8 @@ public class TestHttpEventListener
                 0L,
                 0L,
                 0L,
+                0L,
+                0L,
                 0.0f,
                 Collections.emptyList(),
                 0,

testing/trino-tests/src/test/java/io/trino/execution/TestEventListenerWithSplits.java

@@ -164,6 +164,8 @@ public void testSplitsForNormalQuery()
         // Deterministic statistics
         assertEquals(statistics.getPhysicalInputBytes(), 0);
         assertEquals(statistics.getPhysicalInputRows(), expectedCompletedPositions);
+        assertEquals(statistics.getProcessedInputBytes(), 0);
+        assertEquals(statistics.getProcessedInputRows(), expectedCompletedPositions);
         assertEquals(statistics.getInternalNetworkBytes(), 381);
         assertEquals(statistics.getInternalNetworkRows(), 3);
         assertEquals(statistics.getTotalBytes(), 0);