ES|QL: Initial grammar and changes for FORK (snapshot)

docs/changelog/121948.yaml

@@ -0,0 +1,5 @@
+pr: 121948
+summary: Add initial grammar and changes for FORK
+area: ES|QL
+type: feature
+issues: []

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.compute.operator;
+
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BlockFactory;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.core.Releasables;
+
+import java.util.List;
+import java.util.ListIterator;
+import java.util.function.Supplier;
+
+/**
+ * A merge operator is effectively a "fan-in" operator - accepts input
+ * from several sources and provides it in a single output.
+ */
+public class MergeOperator extends SourceOperator {
+
+    public record MergeOperatorFactory(BlockSuppliers suppliers) implements SourceOperatorFactory {
+        @Override
+        public String describe() {
+            return "MergeOperator[suppliers=" + suppliers + "]";
+        }
+
+        @Override
+        public SourceOperator get(DriverContext driverContext) {
+            return new MergeOperator(driverContext.blockFactory(), suppliers);
+        }
+    }
+
+    private final BlockFactory blockFactory;
+    private final BlockSuppliers suppliers;
+    private boolean finished;
+    private ListIterator<Block[]> subPlanBlocks;
+
+    public MergeOperator(BlockFactory blockFactory, BlockSuppliers suppliers) {
+        super();
+        this.blockFactory = blockFactory;
+        this.suppliers = suppliers;
+    }
+
+    public interface BlockSuppliers extends Supplier<List<Block[]>> {};
+
+    @Override
+    public void finish() {}
+
+    @Override
+    public boolean isFinished() {
+        return finished;
+    }
+
+    @Override
+    public Page getOutput() {
+        if (subPlanBlocks == null) {
+            subPlanBlocks = suppliers.get().listIterator();
+        }
+
+        if (subPlanBlocks.hasNext()) {
+            return new Page(subPlanBlocks.next());
+        }
+        finished = true;
+        return null;
+    }
+
+    @Override
+    public void close() {
+        // release blocks from any subplan not fully consumed.
+        if (subPlanBlocks != null) {
+            while (subPlanBlocks.hasNext()) {
+                Releasables.close(subPlanBlocks.next());
+            }
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "MergeOperator[subPlanBlocks=" + subPlanBlocks + "]";
+    }
+}

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

@@ -0,0 +1,94 @@
+//
+// CSV spec for FORK command
+//
+
+simpleFork
+required_capability: fork
+
+FROM employees
+| FORK ( WHERE emp_no == 10001 )
+       ( WHERE emp_no == 10002 )
+| KEEP emp_no, _fork
+| SORT emp_no
+;
+
+emp_no:integer | _fork:keyword
+10001          | fork1
+10002          | fork2
+;
+
+forkWithWhereSortAndLimit
+required_capability: fork
+
+FROM employees
+| FORK ( WHERE hire_date < "1985-03-01T00:00:00Z" | SORT first_name | LIMIT 5 )
+       ( WHERE hire_date < "1988-03-01T00:00:00Z" | SORT first_name | LIMIT 5 )
+| KEEP emp_no, first_name, _fork
+| SORT emp_no, _fork
+;
+
+emp_no:integer | first_name:keyword | _fork:keyword
+10002          | Bezalel            | fork2
+10009          | Sumant             | fork1
+10014          | Berni              | fork2
+10048          | Florian            | fork1
+10058          | Berhard            | fork2
+10060          | Breannda           | fork2
+10094          | Arumugam           | fork2
+;
+
+fiveFork
+required_capability: fork
+
+FROM employees
+| FORK ( WHERE emp_no == 10005 )
+       ( WHERE emp_no == 10004 )
+       ( WHERE emp_no == 10003 )
+       ( WHERE emp_no == 10002 )
+       ( WHERE emp_no == 10001 )
+| KEEP  _fork, emp_no
+| SORT _fork
+;
+
+_fork:keyword | emp_no:integer 
+fork1         | 10005
+fork2         | 10004
+fork3         | 10003
+fork4         | 10002
+fork5         | 10001
+;
+
+forkWithWhereSortDescAndLimit
+required_capability: fork
+
+FROM employees
+| FORK ( WHERE hire_date < "1985-03-01T00:00:00Z" | SORT first_name DESC | LIMIT 2 )
+       ( WHERE hire_date < "1988-03-01T00:00:00Z" | SORT first_name DESC NULLS LAST | LIMIT 2 )
+| KEEP _fork, emp_no, first_name
+| SORT _fork, first_name DESC
+;
+
+_fork:keyword | emp_no:integer | first_name:keyword 
+fork1         | 10009          | Sumant            
+fork1         | 10048          | Florian           
+fork2         | 10081          | Zhongwei          
+fork2         | 10087          | Xinglin
+;
+
+forkWithCommonPrefilter
+required_capability: fork
+
+FROM employees
+| WHERE emp_no > 10050
+| FORK ( SORT emp_no ASC | LIMIT 2 )
+       ( SORT emp_no DESC NULLS LAST | LIMIT 2 )
+| KEEP _fork, emp_no
+| SORT _fork, emp_no
+;
+
+_fork:keyword | emp_no:integer
+fork1         | 10051
+fork1         | 10052
+fork2         | 10099
+fork2         | 10100
+;