Have Row - LocalRelation transformation use a copying blocks approach

Author: astefan

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

@@ -509,61 +509,112 @@ ALG            |Algiers        |385            |major          |1
 ALL            |Albenga        |499            |mid            |1                  |Albenga
 ;
 
-shadowing-Ignore
+shadowing
 required_capability: join_planning_v1
+required_capability: inlinestats_v8
 
 ROW left = "left", client_ip = "172.21.0.5", env = "env", right = "right"
-| INLINESTATS env=VALUES(right) BY client_ip
+| INLINESTATS env = VALUES(right) BY client_ip
 ;
 
-left:keyword | client_ip:keyword | right:keyword | env:keyword
-left         | 172.21.0.5        | right         | right
+left:keyword |  right:keyword | env:keyword | client_ip:keyword
+left         | right          | right       | 172.21.0.5        
 ;
 
-shadowingMulti-Ignore
+shadowingMulti
 required_capability: join_planning_v1
+required_capability: inlinestats_v8
 
 ROW left = "left", airport = "Zurich Airport ZRH", city = "Zürich", middle = "middle", region = "North-East Switzerland", right = "right"
 | INLINESTATS airport=VALUES(left), region=VALUES(left), city_boundary=VALUES(left) BY city
 ;
 
-left:keyword | city:keyword | middle:keyword | right:keyword | airport:keyword | region:keyword | city_boundary:keyword
-left         | Zürich       | middle         | right         |            left |           left | left
+left:keyword | middle:keyword | right:keyword | airport:keyword | region:keyword | city_boundary:keyword | city:keyword 
+left         | middle         | right         |            left |           left | left                  | Zürich
 ;
 
-shadowingSelf-Ignore
+shadowingSelf
 required_capability: join_planning_v1
+required_capability: inlinestats_v8
 
-ROW city="Raleigh"
-| INLINESTATS city=COUNT(city)
+ROW city = "Raleigh"
+| INLINESTATS city = COUNT(city)
 ;
 
 city:long
 1
 ;
 
-shadowingSelfBySelf-Ignore
+shadowingSelfBySelf
 required_capability: join_planning_v1
+required_capability: inlinestats_v8
 
-ROW city="Raleigh"
-| INLINESTATS city=COUNT(city) BY city
+ROW city = "Raleigh"
+| INLINESTATS city = COUNT(city) BY city
 ;
+warning:Line 2:15: Field 'city' shadowed by field at line 2:37
 
-city:long
-1
+city:keyword
+Raleigh
 ;
 
-shadowingInternal-Ignore
+shadowingInternal
 required_capability: join_planning_v1
+required_capability: inlinestats_v8
 
 ROW city = "Zürich"
-| INLINESTATS x=VALUES(city), x=VALUES(city)
+| INLINESTATS x = VALUES(city), x = VALUES(city)
 ;
+warning:Line 2:15: Field 'x' shadowed by field at line 2:33
 
 city:keyword | x:keyword
 Zürich       | Zürich
 ;
 
+multiInlinestatsWithRow
+required_capability: inlinestats_v8
+
+row x = 1
+| inlinestats x = max(x) + min(x)
+| eval y = x + 1
+| inlinestats sum(y)
+| inlinestats count(y), count(x)
+;
+
+  x:integer    |    y:integer  |  sum(y):long  | count(y):long | count(x):long    
+2              |3              |3              |1              |1
+;
+
+ignoreUnusedEvaledValue-Ignore
+required_capability: inlinestats_v8
+// fails with expected [keys] to be non-empty
+
+ROW x = 1
+| INLINESTATS max(x)
+| EVAL y = x + 1
+| KEEP x
+;
+
+x:integer
+1
+;
+
+ignoreUnusedEvaledValue2-Ignore
+required_capability: inlinestats_v8
+// fails with expected [keys] to be non-empty
+
+from employees
+| inlinestats max(salary)
+| eval y = salary + 1
+| keep salary
+| sort salary desc
+| limit 1
+;
+
+ salary:integer     
+74999     
+;
+
 byConstant
 required_capability: inlinestats_v8
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRowAsLocalRelation.java

@@ -11,8 +11,8 @@
 import org.elasticsearch.xpack.esql.optimizer.LogicalOptimizerContext;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.Row;
+import org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
-import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 
 import java.util.ArrayList;
@@ -29,6 +29,6 @@ protected LogicalPlan rule(Row row, LogicalOptimizerContext context) {
         List<Object> values = new ArrayList<>(fields.size());
         fields.forEach(f -> values.add(f.child().fold(context.foldCtx())));
         var blocks = BlockUtils.fromListRow(PlannerUtils.NON_BREAKING_BLOCK_FACTORY, values);
-        return new LocalRelation(row.source(), row.output(), LocalSupplier.of(blocks));
+        return new LocalRelation(row.source(), row.output(), new CopyingLocalSupplier(blocks));
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/local/CopyingLocalSupplier.java

@@ -0,0 +1,31 @@
+/*
+ * 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.xpack.esql.plan.logical.local;
+
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.BlockUtils;
+import org.elasticsearch.xpack.esql.planner.PlannerUtils;
+
+/**
+ * A {@link LocalSupplier} that allways creates a new copy of the {@link Block}s initially provided at creation time.
+ */
+public class CopyingLocalSupplier extends ImmediateLocalSupplier {
+
+    public CopyingLocalSupplier(Block[] blocks) {
+        super(blocks);
+    }
+
+    @Override
+    public Block[] get() {
+        Block[] blockCopies = new Block[blocks.length];
+        for (int i = 0; i < blockCopies.length; i++) {
+            blockCopies[i] = BlockUtils.deepCopyOf(blocks[i], PlannerUtils.NON_BREAKING_BLOCK_FACTORY);
+        }
+        return blockCopies;
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/local/ImmediateLocalSupplier.java

@@ -18,7 +18,7 @@
  * A {@link LocalSupplier} that contains already filled {@link Block}s.
  */
 public class ImmediateLocalSupplier implements LocalSupplier {
-    private final Block[] blocks;
+    final Block[] blocks;
 
     public ImmediateLocalSupplier(Block[] blocks) {
         this.blocks = blocks;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -235,8 +235,7 @@ private void executeSubPlans(
 
     private LogicalPlanTuple firstSubPlan(LogicalPlan optimizedPlan) {
         Holder<LogicalPlanTuple> subPlan = new Holder<>();
-        // Collect the first inlinejoin (bottom up in the tree or, viewing from the user-friendly query pov, the closest to ES source
-        // inlinestats command)
+        // Collect the first inlinejoin (bottom up in the tree)
         optimizedPlan.forEachUp(InlineJoin.class, ij -> {
             // extract the right side of the plan and replace its source
             if (subPlan.get() == null && ij.right().anyMatch(p -> p instanceof StubRelation)) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/SessionUtils.java

@@ -26,7 +26,9 @@ public static Block[] fromPages(List<Attribute> schema, List<Page> pages) {
         // Limit ourselves to 1mb of results similar to LOOKUP for now.
         long bytesUsed = pages.stream().mapToLong(Page::ramBytesUsedByBlocks).sum();
         if (bytesUsed > ByteSizeValue.ofMb(1).getBytes()) {
-            throw new IllegalArgumentException("first phase result too large [" + ByteSizeValue.ofBytes(bytesUsed) + "] > 1mb");
+            throw new IllegalArgumentException(
+                "INLINESTATS sub-plan execution results too large [" + ByteSizeValue.ofBytes(bytesUsed) + "] > 1mb"
+            );
         }
         int positionCount = pages.stream().mapToInt(Page::getPositionCount).sum();
         Block.Builder[] builders = new Block.Builder[schema.size()];