Merge branch 'main' of https://github.com/Jan-Kazlouski-elastic/elasticsearch into feature/hugging-face-chat-completion-integration

# Conflicts:
#	server/src/main/java/org/elasticsearch/TransportVersions.java

Author: Jan-Kazlouski-elastic

docs/changelog/127988.yaml

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

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -37,6 +37,7 @@
 import org.apache.http.conn.DnsResolver;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.cluster.coordination.stateless.StoreHeartbeatService;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
@@ -176,6 +177,13 @@ public AmazonS3Reference client(RepositoryMetadata repositoryMetadata) {
             if (existing != null && existing.tryIncRef()) {
                 return existing;
             }
+
+            if (lifecycle.started() == false) {
+                // doClose() calls releaseCachedClients() which is also synchronized (this) so if we're STARTED here then the client we
+                // create will definitely not leak on close.
+                throw new AlreadyClosedException("S3Service is in state [" + lifecycle + "]");
+            }
+
             final SdkHttpClient httpClient = buildHttpClient(clientSettings, getCustomDnsResolver());
             Releasable toRelease = httpClient::close;
             try {

muted-tests.yml

@@ -474,6 +474,9 @@ tests:
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test041AmazonCaCertsAreInTheKeystore
   issue: https://github.com/elastic/elasticsearch/issues/128007
+- class: org.elasticsearch.compute.aggregation.SampleDoubleAggregatorFunctionTests
+  method: testSimpleWithCranky
+  issue: https://github.com/elastic/elasticsearch/issues/128024
 
 # Examples:
 #

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

@@ -174,7 +174,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion RESCORE_VECTOR_ALLOW_ZERO_BACKPORT_8_19 = def(8_841_0_27);
     public static final TransportVersion INFERENCE_ADD_TIMEOUT_PUT_ENDPOINT_8_19 = def(8_841_0_28);
     public static final TransportVersion ESQL_REPORT_SHARD_PARTITIONING_8_19 = def(8_841_0_29);
-    public static final TransportVersion ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED_8_19 = def(8_842_0_29);
+    public static final TransportVersion ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED_8_19 = def(8_841_0_30);
     public static final TransportVersion V_9_0_0 = def(9_000_0_09);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_1 = def(9_000_0_10);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_2 = def(9_000_0_11);
@@ -254,7 +254,8 @@ 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 ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED = def(9_077_0_00);
+    public static final TransportVersion ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME = def(9_077_0_00);
+    public static final TransportVersion ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED = def(9_078_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);
     }
 }