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

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. I am
open to suggestions.

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

@@ -253,6 +253,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_ADD_TIMEOUT_PUT_ENDPOINT = def(9_074_0_00);
     public static final TransportVersion ESQL_FIELD_ATTRIBUTE_DROP_TYPE = def(9_075_0_00);
     public static final TransportVersion ESQL_TIME_SERIES_SOURCE_STATUS = def(9_076_0_00);
+    public static final TransportVersion ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME = def(9_077_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -110,6 +110,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,
@@ -223,6 +228,7 @@ public void finish() {
         finished = true;
         Block[] blocks = null;
         IntVector selected = null;
+        long startInNanos = System.nanoTime();
         boolean success = false;
         try {
             selected = blockHash.nonEmpty();
@@ -247,6 +253,7 @@ public void finish() {
             if (success == false && blocks != null) {
                 Releasables.closeExpectNoException(blocks);
             }
+            emitNanos += System.nanoTime() - startInNanos;
         }
     }
 
@@ -269,7 +276,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) {
@@ -320,20 +327,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 {
@@ -348,6 +359,11 @@ protected Status(StreamInput in) throws IOException {
                 rowsReceived = 0;
                 rowsEmitted = 0;
             }
+            if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME)) {
+                emitNanos = in.readVLong();
+            } else {
+                emitNanos = 0;
+            }
         }
 
         @Override
@@ -360,6 +376,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)) {
+                out.writeVLong(emitNanos);
+            }
         }
 
         @Override
@@ -402,6 +421,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();
@@ -416,6 +442,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();
 
         }
@@ -429,12 +459,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);
     }
 }