Merge branch 'main' into ivf_hkmeans_struc2

Author: john-wagster

README.asciidoc

@@ -275,7 +275,7 @@ For the complete Elasticsearch documentation visit
 https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html[elastic.co].
 
 For information about our documentation processes, see the
-xref:docs/README.asciidoc[docs README].
+xref:https://github.com/elastic/elasticsearch/blob/main/docs/README.md[docs README].
 
 [[examples]]
 == Examples and guides

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

@@ -70,11 +70,11 @@ public class QueryPlanningBenchmark {
     private EsqlParser defaultParser;
     private Analyzer manyFieldsAnalyzer;
     private LogicalPlanOptimizer defaultOptimizer;
+    private Configuration config;
 
     @Setup
     public void setup() {
-
-        var config = new Configuration(
+        this.config = new Configuration(
             DateUtils.UTC,
             Locale.US,
             null,
@@ -116,7 +116,7 @@ public void setup() {
     }
 
     private LogicalPlan plan(EsqlParser parser, Analyzer analyzer, LogicalPlanOptimizer optimizer, String query) {
-        var parsed = parser.createStatement(query, new QueryParams(), telemetry);
+        var parsed = parser.createStatement(query, new QueryParams(), telemetry, config);
         var analyzed = analyzer.analyze(parsed);
         var optimized = optimizer.optimize(analyzed);
         return optimized;

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/ValuesSourceReaderBenchmark.java renamed to benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/ValuesSourceReaderBenchmark.java

@@ -7,7 +7,7 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-package org.elasticsearch.benchmark.compute.operator;
+package org.elasticsearch.benchmark._nightly.esql;
 
 import org.apache.lucene.document.FieldType;
 import org.apache.lucene.document.NumericDocValuesField;
@@ -41,7 +41,8 @@
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.lucene.LuceneSourceOperator;
 import org.elasticsearch.compute.lucene.ShardRefCounted;
-import org.elasticsearch.compute.lucene.ValuesSourceReaderOperator;
+import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperator;
+import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperatorStatus;
 import org.elasticsearch.compute.operator.topn.TopNOperator;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.index.IndexSettings;
@@ -84,10 +85,18 @@
 @State(Scope.Thread)
 @Fork(1)
 public class ValuesSourceReaderBenchmark {
+    private static final String[] SUPPORTED_LAYOUTS = new String[] { "in_order", "shuffled", "shuffled_singles" };
+    private static final String[] SUPPORTED_NAMES = new String[] {
+        "long",
+        "int",
+        "double",
+        "keyword",
+        "stored_keyword",
+        "3_stored_keywords" };
+
     private static final int BLOCK_LENGTH = 16 * 1024;
     private static final int INDEX_SIZE = 10 * BLOCK_LENGTH;
     private static final int COMMIT_INTERVAL = 500;
-    private static final BigArrays BIG_ARRAYS = BigArrays.NON_RECYCLING_INSTANCE;
     private static final BlockFactory blockFactory = BlockFactory.getInstance(
         new NoopCircuitBreaker("noop"),
         BigArrays.NON_RECYCLING_INSTANCE
@@ -103,8 +112,8 @@ static void selfTest() {
             ValuesSourceReaderBenchmark benchmark = new ValuesSourceReaderBenchmark();
             benchmark.setupIndex();
             try {
-                for (String layout : ValuesSourceReaderBenchmark.class.getField("layout").getAnnotationsByType(Param.class)[0].value()) {
-                    for (String name : ValuesSourceReaderBenchmark.class.getField("name").getAnnotationsByType(Param.class)[0].value()) {
+                for (String layout : ValuesSourceReaderBenchmark.SUPPORTED_LAYOUTS) {
+                    for (String name : ValuesSourceReaderBenchmark.SUPPORTED_NAMES) {
                         benchmark.layout = layout;
                         benchmark.name = name;
                         try {
@@ -118,7 +127,7 @@ static void selfTest() {
             } finally {
                 benchmark.teardownIndex();
             }
-        } catch (IOException | NoSuchFieldException e) {
+        } catch (IOException e) {
             throw new AssertionError(e);
         }
     }
@@ -320,10 +329,10 @@ public FieldNamesFieldMapper.FieldNamesFieldType fieldNames() {
      *     each page has a single document rather than {@code BLOCK_SIZE} docs.</li>
      * </ul>
      */
-    @Param({ "in_order", "shuffled", "shuffled_singles" })
+    @Param({ "in_order", "shuffled" })
     public String layout;
 
-    @Param({ "long", "int", "double", "keyword", "stored_keyword", "3_stored_keywords" })
+    @Param({ "long", "keyword", "stored_keyword" })
     public String name;
 
     private Directory directory;
@@ -343,7 +352,7 @@ public void benchmark() {
         );
         long sum = 0;
         for (Page page : pages) {
-            op.addInput(page);
+            op.addInput(page.shallowCopy());
             switch (name) {
                 case "long" -> {
                     LongVector values = op.getOutput().<LongBlock>getBlock(1).asVector();
@@ -411,7 +420,7 @@ public void benchmark() {
             throw new AssertionError("[" + layout + "][" + name + "] expected [" + expected + "] but was [" + sum + "]");
         }
         boolean foundStoredFieldLoader = false;
-        ValuesSourceReaderOperator.Status status = (ValuesSourceReaderOperator.Status) op.status();
+        ValuesSourceReaderOperatorStatus status = (ValuesSourceReaderOperatorStatus) op.status();
         for (Map.Entry<String, Integer> e : status.readersBuilt().entrySet()) {
             if (e.getKey().indexOf("stored_fields") >= 0) {
                 foundStoredFieldLoader = true;

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/ValuesAggregatorBenchmark.java

@@ -95,7 +95,8 @@ static void selfTest() {
         try {
             for (String groups : ValuesAggregatorBenchmark.class.getField("groups").getAnnotationsByType(Param.class)[0].value()) {
                 for (String dataType : ValuesAggregatorBenchmark.class.getField("dataType").getAnnotationsByType(Param.class)[0].value()) {
-                    run(Integer.parseInt(groups), dataType, 10);
+                    run(Integer.parseInt(groups), dataType, 10, 0);
+                    run(Integer.parseInt(groups), dataType, 10, 1);
                 }
             }
         } catch (NoSuchFieldException e) {
@@ -113,7 +114,10 @@ static void selfTest() {
     @Param({ BYTES_REF, INT, LONG })
     public String dataType;
 
-    private static Operator operator(DriverContext driverContext, int groups, String dataType) {
+    @Param({ "0", "1" })
+    public int numOrdinalMerges;
+
+    private static Operator operator(DriverContext driverContext, int groups, String dataType, int numOrdinalMerges) {
         if (groups == 1) {
             return new AggregationOperator(
                 List.of(supplier(dataType).aggregatorFactory(AggregatorMode.SINGLE, List.of(0)).apply(driverContext)),
@@ -125,7 +129,24 @@ private static Operator operator(DriverContext driverContext, int groups, String
             List.of(supplier(dataType).groupingAggregatorFactory(AggregatorMode.SINGLE, List.of(1))),
             () -> BlockHash.build(groupSpec, driverContext.blockFactory(), 16 * 1024, false),
             driverContext
-        );
+        ) {
+            @Override
+            public Page getOutput() {
+                mergeOrdinal();
+                return super.getOutput();
+            }
+
+            // simulate OrdinalsGroupingOperator
+            void mergeOrdinal() {
+                var merged = supplier(dataType).groupingAggregatorFactory(AggregatorMode.SINGLE, List.of(1)).apply(driverContext);
+                for (int i = 0; i < numOrdinalMerges; i++) {
+                    for (int p = 0; p < groups; p++) {
+                        merged.addIntermediateRow(p, aggregators.getFirst(), p);
+                    }
+                }
+                aggregators.set(0, merged);
+            }
+        };
     }
 
     private static AggregatorFunctionSupplier supplier(String dataType) {
@@ -331,12 +352,12 @@ private static Block groupingBlock(int groups) {
 
     @Benchmark
     public void run() {
-        run(groups, dataType, OP_COUNT);
+        run(groups, dataType, OP_COUNT, numOrdinalMerges);
     }
 
-    private static void run(int groups, String dataType, int opCount) {
+    private static void run(int groups, String dataType, int opCount, int numOrdinalMerges) {
         DriverContext driverContext = driverContext();
-        try (Operator operator = operator(driverContext, groups, dataType)) {
+        try (Operator operator = operator(driverContext, groups, dataType, numOrdinalMerges)) {
             Page page = page(groups, dataType);
             for (int i = 0; i < opCount; i++) {
                 operator.addInput(page.shallowCopy());

benchmarks/src/main/java/org/elasticsearch/benchmark/vector/Int7uScorerBenchmark.java

@@ -24,6 +24,8 @@
 import org.apache.lucene.util.quantization.ScalarQuantizer;
 import org.elasticsearch.common.logging.LogConfigurator;
 import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 import org.elasticsearch.simdvec.VectorScorerFactory;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -61,6 +63,10 @@ public class Int7uScorerBenchmark {
 
     static {
         LogConfigurator.configureESLogging(); // native access requires logging to be initialized
+        if (supportsHeapSegments() == false) {
+            final Logger LOG = LogManager.getLogger(Int7uScorerBenchmark.class);
+            LOG.warn("*Query targets cannot run on " + "JDK " + Runtime.version());
+        }
     }
 
     @Param({ "96", "768", "1024" })
@@ -129,15 +135,17 @@ public void setup() throws IOException {
         nativeSqrScorer = factory.getInt7SQVectorScorerSupplier(EUCLIDEAN, in, values, scoreCorrectionConstant).get().scorer();
         nativeSqrScorer.setScoringOrdinal(0);
 
-        // setup for getInt7SQVectorScorer / query vector scoring
-        float[] queryVec = new float[dims];
-        for (int i = 0; i < dims; i++) {
-            queryVec[i] = ThreadLocalRandom.current().nextFloat();
+        if (supportsHeapSegments()) {
+            // setup for getInt7SQVectorScorer / query vector scoring
+            float[] queryVec = new float[dims];
+            for (int i = 0; i < dims; i++) {
+                queryVec[i] = ThreadLocalRandom.current().nextFloat();
+            }
+            luceneDotScorerQuery = luceneScorer(values, VectorSimilarityFunction.DOT_PRODUCT, queryVec);
+            nativeDotScorerQuery = factory.getInt7SQVectorScorer(VectorSimilarityFunction.DOT_PRODUCT, values, queryVec).get();
+            luceneSqrScorerQuery = luceneScorer(values, VectorSimilarityFunction.EUCLIDEAN, queryVec);
+            nativeSqrScorerQuery = factory.getInt7SQVectorScorer(VectorSimilarityFunction.EUCLIDEAN, values, queryVec).get();
         }
-        luceneDotScorerQuery = luceneScorer(values, VectorSimilarityFunction.DOT_PRODUCT, queryVec);
-        nativeDotScorerQuery = factory.getInt7SQVectorScorer(VectorSimilarityFunction.DOT_PRODUCT, values, queryVec).get();
-        luceneSqrScorerQuery = luceneScorer(values, VectorSimilarityFunction.EUCLIDEAN, queryVec);
-        nativeSqrScorerQuery = factory.getInt7SQVectorScorer(VectorSimilarityFunction.EUCLIDEAN, values, queryVec).get();
     }
 
     @TearDown
@@ -208,6 +216,10 @@ public float squareDistanceNativeQuery() throws IOException {
         return nativeSqrScorerQuery.score(1);
     }
 
+    static boolean supportsHeapSegments() {
+        return Runtime.version().feature() >= 22;
+    }
+
     QuantizedByteVectorValues vectorValues(int dims, int size, IndexInput in, VectorSimilarityFunction sim) throws IOException {
         var sq = new ScalarQuantizer(0.1f, 0.9f, (byte) 7);
         var slice = in.slice("values", 0, in.length());

benchmarks/src/test/java/org/elasticsearch/benchmark/compute/operator/ValuesSourceReaderBenchmarkTests.java renamed to benchmarks/src/test/java/org/elasticsearch/benchmark/_nightly/esql/ValuesSourceReaderBenchmarkTests.java

@@ -7,7 +7,7 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-package org.elasticsearch.benchmark.compute.operator;
+package org.elasticsearch.benchmark._nightly.esql;
 
 import org.elasticsearch.test.ESTestCase;
 

benchmarks/src/test/java/org/elasticsearch/benchmark/vector/Int7uScorerBenchmarkTests.java

@@ -42,8 +42,10 @@ public void testDotProduct() throws Exception {
                 assertEquals(expected, bench.dotProductLucene(), delta);
                 assertEquals(expected, bench.dotProductNative(), delta);
 
-                expected = bench.dotProductLuceneQuery();
-                assertEquals(expected, bench.dotProductNativeQuery(), delta);
+                if (Int7uScorerBenchmark.supportsHeapSegments()) {
+                    expected = bench.dotProductLuceneQuery();
+                    assertEquals(expected, bench.dotProductNativeQuery(), delta);
+                }
             } finally {
                 bench.teardown();
             }
@@ -60,8 +62,10 @@ public void testSquareDistance() throws Exception {
                 assertEquals(expected, bench.squareDistanceLucene(), delta);
                 assertEquals(expected, bench.squareDistanceNative(), delta);
 
-                expected = bench.squareDistanceLuceneQuery();
-                assertEquals(expected, bench.squareDistanceNativeQuery(), delta);
+                if (Int7uScorerBenchmark.supportsHeapSegments()) {
+                    expected = bench.squareDistanceLuceneQuery();
+                    assertEquals(expected, bench.squareDistanceNativeQuery(), delta);
+                }
             } finally {
                 bench.teardown();
             }

build-tools-internal/src/main/resources/forbidden/es-all-signatures.txt

@@ -35,6 +35,8 @@ java.nio.channels.SocketChannel#connect(java.net.SocketAddress)
 # org.elasticsearch.core.Booleans#parseBoolean(java.lang.String) directly on the string.
 @defaultMessage use org.elasticsearch.core.Booleans#parseBoolean(java.lang.String)
 java.lang.Boolean#getBoolean(java.lang.String)
+java.lang.Boolean#parseBoolean(java.lang.String)
+java.lang.Boolean#valueOf(java.lang.String)
 
 org.apache.lucene.util.IOUtils @ use @org.elasticsearch.core.internal.io instead
 

docs/changelog/128917.yaml

@@ -0,0 +1,6 @@
+pr: 128917
+summary: Adopt a "LogicalPlan" approach to running multiple sub-queries (with INLINESTATS
+  so far)
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/129013.yaml

@@ -0,0 +1,13 @@
+pr: 129013
+summary: "Add remote index support to LOOKUP JOIN"
+area: ES|QL
+type: feature
+issues: [ ]
+highlight:
+  title: Add remote index support to LOOKUP JOIN
+  body: |-
+    Queries containing LOOKUP JOIN now can be preformed on cross-cluster indices, for example:
+    [source,yaml]
+    ----------------------------
+    FROM logs-*, remote:logs-* | LOOKUP JOIN clients on ip | SORT timestamp | LIMIT 100
+    ----------------------------