Merge branch 'main' into esql/explain

Author: luigidellaquila

docs/changelog/129606.yaml

@@ -0,0 +1,35 @@
+pr: 129606
+summary: Release FORK in tech preview
+area: ES|QL
+type: feature
+issues: []
+highlight:
+  title: Release FORK in tech preview
+  body: |-
+    Fork is a foundational building block that allows multiple branches of execution.
+    Conceptually, fork is:
+    - a bifurcation of the stream, with all data going to each fork branch, followed by
+    - a merge of the branches, enhanced with a discriminator column called FORK:
+
+    Example:
+
+    [source,yaml]
+    ----------------------------
+    FROM test
+    | FORK
+    ( WHERE content:"fox" )
+    ( WHERE content:"dog" )
+    | SORT _fork
+    ----------------------------
+
+    The FORK command add a discriminator column called `_fork`:
+
+    [source,yaml]
+    ----------------------------
+    | id  | content   | _fork |
+    |-----|-----------|-------|
+    | 3   | brown fox | fork1 |
+    | 4   | white dog | fork2 |
+    ----------------------------
+
+  notable: true

server/src/main/java/org/elasticsearch/index/codec/vectors/DefaultIVFVectorsReader.java

@@ -122,7 +122,7 @@ NeighborQueue scorePostingLists(FieldInfo fieldInfo, KnnCollector knnCollector,
     PostingVisitor getPostingVisitor(FieldInfo fieldInfo, IndexInput indexInput, float[] target, IntPredicate needsScoring)
         throws IOException {
         FieldEntry entry = fields.get(fieldInfo.number);
-        return new MemorySegmentPostingsVisitor(target, indexInput, entry, fieldInfo, needsScoring);
+        return new MemorySegmentPostingsVisitor(target, indexInput.clone(), entry, fieldInfo, needsScoring);
     }
 
     // TODO can we do this in off-heap blocks?

server/src/test/java/org/elasticsearch/index/codec/vectors/IVFVectorsFormatTests.java

@@ -34,6 +34,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.lang.String.format;
 import static org.elasticsearch.index.codec.vectors.IVFVectorsFormat.MAX_VECTORS_PER_CLUSTER;
@@ -128,4 +129,49 @@ public void testSimpleOffHeapSize() throws IOException {
             }
         }
     }
+
+    // this is a modified version of lucene's TestSearchWithThreads test case
+    public void testWithThreads() throws Exception {
+        final int numThreads = random().nextInt(2, 5);
+        final int numSearches = atLeast(100);
+        final int numDocs = atLeast(1000);
+        final int dimensions = random().nextInt(12, 500);
+        try (Directory dir = newDirectory(); IndexWriter w = new IndexWriter(dir, newIndexWriterConfig())) {
+            for (int docCount = 0; docCount < numDocs; docCount++) {
+                final Document doc = new Document();
+                doc.add(new KnnFloatVectorField("f", randomVector(dimensions), VectorSimilarityFunction.EUCLIDEAN));
+                w.addDocument(doc);
+            }
+            w.forceMerge(1);
+            try (IndexReader reader = DirectoryReader.open(w)) {
+                final AtomicBoolean failed = new AtomicBoolean();
+                Thread[] threads = new Thread[numThreads];
+                for (int threadID = 0; threadID < numThreads; threadID++) {
+                    threads[threadID] = new Thread(() -> {
+                        try {
+                            long totSearch = 0;
+                            for (; totSearch < numSearches && failed.get() == false; totSearch++) {
+                                float[] vector = randomVector(dimensions);
+                                LeafReader leafReader = getOnlyLeafReader(reader);
+                                leafReader.searchNearestVectors("f", vector, 10, leafReader.getLiveDocs(), Integer.MAX_VALUE);
+                            }
+                            assertTrue(totSearch > 0);
+                        } catch (Exception exc) {
+                            failed.set(true);
+                            throw new RuntimeException(exc);
+                        }
+                    });
+                    threads[threadID].setDaemon(true);
+                }
+
+                for (Thread t : threads) {
+                    t.start();
+                }
+
+                for (Thread t : threads) {
+                    t.join();
+                }
+            }
+        }
+    }
 }

x-pack/plugin/esql/qa/testFixtures/src/main/resources/rerank.csv-spec

@@ -96,7 +96,7 @@ book_no:keyword | title:text                                            | author
 
 
 reranker after RRF
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 required_capability: rerank

x-pack/plugin/esql/qa/testFixtures/src/main/resources/rrf.csv-spec

@@ -3,7 +3,7 @@
 //
 
 simpleRrf
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 
@@ -22,7 +22,7 @@ _score:double | _fork:keyword | emp_no:integer
 ;
 
 rrfWithMatchAndScore
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 
@@ -44,7 +44,7 @@ _score:double | _fork:keyword  | _id:keyword
 ;
 
 rrfWithDisjunctionAndPostFilter
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 
@@ -66,7 +66,7 @@ _score:double | _fork:keyword  | _id:keyword
 ;
 
 rrfWithStats
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 
@@ -86,7 +86,7 @@ count_fork:long | _fork:keyword
 ;
 
 rrfWithMultipleForkBranches
-required_capability: fork
+required_capability: fork_v9
 required_capability: rrf
 required_capability: match_operator_colon
 
@@ -112,6 +112,7 @@ _score:double | author:keyword                    | title:keyword        | _fork
 ;
 
 rrfWithSemanticSearch
+required_capability: fork_v9
 required_capability: rrf
 required_capability: semantic_text_field_caps
 required_capability: metadata_score

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ForkIT.java

@@ -32,7 +32,7 @@ public class ForkIT extends AbstractEsqlIntegTestCase {
 
     @Before
     public void setupIndex() {
-        assumeTrue("requires FORK capability", EsqlCapabilities.Cap.FORK.isEnabled());
+        assumeTrue("requires FORK capability", EsqlCapabilities.Cap.FORK_V9.isEnabled());
         createAndPopulateIndices();
     }
 

x-pack/plugin/esql/src/main/antlr/EsqlBaseLexer.tokens

@@ -58,11 +58,11 @@ processingCommand
     | changePointCommand
     | completionCommand
     | sampleCommand
+    | forkCommand
     // in development
     | {this.isDevVersion()}? inlinestatsCommand
     | {this.isDevVersion()}? lookupCommand
     | {this.isDevVersion()}? insistCommand
-    | {this.isDevVersion()}? forkCommand
     | {this.isDevVersion()}? rerankCommand
     | {this.isDevVersion()}? rrfCommand
     ;
@@ -282,7 +282,7 @@ insistCommand
     ;
 
 forkCommand
-    : DEV_FORK forkSubQueries
+    : FORK forkSubQueries
     ;
 
 forkSubQueries

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

@@ -9,7 +9,7 @@ lexer grammar Fork;
 //
 // Fork
 //
-DEV_FORK :        {this.isDevVersion()}? 'fork'          -> pushMode(FORK_MODE);
+FORK : 'fork' -> pushMode(FORK_MODE);
 
 mode FORK_MODE;
 // commands needs to break out of their mode and the default mode when they encounter RP