ESQL: Fix double release in inline stats when LocalRelation is reused (elastic#136467)

kanoshiou · chrisparrinello · commit ffbc38b0f350 · 2025-10-24T12:39:18.000-05:00
When the optimizer creates a `LocalRelation` using `LocalSupplier.of()` (which delegates to `ImmediateLocalSupplier`), the same page instance can be referenced and released multiple times during inline stats execution. This happens because the `LocalRelation` serves as a source for both sides of the `InlineJoin` - once for the left-hand side and again for the right-hand side, causing a double release error and connection closure. This PR modifies the `InlineJoin.firstSubPlan()` method to ensure that all `LocalRelation` nodes use `CopyingLocalSupplier` (if they don't already) after extracting the subplan. This prevents double release issues when the same page is reused across executions. Closes elastic#135679
diff --git a/docs/changelog/136467.yaml b/docs/changelog/136467.yaml
@@ -0,0 +1,6 @@
+pr: 136467
+summary: "ESQL: Fix double release in inline stats when `LocalRelation` is reused"
+area: ES|QL
+type: bug
+issues:
+  - 135679
diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec
@@ -3990,3 +3990,179 @@ Canada            |English               |null                       |null
 Mv-Land           |Mv-Lang               |null                       |null
 Mv-Land2          |Mv-Lang2              |null                       |null
 ;
+
+// PruneColumns
+inlineStatsAfterPruningAggregate
+required_capability: inline_stats_double_release_fix
+
+row a = 1
+| stats b = max(a)
+| inline stats c = count(*)
+| drop b
+;
+
+c:long
+1
+;
+
+// PropagateEmptyRelation
+inlineStatsAfterPruningAggregate2
+required_capability: inline_stats_double_release_fix
+
+row a = 12
+| where false
+| stats b = max(a)
+| inline stats c = count(b)
+;
+
+b:integer | c:long
+null      | 0
+;
+
+
+// ReplaceStatsFilteredAggWithEval
+inlineStatsAfterPruningAggregate3
+required_capability: inline_stats_double_release_fix
+
+row a= 12
+| stats b = sum(a) where false
+| inline stats c = max(b)
+;
+
+b:long | c:long
+null   | null
+;
+
+
+inlineStatsAfterPruningAggregate4
+required_capability: inline_stats_double_release_fix
+
+from k8s
+| stats PRezTcny = median(network.cost), `network.cost` = count(network.cost) 
+| drop network.cost 
+| inline stats QaoRGYyf = count(*) 
+| drop `PRezTcny`
+;
+
+QaoRGYyf:long    
+1              
+;
+
+
+inlineStatsAfterPruningAggregate5
+// https://github.com/elastic/elasticsearch/issues/135679
+required_capability: inline_stats_double_release_fix
+required_capability: join_lookup_v12
+
+from colors,airports_web,boo*
+| rename scalerank as language_code
+| lookup join languages_lookup on language_code
+| keep `ratings`, `rgb_vector`
+| mv_expand ratings
+| keep `ratings`, `rgb_vector`
+| inline stats  rycaPYFzpAId = absent(ratings), WkdxfjwuR = count(ratings) + avg(ratings)
+| rename ratings AS `WkdxfjwuR`
+| inline stats  WkdxfjwuR = sum(WkdxfjwuR) + median(WkdxfjwuR), sjivSBvV = count(*)
+| sort sjivSBvV NULLS FIRST
+| where false OR  NOT true OR false OR  NOT false
+| where false OR false AND false AND  NOT true AND false
+| stats  rycaPYFzpAId = median(sjivSBvV) + count(sjivSBvV) + sum(sjivSBvV), NUiLBJXQoMXd = present(sjivSBvV)
+| mv_expand `NUiLBJXQoMXd`
+| inline stats  NUiLBJXQoMXd = count(rycaPYFzpAId)
+;
+
+rycaPYFzpAId:double  | NUiLBJXQoMXd:long  
+null                 | 0              
+;
+
+
+inlineStatsAfterPruningAggregate6
+// https://github.com/elastic/elasticsearch/issues/135679
+required_capability: inline_stats_double_release_fix
+required_capability: join_lookup_v12
+
+from d*,*
+| rename scalerank as other2
+| rename author.keyword as name_str
+| rename intersects as is_active_bool
+| rename year as id_int
+| rename status as other1 
+| lookup join multi_column_joinable_lookup on other2, name_str, is_active_bool, id_int, other1 
+| rename street AS `huZQQIKVli`, `network.eth0.currently_connected_clients` AS is_rehired 
+| eval  GhMzAVXivNI = message.raw, VDJKNsopeKQq = substring(threat_level, 1, 3), xeUVJNra = length(huZQQIKVli), kOQjzWgZNhwI = false 
+| where false AND  NOT false AND true AND  NOT false 
+| CHANGE_POINT salary_change.int ON birth_date AS UpYfyPcYFkes, EwnNgKIrDEKS 
+| sort owner.name DESC NULLS LAST, message.raw, country.keyword DESC NULLS LAST, UpYfyPcYFkes ASC NULLS FIRST, name_str ASC NULLS FIRST, num, event_duration DESC NULLS LAST, extra2 DESC, languages.byte ASC NULLS FIRST, client.ip NULLS LAST, ip1, collection NULLS FIRST, city.country.name DESC, destination.IP DESC NULLS LAST, host.version DESC, hex_code ASC NULLS LAST, language.code ASC NULLS LAST, language.name ASC, details DESC, language_name DESC NULLS FIRST, event ASC, salary_change.keyword DESC NULLS FIRST, lk ASC 
+| stats  CLnBmEGor = sum(height.double) + sum(height.double) + avg(height.double), JLSxMYld = count(extra2) + max(extra2) + avg(extra2), `network.total_bytes_out` = min(language_code_integer), `version` = values(env), contains = min(languages.short) + min(languages.short) 
+| mv_expand `CLnBmEGor` 
+| dissect version "%{JLSxMYld} %{LTMJrburyt}" 
+| enrich languages_policy on LTMJrburyt 
+| inline stats  QyCBayGHtAF = count(CLnBmEGor), yUMkeLezSN = present(contains + network.total_bytes_out), `JLSxMYld` = max(contains), GYbfUvbf = count(*)
+;
+
+CLnBmEGor:double | network.total_bytes_out:integer | version:keyword | contains:integer | LTMJrburyt:keyword | language_name:keyword | QyCBayGHtAF:long | yUMkeLezSN:boolean | JLSxMYld:integer | GYbfUvbf:long
+null             | null                            | null            | null             | null               | null                  | 0                | false              | null             | 1
+;
+
+
+inlineStatsAfterPruningAggregate7
+// https://github.com/elastic/elasticsearch/issues/135679
+required_capability: inline_stats_double_release_fix
+required_capability: join_lookup_v12
+required_capability: fork_v9
+
+from app_log* 
+| FORK  (where false OR  NOT true) (where true) (where true) (where false OR true) (where false) (where true | mv_expand @timestamp) 
+| WHERE _fork == "fork2" 
+| DROP _fork
+| rename service_id as name_str 
+| lookup join multi_column_joinable_lookup on name_str 
+| inline stats  `is_active_bool` = count(id_int) + max(id_int), siofAMlpoHvh = max(id_int) + max(id_int) + max(id_int), `message` = present(other1), `id_int` = max(id_int) 
+| stats  `message` = count(is_active_bool), id_int2 = median(is_active_bool), `siofAMlpoHvh` = count(*), `id_int` = max(id_int) 
+| eval  NUtENNcs = message, message = "zaYFcCTy", id_int2 = null, VizcTbjMnBlQ = siofAMlpoHvh, siofAMlpoHvh = siofAMlpoHvh + siofAMlpoHvh + siofAMlpoHvh, gPzLgKdLfZ = null, `message` = 1099438720, sdsrveQOoi = -2373351299779434224, hpxNweCm = false, PwyOmHforRKI = -2230544247357512169 
+| stats  VizcTbjMnBlQ = count(*), siofAMlpoHvh2 = median(sdsrveQOoi) + max(sdsrveQOoi), DUuKdsGgU = avg(PwyOmHforRKI) + count(PwyOmHforRKI), `gPzLgKdLfZ` = present(sdsrveQOoi + VizcTbjMnBlQ + PwyOmHforRKI), `siofAMlpoHvh` = count(sdsrveQOoi) + max(sdsrveQOoi) 
+| limit 61 
+| inline stats  gPzLgKdLfZ = sum(VizcTbjMnBlQ), `siofAMlpoHvh2` = count(*), zvgqSiVGqli = max(VizcTbjMnBlQ), WFjgBzYRXwa = median(VizcTbjMnBlQ) + count(VizcTbjMnBlQ) + avg(VizcTbjMnBlQ) 
+| keep `VizcTbjMnBlQ` 
+| inline stats  wXPJguff = median(VizcTbjMnBlQ) + avg(VizcTbjMnBlQ) + max(VizcTbjMnBlQ)
+;
+
+VizcTbjMnBlQ:long | wXPJguff:double
+1                 | 3.0       
+;
+
+
+inlineStatsAfterPruningAggregate8
+// https://github.com/elastic/elasticsearch/issues/135679
+required_capability: inline_stats_double_release_fix
+
+from dense_vector,airport_city_boundaries
+| eval  kMvNZVQSH = -4048759096317256968, GzAYNHaHuIS = 4262770411405329967, EwPpQFmOPoB = "lkdgnbTrBT", `abbrev` = "LiXI" 
+| keep `city`, `kMvNZVQSH`, GzAYNHaHuIS, city_boundary, EwPpQFmOPoB, `city_boundary`, `id`, kMvNZVQSH, id, airport 
+| limit 6888 
+| keep `kMvNZVQSH`, EwPpQFmOPoB, city, `id`, kMvNZVQSH 
+| stats  `EwPpQFmOPoB2` = min(id), `EwPpQFmOPoB3` = count(*), kZiJPhNZ = count(id), ErGhdKhdv = count_distinct(city), EwPpQFmOPoB = min(id) 
+| eval  ErGhdKhdv = null, fHSyxniFCS = -710891486, ErGhdKhdv = -1104488048, zdxdQdVIyFh = -1579062572, `kZiJPhNZ` = null, SzmOySyUh = null, eWVVRiwP = 1092261898855405071, nUxVhGhcpkiV = true, RvqGioBAAzYs = 169356242 
+| eval  `kZiJPhNZ` = "gWJHDTcDMjRVSnlzSX", lxLgaagX = 4700531997851493340, `kZiJPhNZ` = false 
+| keep `zdxdQdVIyFh` 
+| inline stats  `zdxdQdVIyFh` = count(*), WbWVJnYJ = count(*), zdxdQdVIyFh2 = max(zdxdQdVIyFh), uKjfeKTfaH = max(zdxdQdVIyFh), QpovVxjMWSjh = min(zdxdQdVIyFh)
+;
+
+zdxdQdVIyFh:long | WbWVJnYJ:long | zdxdQdVIyFh2:integer | uKjfeKTfaH:integer | QpovVxjMWSjh:integer
+1                | 1             | -1579062572          | -1579062572        | -1579062572
+;
+
+
+inlineStatsAfterPruningAggregate9
+required_capability: inline_stats_double_release_fix
+
+from employees
+| stats m = count(emp_no) 
+| eval c = 0
+| keep c 
+| inline stats c = count(*)
+;
+
+c:long
+1
+;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java
@@ -1521,6 +1521,11 @@ public enum Cap {
          */
         GROK_MULTI_PATTERN,
 
+        /**
+         * Fix double release in inline stats when LocalRelation is reused
+         */
+        INLINE_STATS_DOUBLE_RELEASE_FIX(INLINESTATS_V11.enabled)
+
         ;
 
         private final boolean enabled;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRowAsLocalRelation.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/ReplaceRowAsLocalRelation.java
@@ -12,8 +12,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;
@@ -30,6 +30,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(), new CopyingLocalSupplier(blocks.length == 0 ? new Page(0) : new Page(blocks)));
+        return new LocalRelation(row.source(), row.output(), LocalSupplier.of(blocks.length == 0 ? new Page(0) : new Page(blocks)));
     }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/InlineJoin.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/InlineJoin.java
@@ -24,6 +24,7 @@
 import org.elasticsearch.xpack.esql.plan.logical.Eval;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.UnaryPlan;
+import org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
 
 import java.io.IOException;
@@ -130,6 +131,12 @@ public record LogicalPlanTuple(LogicalPlan stubReplacedSubPlan, LogicalPlan orig
      * Finds the "first" (closest to the source command or bottom up in the tree) {@link InlineJoin}, replaces the {@link StubRelation}
      * of the right-hand side with left-hand side's source and returns a tuple.
      *
+     * <p>After extracting the subplan, the method ensures that all {@link LocalRelation} nodes use {@link CopyingLocalSupplier}
+     * to avoid double release when the same page is reused across executions. This guarantees correctness when substituting
+     * the subplan result back into the main plan.
+     *
+     * <p>Example transformation:
+     * <pre>
      * Original optimized plan:
      * Limit[1000[INTEGER],true]
      * \_InlineJoin[LEFT,[],[],[]]
@@ -146,33 +153,56 @@ public record LogicalPlanTuple(LogicalPlan stubReplacedSubPlan, LogicalPlan orig
      * Aggregate[[],[MAX(x{r}#99,true[BOOLEAN]) AS y#102]]
      * \_Limit[1000[INTEGER],false]
      *   \_LocalRelation[[x{r}#99],[IntVectorBlock[vector=ConstantIntVector[positions=1, value=1]]]]
+     * </pre>
+     *
+     * <p>In the end the method returns a tuple like this:
+     * <pre>
+     * LogicalPlanTuple[
+     *      stubReplacedSubPlan=
+     *                          Aggregate[[],[MAX(x{r}#99,true[BOOLEAN]) AS y#102]]
+     *                          \_LocalRelation[[x{r}#99],org.elasticsearch.xpack.esql.plan.logical.local.CopyingLocalSupplier@7c1]
+     *      originalSubPlan=
+     *                          Aggregate[[],[MAX(x{r}#99,true[BOOLEAN]) AS y#102]]
+     *                          \_StubRelation[[x{r}#99]]
+     *                  ]
+     * </pre>
      */
     public static LogicalPlanTuple firstSubPlan(LogicalPlan optimizedPlan, Set<LocalRelation> subPlansResults) {
-        Holder<LogicalPlanTuple> subPlan = new Holder<>();
+        final Holder<LogicalPlan> stubReplacedSubPlanHolder = new Holder<>();
+        final Holder<LogicalPlan> originalSubPlanHolder = new Holder<>();
         // 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) {
+            if (stubReplacedSubPlanHolder.get() == null) {
                 if (ij.right().anyMatch(p -> p instanceof StubRelation)) {
-                    var p = replaceStub(ij.left(), ij.right());
-                    p.setOptimized();
-                    subPlan.set(new LogicalPlanTuple(p, ij.right()));
+                    stubReplacedSubPlanHolder.set(replaceStub(ij.left(), ij.right()));
                 } else if (ij.right() instanceof LocalRelation relation
                     && (subPlansResults.isEmpty() || subPlansResults.contains(relation) == false)
                     || ij.right() instanceof LocalRelation == false && ij.right().anyMatch(p -> p instanceof LocalRelation)) {
                         // In case the plan was optimized further and the StubRelation was replaced with a LocalRelation
                         // or the right hand side became a LocalRelation alltogether, there is no need to replace the source of the
                         // right-hand side anymore.
-                        var p = ij.right();
-                        p.setOptimized();
-                        subPlan.set(new LogicalPlanTuple(p, ij.right()));
+                        stubReplacedSubPlanHolder.set(ij.right());
                         // TODO: INLINE STATS this is essentially an optimization similar to the one in PruneInlineJoinOnEmptyRightSide
                         // this further supports the idea of running the optimization step again after the substitutions (see EsqlSession
                         // executeSubPlan() method where we could run the optimizer after the results are replaced in place).
                     }
+                originalSubPlanHolder.set(ij.right());
             }
         });
-        return subPlan.get();
+        LogicalPlanTuple tuple = null;
+        var plan = stubReplacedSubPlanHolder.get();
+        if (plan != null) {
+            plan = plan.transformUp(LocalRelation.class, lr -> {
+                if (lr.supplier() instanceof CopyingLocalSupplier == false) {
+                    return new LocalRelation(lr.source(), lr.output(), new CopyingLocalSupplier(lr.supplier().get()));
+                }
+                return lr;
+            });
+            plan.setOptimized();
+            tuple = new LogicalPlanTuple(plan, originalSubPlanHolder.get());
+        }
+        return tuple;
     }
 
     public InlineJoin(Source source, LogicalPlan left, LogicalPlan right, JoinConfig config) {
