Merge branch 'main' into sagemaker-openai-completion

Author: prwhelan

.buildkite/pipelines/periodic-micro-benchmarks.yml

@@ -0,0 +1,10 @@
+steps:
+  - label: periodic-micro-benchmarks
+    command: |
+      .ci/scripts/run-gradle.sh -p benchmarks/ run --args 'org.elasticsearch.benchmark._nightly -rf json -rff build/result.json'
+    timeout_in_minutes: 300
+    agents:
+      provider: gcp
+      image: family/elasticsearch-ubuntu-2004
+      machineType: custom-32-98304
+      buildDirectory: /dev/shm/bk

benchmarks/src/main/java/org/elasticsearch/benchmark/esql/QueryPlanningBenchmark.java renamed to benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/QueryPlanningBenchmark.java

@@ -7,7 +7,7 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-package org.elasticsearch.benchmark.esql;
+package org.elasticsearch.benchmark._nightly.esql;
 
 import org.elasticsearch.common.logging.LogConfigurator;
 import org.elasticsearch.common.settings.Settings;
@@ -67,9 +67,9 @@ public class QueryPlanningBenchmark {
     }
 
     private PlanTelemetry telemetry;
-    private EsqlParser parser;
-    private Analyzer analyzer;
-    private LogicalPlanOptimizer optimizer;
+    private EsqlParser defaultParser;
+    private Analyzer manyFieldsAnalyzer;
+    private LogicalPlanOptimizer defaultOptimizer;
 
     @Setup
     public void setup() {
@@ -100,8 +100,8 @@ public void setup() {
         var functionRegistry = new EsqlFunctionRegistry();
 
         telemetry = new PlanTelemetry(functionRegistry);
-        parser = new EsqlParser();
-        analyzer = new Analyzer(
+        defaultParser = new EsqlParser();
+        manyFieldsAnalyzer = new Analyzer(
             new AnalyzerContext(
                 config,
                 functionRegistry,
@@ -112,10 +112,10 @@ public void setup() {
             ),
             new Verifier(new Metrics(functionRegistry), new XPackLicenseState(() -> 0L))
         );
-        optimizer = new LogicalPlanOptimizer(new LogicalOptimizerContext(config, FoldContext.small()));
+        defaultOptimizer = new LogicalPlanOptimizer(new LogicalOptimizerContext(config, FoldContext.small()));
     }
 
-    private LogicalPlan plan(String query) {
+    private LogicalPlan plan(EsqlParser parser, Analyzer analyzer, LogicalPlanOptimizer optimizer, String query) {
         var parsed = parser.createStatement(query, new QueryParams(), telemetry);
         var analyzed = analyzer.analyze(parsed);
         var optimized = optimizer.optimize(analyzed);
@@ -124,6 +124,6 @@ private LogicalPlan plan(String query) {
 
     @Benchmark
     public void manyFields(Blackhole blackhole) {
-        blackhole.consume(plan("FROM test | LIMIT 10"));
+        blackhole.consume(plan(defaultParser, manyFieldsAnalyzer, defaultOptimizer, "FROM test | LIMIT 10"));
     }
 }

catalog-info.yaml

@@ -263,3 +263,41 @@ spec:
         Daily:
           branch: main
           cronline: "0 12 * * * America/New_York"
+---
+# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/e57ee3bed7a6f73077a3f55a38e76e40ec87a7cf/rre.schema.json
+apiVersion: backstage.io/v1alpha1
+kind: Resource
+metadata:
+  name: buildkite-pipeline-elasticsearch-periodic-micro-benchmarks
+  description: Runs periodic micro benchmarks fom the main branch
+  links:
+    - title: Pipeline
+      url: https://buildkite.com/elastic/elasticsearch-periodic-micro-benchmarks
+spec:
+  type: buildkite-pipeline
+  system: buildkite
+  owner: group:elasticsearch-team
+  implementation:
+    apiVersion: buildkite.elastic.dev/v1
+    kind: Pipeline
+    metadata:
+      description: ":elasticsearch: Runs nightly micro benchmarks fom the main branch"
+      name: elasticsearch / periodic / micro-benchmarks
+    spec:
+      repository: elastic/elasticsearch
+      pipeline_file: .buildkite/pipelines/periodic-micro-benchmarks.yml
+      branch_configuration: main
+      teams:
+        elasticsearch-team: {}
+        ml-core: {}
+        everyone:
+          access_level: BUILD_AND_READ
+      provider_settings:
+        build_branches: false
+        build_pull_requests: false
+        publish_commit_status: false
+        trigger_mode: none
+      schedules:
+        Daily:
+          branch: main
+          cronline: "@daily"

distribution/docker/src/docker/dockerfiles/cloud_ess_fips/Dockerfile

@@ -24,7 +24,7 @@
 # Extract Elasticsearch artifact
 ################################################################################
 
-FROM docker.elastic.co/wolfi/chainguard-base-fips:latest@sha256:a9e07d97c55f5b854b1c203509e57b6f466eb26318540b2e1b59734355f747b6 AS builder
+FROM docker.elastic.co/wolfi/chainguard-base-fips:latest@sha256:68e0781cd592beda39880428985d5fecca1cf2abb18365da73bf1f7ebd994974 AS builder
 
 # Install required packages to extract the Elasticsearch distribution
 RUN <%= retry.loop(package_manager, "export DEBIAN_FRONTEND=noninteractive && ${package_manager} update && ${package_manager} update && ${package_manager} add --no-cache curl") %>
@@ -103,7 +103,7 @@ WORKDIR /usr/share/elasticsearch/config
 # Add entrypoint
 ################################################################################
 
-FROM docker.elastic.co/wolfi/chainguard-base-fips:latest@sha256:a9e07d97c55f5b854b1c203509e57b6f466eb26318540b2e1b59734355f747b6
+FROM docker.elastic.co/wolfi/chainguard-base-fips:latest@sha256:68e0781cd592beda39880428985d5fecca1cf2abb18365da73bf1f7ebd994974
 
 RUN <%= retry.loop(package_manager,
           "export DEBIAN_FRONTEND=noninteractive && \n" +

distribution/docker/src/docker/dockerfiles/wolfi/Dockerfile

@@ -24,7 +24,7 @@
 # Extract Elasticsearch artifact
 ################################################################################
 
-FROM docker.elastic.co/wolfi/chainguard-base:latest@sha256:55ee1dca9780931b0929d6eb734f455790c06ddbb59f55008e0cddebfbfd1e2e AS builder
+FROM docker.elastic.co/wolfi/chainguard-base:latest@sha256:3d19648819612728a676ab4061edfb3283bd7117a22c6c4479ee1c1d51831832 AS builder
 
 # Install required packages to extract the Elasticsearch distribution
 RUN <%= retry.loop(package_manager, "export DEBIAN_FRONTEND=noninteractive && ${package_manager} update && ${package_manager} update && ${package_manager} add --no-cache curl") %>
@@ -79,7 +79,7 @@ RUN sed -i -e 's/ES_DISTRIBUTION_TYPE=tar/ES_DISTRIBUTION_TYPE=docker/' bin/elas
 # Add entrypoint
 ################################################################################
 
-FROM docker.elastic.co/wolfi/chainguard-base:latest@sha256:55ee1dca9780931b0929d6eb734f455790c06ddbb59f55008e0cddebfbfd1e2e
+FROM docker.elastic.co/wolfi/chainguard-base:latest@sha256:3d19648819612728a676ab4061edfb3283bd7117a22c6c4479ee1c1d51831832
 
 RUN <%= retry.loop(package_manager,
           "export DEBIAN_FRONTEND=noninteractive && \n" +

docs/changelog/128464.yaml

@@ -0,0 +1,5 @@
+pr: 128464
+summary: Add support for parameters in LIMIT command
+area: ES|QL
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/support/SubscribableListener.java

@@ -32,15 +32,17 @@
 import java.util.concurrent.Executor;
 
 /**
- * An {@link ActionListener} to which other {@link ActionListener} instances can subscribe, such that when this listener is
- * completed it fans-out its result to the subscribed listeners.
+ * An {@link ActionListener} to which other {@link ActionListener} instances can subscribe, such that when this listener is completed it
+ * fans-out its result to the subscribed listeners.
  * <p>
- * If this listener is complete, {@link #addListener} completes the subscribing listener immediately
- * with the result with which this listener was completed. Otherwise, the subscribing listener is retained
- * and completed when this listener is completed.
+ * If this listener is complete, {@link #addListener} completes the subscribing listener immediately with the result with which this
+ * listener was completed. Otherwise, the subscribing listener is retained and completed when this listener is completed.
  * <p>
- * Exceptions are passed to subscribed listeners without modification. {@link ListenableActionFuture} and {@link ListenableFuture} are child
- * classes that provide additional exception handling.
+ * Exceptions are passed to subscribed listeners without modification. {@link ListenableActionFuture} and {@link ListenableFuture} are
+ * subclasses which modify the exceptions passed to subscribed listeners.
+ * <p>
+ * If this listener is completed more than once then all results other than the first (whether successful or otherwise) are silently
+ * discarded. All subscribed listeners will be notified of the same result, exactly once, even if several completions occur concurrently.
  * <p>
  * A sequence of async steps can be chained together using a series of {@link SubscribableListener}s, similar to {@link CompletionStage}
  * (without the {@code catch (Throwable t)}). Listeners can be created for each step, where the next step subscribes to the result of the

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

@@ -56,6 +56,9 @@
  * use {@link ExchangeSourceHandler#addRemoteSink(RemoteSink, boolean, Runnable, int, ActionListener)}.
  */
 public final class ExchangeService extends AbstractLifecycleComponent {
+
+    private static final Logger logger = LogManager.getLogger(ExchangeService.class);
+
     // TODO: Make this a child action of the data node transport to ensure that exchanges
     // are accessed only by the user initialized the session.
     public static final String EXCHANGE_ACTION_NAME = "internal:data/read/esql/exchange";
@@ -71,8 +74,6 @@ public final class ExchangeService extends AbstractLifecycleComponent {
     public static final String INACTIVE_SINKS_INTERVAL_SETTING = "esql.exchange.sink_inactive_interval";
     public static final TimeValue INACTIVE_SINKS_INTERVAL_DEFAULT = TimeValue.timeValueMinutes(5);
 
-    private static final Logger LOGGER = LogManager.getLogger(ExchangeService.class);
-
     private final ThreadPool threadPool;
     private final Executor executor;
     private final BlockFactory blockFactory;
@@ -87,7 +88,7 @@ public ExchangeService(Settings settings, ThreadPool threadPool, String executor
         final var inactiveInterval = settings.getAsTime(INACTIVE_SINKS_INTERVAL_SETTING, INACTIVE_SINKS_INTERVAL_DEFAULT);
         // Run the reaper every half of the keep_alive interval
         this.threadPool.scheduleWithFixedDelay(
-            new InactiveSinksReaper(LOGGER, threadPool, inactiveInterval),
+            new InactiveSinksReaper(threadPool, inactiveInterval),
             TimeValue.timeValueMillis(Math.max(1, inactiveInterval.millis() / 2)),
             executor
         );
@@ -249,12 +250,10 @@ public void messageReceived(ExchangeRequest request, TransportChannel channel, T
     }
 
     private final class InactiveSinksReaper extends AbstractRunnable {
-        private final Logger logger;
         private final TimeValue keepAlive;
         private final ThreadPool threadPool;
 
-        InactiveSinksReaper(Logger logger, ThreadPool threadPool, TimeValue keepAlive) {
-            this.logger = logger;
+        InactiveSinksReaper(ThreadPool threadPool, TimeValue keepAlive) {
             this.keepAlive = keepAlive;
             this.threadPool = threadPool;
         }

x-pack/plugin/esql/src/main/antlr/EsqlBaseParser.g4

@@ -187,7 +187,7 @@ identifierOrParameter
     ;
 
 limitCommand
-    : LIMIT INTEGER_LITERAL
+    : LIMIT constant
     ;
 
 sortCommand

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1128,7 +1128,12 @@ public enum Cap {
         /**
          * Dense vector field type support
          */
-        DENSE_VECTOR_FIELD_TYPE(EsqlCorePlugin.DENSE_VECTOR_FEATURE_FLAG);
+        DENSE_VECTOR_FIELD_TYPE(EsqlCorePlugin.DENSE_VECTOR_FEATURE_FLAG),
+
+        /**
+         * Support parameters for LiMIT command.
+         */
+        PARAMETER_FOR_LIMIT;
 
         private final boolean enabled;