Add emit time to hash aggregation status (elastic#127988) (elastic#128225)

The hash aggregation operator may take time to emit the output pages,
including keys and aggregated values. This change adds an emit_time
field to the status. While I considered including this in hash_nanos and
aggregation_nanos, having a separate section feels more natural.

Author: dnhatn

docs/changelog/127988.yaml

@@ -0,0 +1,5 @@
+pr: 127988
+summary: Add emit time to hash aggregation status
+area: ES|QL
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -222,6 +222,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED_8_19 = def(8_841_0_31);
     public static final TransportVersion FIELD_CAPS_ADD_CLUSTER_ALIAS = def(8_841_0_32);
     public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM_BACKPORT_8_19 = def(8_841_0_33);
+    public static final TransportVersion ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME_8_19 = def(8_841_0_34);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/HashAggregationOperator.java

@@ -108,6 +108,11 @@ public String describe() {
      */
     private long rowsEmitted;
 
+    /**
+     * Total nanos for emitting the output
+     */
+    protected long emitNanos;
+
     @SuppressWarnings("this-escape")
     public HashAggregationOperator(
         List<GroupingAggregator.Factory> aggregators,
@@ -217,6 +222,7 @@ public void finish() {
         finished = true;
         Block[] blocks = null;
         IntVector selected = null;
+        long startInNanos = System.nanoTime();
         boolean success = false;
         try {
             selected = blockHash.nonEmpty();
@@ -240,6 +246,7 @@ public void finish() {
             if (success == false && blocks != null) {
                 Releasables.closeExpectNoException(blocks);
             }
+            emitNanos += System.nanoTime() - startInNanos;
         }
     }
 
@@ -258,7 +265,7 @@ public void close() {
 
     @Override
     public Operator.Status status() {
-        return new Status(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted);
+        return new Status(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted, emitNanos);
     }
 
     protected static void checkState(boolean condition, String msg) {
@@ -309,20 +316,24 @@ public static class Status implements Operator.Status {
          */
         private final long rowsEmitted;
 
+        private final long emitNanos;
+
         /**
          * Build.
          * @param hashNanos Nanoseconds this operator has spent hashing grouping keys.
          * @param aggregationNanos Nanoseconds this operator has spent running the aggregations.
          * @param pagesProcessed Count of pages this operator has processed.
          * @param rowsReceived Count of rows this operator has received.
          * @param rowsEmitted Count of rows this operator has emitted.
+         * @param emitNanos Nanoseconds this operator has spent emitting the output.
          */
-        public Status(long hashNanos, long aggregationNanos, int pagesProcessed, long rowsReceived, long rowsEmitted) {
+        public Status(long hashNanos, long aggregationNanos, int pagesProcessed, long rowsReceived, long rowsEmitted, long emitNanos) {
             this.hashNanos = hashNanos;
             this.aggregationNanos = aggregationNanos;
             this.pagesProcessed = pagesProcessed;
             this.rowsReceived = rowsReceived;
             this.rowsEmitted = rowsEmitted;
+            this.emitNanos = emitNanos;
         }
 
         protected Status(StreamInput in) throws IOException {
@@ -337,6 +348,11 @@ protected Status(StreamInput in) throws IOException {
                 rowsReceived = 0;
                 rowsEmitted = 0;
             }
+            if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME_8_19)) {
+                emitNanos = in.readVLong();
+            } else {
+                emitNanos = 0;
+            }
         }
 
         @Override
@@ -349,6 +365,9 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeVLong(rowsReceived);
                 out.writeVLong(rowsEmitted);
             }
+            if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME_8_19)) {
+                out.writeVLong(emitNanos);
+            }
         }
 
         @Override
@@ -391,6 +410,13 @@ public long rowsEmitted() {
             return rowsEmitted;
         }
 
+        /**
+         * Nanoseconds this operator has spent emitting the output.
+         */
+        public long emitNanos() {
+            return emitNanos;
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
@@ -405,6 +431,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field("pages_processed", pagesProcessed);
             builder.field("rows_received", rowsReceived);
             builder.field("rows_emitted", rowsEmitted);
+            builder.field("emit_nanos", emitNanos);
+            if (builder.humanReadable()) {
+                builder.field("emit_time", TimeValue.timeValueNanos(emitNanos));
+            }
             return builder.endObject();
 
         }
@@ -418,12 +448,13 @@ public boolean equals(Object o) {
                 && aggregationNanos == status.aggregationNanos
                 && pagesProcessed == status.pagesProcessed
                 && rowsReceived == status.rowsReceived
-                && rowsEmitted == status.rowsEmitted;
+                && rowsEmitted == status.rowsEmitted
+                && emitNanos == status.emitNanos;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted);
+            return Objects.hash(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted, emitNanos);
         }
 
         @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/HashAggregationOperatorStatusTests.java

@@ -16,7 +16,7 @@
 
 public class HashAggregationOperatorStatusTests extends AbstractWireSerializingTestCase<HashAggregationOperator.Status> {
     public static HashAggregationOperator.Status simple() {
-        return new HashAggregationOperator.Status(500012, 200012, 123, 111, 222);
+        return new HashAggregationOperator.Status(500012, 200012, 123, 111, 222, 180017);
     }
 
     public static String simpleToJson() {
@@ -28,7 +28,9 @@ public static String simpleToJson() {
               "aggregation_time" : "200micros",
               "pages_processed" : 123,
               "rows_received" : 111,
-              "rows_emitted" : 222
+              "rows_emitted" : 222,
+              "emit_nanos" : 180017,
+              "emit_time" : "180micros"
             }""";
     }
 
@@ -48,6 +50,7 @@ public HashAggregationOperator.Status createTestInstance() {
             randomNonNegativeLong(),
             randomNonNegativeInt(),
             randomNonNegativeLong(),
+            randomNonNegativeLong(),
             randomNonNegativeLong()
         );
     }
@@ -59,14 +62,16 @@ protected HashAggregationOperator.Status mutateInstance(HashAggregationOperator.
         int pagesProcessed = instance.pagesProcessed();
         long rowsReceived = instance.rowsReceived();
         long rowsEmitted = instance.rowsEmitted();
-        switch (between(0, 4)) {
+        long emitNanos = instance.emitNanos();
+        switch (between(0, 5)) {
             case 0 -> hashNanos = randomValueOtherThan(hashNanos, ESTestCase::randomNonNegativeLong);
             case 1 -> aggregationNanos = randomValueOtherThan(aggregationNanos, ESTestCase::randomNonNegativeLong);
             case 2 -> pagesProcessed = randomValueOtherThan(pagesProcessed, ESTestCase::randomNonNegativeInt);
             case 3 -> rowsReceived = randomValueOtherThan(rowsReceived, ESTestCase::randomNonNegativeLong);
             case 4 -> rowsEmitted = randomValueOtherThan(rowsEmitted, ESTestCase::randomNonNegativeLong);
+            case 5 -> emitNanos = randomValueOtherThan(emitNanos, ESTestCase::randomNonNegativeLong);
             default -> throw new UnsupportedOperationException();
         }
-        return new HashAggregationOperator.Status(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted);
+        return new HashAggregationOperator.Status(hashNanos, aggregationNanos, pagesProcessed, rowsReceived, rowsEmitted, emitNanos);
     }
 }