From de8e2a587bdecea0d05f211cc6e8af618c5119f7 Mon Sep 17 00:00:00 2001 From: Gal Lalouche Date: Fri, 13 Feb 2026 18:08:24 +0200 Subject: [PATCH 01/18] ESQL: Fix ExchangeSinkExec output type --- .../xpack/esql/planner/PlannerUtils.java | 12 +++++++++--- .../xpack/esql/plugin/ComputeService.java | 6 ++++++ .../esql/plugin/LateMaterializationPlanner.java | 13 ++++++++++--- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- ...educe_physical_optimization_data_driver.expected | 2 +- .../local_reduce_planned_data_driver.expected | 2 +- 23 files changed, 45 insertions(+), 26 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java index fc7fbaaf8a5ae..276b2aae51c9b 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java @@ -14,6 +14,7 @@ import org.elasticsearch.compute.data.BlockFactory; import org.elasticsearch.compute.data.ElementType; import org.elasticsearch.compute.operator.PlanTimeProfile; +import org.elasticsearch.core.Assertions; import org.elasticsearch.core.Nullable; import org.elasticsearch.core.Tuple; import org.elasticsearch.index.IndexMode; @@ -40,6 +41,7 @@ import org.elasticsearch.xpack.esql.optimizer.LocalLogicalPlanOptimizer; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalPlanOptimizer; +import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates; import org.elasticsearch.xpack.esql.plan.QueryPlan; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; @@ -226,8 +228,6 @@ public static PhysicalPlan localPlan( LocalPhysicalPlanOptimizer physicalOptimizer, PlanTimeProfile planTimeProfile ) { - // TODO add a test assertion for the consistency checker (after https://github.com/elastic/elasticsearch/issues/141654, see - // https://github.com/elastic/elasticsearch/pull/141082/changes#r2745334028); var isCoordPlan = new Holder<>(Boolean.TRUE); Set lookupJoinExecRightChildren = plan.collect(LookupJoinExec.class::isInstance) .stream() @@ -270,7 +270,13 @@ public static PhysicalPlan localPlan( }); PhysicalPlan resultPlan = isCoordPlan.get() ? plan : localPhysicalPlan; - + // This check is needed because in test code we sometimes invoke localPlan with a non-ExchangeSinkExec root. + if (resultPlan instanceof ExchangeSinkExec sink) { + resultPlan = new ExchangeSinkExec(sink.source(), sink.child().output(), sink.isIntermediateAgg(), sink.child()); + } + if (Assertions.ENABLED) { + PhysicalVerifier.LOCAL_INSTANCE.verify(resultPlan, resultPlan.output()); + } return resultPlan; } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 009b71b0a779e..bb1b58f0def6a 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -30,6 +30,7 @@ import org.elasticsearch.compute.operator.exchange.ExchangeSink; import org.elasticsearch.compute.operator.exchange.ExchangeSinkHandler; import org.elasticsearch.compute.operator.exchange.ExchangeSourceHandler; +import org.elasticsearch.core.Assertions; import org.elasticsearch.core.RefCounted; import org.elasticsearch.core.Releasable; import org.elasticsearch.core.Releasables; @@ -57,6 +58,7 @@ import org.elasticsearch.xpack.esql.enrich.LookupFromIndexService; import org.elasticsearch.xpack.esql.inference.InferenceService; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; +import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; @@ -823,6 +825,10 @@ public static ReductionPlan reductionPlan( if (planTimeProfile != null) { planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); } + if (Assertions.ENABLED) { + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), reductionPlan.nodeReducePlan().child().output()); + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionPlan.dataNodePlan().child().output()); + } return reductionPlan; } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java index 665d30381de0d..a6f0f48087e31 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java @@ -128,8 +128,12 @@ public static Optional planReduceDriverTopN( } var updatedFragment = new Project(Source.EMPTY, withAddedDocToRelation, expectedDataOutput); FragmentExec updatedFragmentExec = fragmentExec.withFragment(updatedFragment); - // TODO This ignores the possible change in output, see #141654 - ExchangeSinkExec updatedDataPlan = originalPlan.replaceChild(updatedFragmentExec); + ExchangeSinkExec updatedDataPlan = new ExchangeSinkExec( + originalPlan.source(), + expectedDataOutput, + originalPlan.isIntermediateAgg(), + updatedFragmentExec + ); // Replace the TopN child with the data driver as the source. PhysicalPlan reductionPlan = toPhysical(fragmentExec.fragment(), context).transformDown(TopNExec.class, t -> { @@ -138,7 +142,10 @@ public static Optional planReduceDriverTopN( boolean fragmentIsSorted = updatedFragment.child() instanceof TopN; return fragmentIsSorted ? t.replaceChild(exchangeExec).withSortedInput() : t.replaceChild(exchangeExec); }); - ExchangeSinkExec reductionPlanWithSize = originalPlan.replaceChild( + ExchangeSinkExec reductionPlanWithSize = new ExchangeSinkExec( + originalPlan.source(), + reductionPlan.output(), + originalPlan.isIntermediateAgg(), EstimatesRowSize.estimateRowSize(updatedFragmentExec.estimatedRowSize(), reductionPlan) ); diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_physical_optimization_data_driver.expected index 79d08e098a4ac..723ce904e88d3 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_ProjectExec[[_doc{f}, hire_date{f}]] \_FieldExtractExec[hire_date{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=hire_date{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[24] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_planned_data_driver.expected index 42c2f1deccb35..391e45f5b0835 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testBasicTopNLateMaterialization/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, hire_date{f}]] \_TopN[[Order[hire_date{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_physical_optimization_data_driver.expected index 6c0cb294fc861..9bd1091990166 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[hire_date{f}, $$order_by{r}],false] +ExchangeSinkExec[[_doc{f}, $$order_by{r}],false] \_ProjectExec[[_doc{f}, $$order_by{r}]] \_TopNExec[[Order[$$order_by{r},ASC,LAST]],20[INTEGER],36] \_EvalExec[[SIN(height{f}) * 2[INTEGER] AS $$order_by]] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_planned_data_driver.expected index 2c2c810fdcd5f..edd351fd4feb2 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepAfterSort/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[hire_date{f}, $$order_by{r}],false] +ExchangeSinkExec[[_doc{f}, $$order_by{r}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, $$order_by{r}]] \_TopN[[Order[$$order_by{r},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_physical_optimization_data_driver.expected index abec74e514128..18dffd6ba00d3 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[height{f}, hire_date{f}, $$order_by{r}],false] +ExchangeSinkExec[[_doc{f}, height{f}, $$order_by{r}],false] \_ProjectExec[[_doc{f}, height{f}, $$order_by{r}]] \_TopNExec[[Order[$$order_by{r},ASC,LAST]],20[INTEGER],36] \_EvalExec[[SIN(height{f}) * 2[INTEGER] AS $$order_by]] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_planned_data_driver.expected index d6d06ec07d272..450d2bf335b05 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testExpressionSortTopNKeepBeforeSort/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[height{f}, hire_date{f}, $$order_by{r}],false] +ExchangeSinkExec[[_doc{f}, height{f}, $$order_by{r}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, height{f}, $$order_by{r}]] \_TopN[[Order[$$order_by{r},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_physical_optimization_data_driver.expected index afa8bc120313d..16144b0f1de0f 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[avg_worked_seconds{f}, birth_date{f}, emp_no{f}, first_name{f}, gender{f}, height{f}, height.float{f}, height.half_float{f}, height.scaled_float{f}, hire_date{f}, is_rehired{f}, job_positions{f}, languages{f}, languages.byte{f}, languages.long{f}, languages.short{f}, last_name{f}, salary{f}, salary_change{f}, salary_change.int{f}, salary_change.keyword{f}, salary_change.long{f}, still_hired{f}, language_code{r}, language_name{f}],false] +ExchangeSinkExec[[_doc{f}, emp_no{f}, languages{f}, language_code{r}, language_name{f}],false] \_ProjectExec[[_doc{f}, emp_no{f}, languages{f}, language_code{r}, language_name{f}]] \_TopNExec[[Order[emp_no{f},ASC,LAST]],20[INTEGER],82] \_FieldExtractExec[emp_no{f}]<[],[]> diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_planned_data_driver.expected index ef61e922e587d..d8621a842a5f3 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testLookupJoinOnDataNode/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[avg_worked_seconds{f}, birth_date{f}, emp_no{f}, first_name{f}, gender{f}, height{f}, height.float{f}, height.half_float{f}, height.scaled_float{f}, hire_date{f}, is_rehired{f}, job_positions{f}, languages{f}, languages.byte{f}, languages.long{f}, languages.short{f}, last_name{f}, salary{f}, salary_change{f}, salary_change.int{f}, salary_change.keyword{f}, salary_change.long{f}, still_hired{f}, language_code{r}, language_name{f}],false] +ExchangeSinkExec[[_doc{f}, emp_no{f}, languages{f}, language_code{r}, language_name{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, emp_no{f}, languages{f}, language_code{r}, language_name{f}]] \_TopN[[Order[emp_no{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_physical_optimization_data_driver.expected index 4ba472a701132..8d71c6318ff7d 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, height{f}, hire_date{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}, height{f}],false] \_ProjectExec[[_doc{f}, hire_date{f}, height{f}]] \_FieldExtractExec[hire_date{f}, height{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=hire_date{f}, direction=ASC, nulls=LAST], FieldSort[field=height{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[32] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_planned_data_driver.expected index 8ab7e96ac70e1..5ecfdc0e91ea9 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleFieldSortTopN/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, height{f}, hire_date{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}, height{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, hire_date{f}, height{f}]] \_TopN[[Order[hire_date{f},ASC,LAST], Order[height{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_physical_optimization_data_driver.expected index 79d08e098a4ac..723ce904e88d3 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_ProjectExec[[_doc{f}, hire_date{f}]] \_FieldExtractExec[hire_date{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=hire_date{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[24] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_planned_data_driver.expected index 42c2f1deccb35..391e45f5b0835 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testMultipleTopN/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, hire_date{f}]] \_TopN[[Order[hire_date{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_physical_optimization_data_driver.expected index b49e189701890..b3b5c1a34e4bc 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, height{f}],false] +ExchangeSinkExec[[_doc{f}, height{f}],false] \_ProjectExec[[_doc{f}, height{f}]] \_FieldExtractExec[height{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=height{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[24] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_planned_data_driver.expected index bf2fa9c12991a..99ac7bd279208 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testPushedDownTopN/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, height{f}],false] +ExchangeSinkExec[[_doc{f}, height{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, height{f}]] \_TopN[[Order[height{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_physical_optimization_data_driver.expected index 7a2b16378c969..81d5401085264 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, salary{f}, hire_date{f}],false] \_ProjectExec[[_doc{f}, salary{f}, hire_date{f}]] \_FieldExtractExec[salary{f}, hire_date{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=hire_date{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[28] queryBuilderAndTags [[QueryBuilderAndTags[query={ diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_planned_data_driver.expected index 041b22266a09e..435ccfdcf08e0 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testSomeFieldsNeededBeforeLateMaterialization/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, salary{f}, hire_date{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, salary{f}, hire_date{f}]] \_TopN[[Order[hire_date{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_physical_optimization_data_driver.expected index 6d737b8a8eb59..723ce904e88d3 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_ProjectExec[[_doc{f}, hire_date{f}]] \_FieldExtractExec[hire_date{f}]<[],[]> \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[[FieldSort[field=hire_date{f}, direction=ASC, nulls=LAST]]] estimatedRowSize[24] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_planned_data_driver.expected index 5cc08a59070fd..391e45f5b0835 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNThenStats/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, hire_date{f}]] \_TopN[[Order[hire_date{f},ASC,LAST]],20[INTEGER],false] diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected index 86be6bee3c591..d809e325e114a 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{r}],false] \_ProjectExec[[_doc{f}, birth_date{r} AS hire_date]] \_EvalExec[[null[DATETIME] AS birth_date]] \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[] estimatedRowSize[12] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_planned_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_planned_data_driver.expected index 42c2f1deccb35..391e45f5b0835 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_planned_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_planned_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[emp_no{f}, hire_date{f}, salary{f}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_FragmentExec[filter=null, estimatedRowSize=0, reducer=[], fragment=[<> Project[[_doc{f}, hire_date{f}]] \_TopN[[Order[hire_date{f},ASC,LAST]],20[INTEGER],false] From 3942dd014751d595cac3b2f3620aec498f3cf5fe Mon Sep 17 00:00:00 2001 From: Gal Lalouche Date: Fri, 13 Feb 2026 18:40:49 +0200 Subject: [PATCH 02/18] Fix SpecIT failing --- .../xpack/esql/plan/physical/ExchangeSinkExec.java | 5 +++++ .../org/elasticsearch/xpack/esql/planner/PlannerUtils.java | 2 +- .../org/elasticsearch/xpack/esql/plugin/ComputeService.java | 5 +++++ 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ExchangeSinkExec.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ExchangeSinkExec.java index 940471484c3e7..21f4464c2470a 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ExchangeSinkExec.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/ExchangeSinkExec.java @@ -77,6 +77,11 @@ public ExchangeSinkExec replaceChild(PhysicalPlan newChild) { return new ExchangeSinkExec(source(), output, intermediateAgg, newChild); } + /** Like {@link #replaceChild(PhysicalPlan)} but sets the output to the new child's output. */ + public ExchangeSinkExec replaceChildAndUpdateOutput(PhysicalPlan newChild) { + return new ExchangeSinkExec(source(), newChild.output(), intermediateAgg, newChild); + } + @Override public boolean equals(Object o) { if (super.equals(o)) { diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java index 276b2aae51c9b..5aaadfe750384 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java @@ -272,7 +272,7 @@ public static PhysicalPlan localPlan( PhysicalPlan resultPlan = isCoordPlan.get() ? plan : localPhysicalPlan; // This check is needed because in test code we sometimes invoke localPlan with a non-ExchangeSinkExec root. if (resultPlan instanceof ExchangeSinkExec sink) { - resultPlan = new ExchangeSinkExec(sink.source(), sink.child().output(), sink.isIntermediateAgg(), sink.child()); + resultPlan = sink.replaceChildAndUpdateOutput(sink.child()); } if (Assertions.ENABLED) { PhysicalVerifier.LOCAL_INSTANCE.verify(resultPlan, resultPlan.output()); diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index bb1b58f0def6a..685e544d3b0ef 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -825,6 +825,11 @@ public static ReductionPlan reductionPlan( if (planTimeProfile != null) { planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); } + reductionPlan = new ReductionPlan( + reductionPlan.nodeReducePlan().replaceChildAndUpdateOutput(reductionPlan.nodeReducePlan().child()), + reductionPlan.dataNodePlan().replaceChildAndUpdateOutput(reductionPlan.dataNodePlan().child()), + reductionPlan.localPhysicalOptimization() + ); if (Assertions.ENABLED) { PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), reductionPlan.nodeReducePlan().child().output()); PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionPlan.dataNodePlan().child().output()); From befbc1aafcb848c9f8dbb6419ab8cf54b3c57be2 Mon Sep 17 00:00:00 2001 From: Gal Lalouche Date: Fri, 13 Feb 2026 20:35:25 +0200 Subject: [PATCH 03/18] Fix source input --- .../xpack/esql/plugin/ComputeService.java | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 685e544d3b0ef..fa1c07f7529d2 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -819,7 +819,18 @@ public static ReductionPlan reductionPlan( // Fallback to the behavior listed below, i.e., a regular top n reduction without loading new fields. .orElseGet(() -> runNodeLevelReduction ? placePlanBetweenExchanges.apply(topN.plan()) : defaultResult); case PlannerUtils.TopNReduction topN when runNodeLevelReduction -> placePlanBetweenExchanges.apply(topN.plan()); - case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> placePlanBetweenExchanges.apply(rp.plan()); + case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> { + PhysicalPlan reductionSource = new ExchangeSourceExec( + originalPlan.source(), + new ArrayList<>(rp.plan().references()), + originalPlan.isIntermediateAgg() + ); + yield new ReductionPlan( + originalPlan.replaceChild(rp.plan().replaceChildren(List.of(reductionSource))), + originalPlan, + LocalPhysicalOptimization.ENABLED + ); + } default -> defaultResult; }; if (planTimeProfile != null) { From 6ed1ae0cb190641f963cc7b8eb43038e4fc3c48f Mon Sep 17 00:00:00 2001 From: Gal Lalouche Date: Fri, 13 Feb 2026 20:35:25 +0200 Subject: [PATCH 04/18] Fix source input --- .../esql/plugin/LateMaterializationPlanner.java | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java index a6f0f48087e31..734a3c991dd6e 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java @@ -128,12 +128,7 @@ public static Optional planReduceDriverTopN( } var updatedFragment = new Project(Source.EMPTY, withAddedDocToRelation, expectedDataOutput); FragmentExec updatedFragmentExec = fragmentExec.withFragment(updatedFragment); - ExchangeSinkExec updatedDataPlan = new ExchangeSinkExec( - originalPlan.source(), - expectedDataOutput, - originalPlan.isIntermediateAgg(), - updatedFragmentExec - ); + ExchangeSinkExec updatedDataPlan = originalPlan.replaceChildAndUpdateOutput(updatedFragmentExec); // Replace the TopN child with the data driver as the source. PhysicalPlan reductionPlan = toPhysical(fragmentExec.fragment(), context).transformDown(TopNExec.class, t -> { @@ -142,12 +137,8 @@ public static Optional planReduceDriverTopN( boolean fragmentIsSorted = updatedFragment.child() instanceof TopN; return fragmentIsSorted ? t.replaceChild(exchangeExec).withSortedInput() : t.replaceChild(exchangeExec); }); - ExchangeSinkExec reductionPlanWithSize = new ExchangeSinkExec( - originalPlan.source(), - reductionPlan.output(), - originalPlan.isIntermediateAgg(), - EstimatesRowSize.estimateRowSize(updatedFragmentExec.estimatedRowSize(), reductionPlan) - ); + PhysicalPlan sizedReductionPlan = EstimatesRowSize.estimateRowSize(updatedFragmentExec.estimatedRowSize(), reductionPlan); + ExchangeSinkExec reductionPlanWithSize = originalPlan.replaceChildAndUpdateOutput(sizedReductionPlan); // The TopN reduction plan should not be further optimized locally on the node reduce driver, since we took great pains to // preplan in advance, including all the necessary field extractions! From 3ce094a0588a1cabe5c0c67fb07a21ff836fe056 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 18:59:51 +0100 Subject: [PATCH 05/18] Fix testTopNWithMissingSortField --- .../plugin/LateMaterializationPlannerGoldenTests.java | 10 ++++++++-- ...l_reduce_physical_optimization_data_driver.expected | 4 ++-- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java index 8c0f139aadc7a..7af74c57d3f37 100644 --- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java +++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java @@ -152,12 +152,18 @@ private static EsqlTestUtils.TestSearchStats missingFieldStats(String missingFie return new EsqlTestUtils.TestSearchStats() { @Override public boolean exists(FieldAttribute.FieldName field) { - return false; + if (field.string().equals(missingField)) { + return false; + } + return true; } @Override public boolean isIndexed(FieldAttribute.FieldName field) { - return false; + if (field.string().equals(missingField)) { + return false; + } + return true; } }; } diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected index d809e325e114a..218985eb5ae40 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ ExchangeSinkExec[[_doc{f}, hire_date{r}],false] -\_ProjectExec[[_doc{f}, birth_date{r} AS hire_date]] - \_EvalExec[[null[DATETIME] AS birth_date]] +\_ProjectExec[[_doc{f}, hire_date{r}]] + \_EvalExec[[null[DATETIME] AS hire_date]] \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[] estimatedRowSize[12] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file From edfe40227b74dd0fcb3f043292e679d6a2eb0657 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 19:31:52 +0100 Subject: [PATCH 06/18] Move consistency check to top of localPlan --- .../xpack/esql/planner/PlannerUtils.java | 12 +++++------- ...reduce_physical_optimization_data_driver.expected | 2 +- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java index 0e3c1a1271ed9..298b602084757 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java @@ -239,6 +239,11 @@ public static PhysicalPlan localPlan( LocalPhysicalPlanOptimizer physicalOptimizer, PlanTimeProfile planTimeProfile ) { + if (Assertions.ENABLED) { + // The data node plan can be changed significantly due to late materialization on the data node; the exchange sink exec + // can end up being inconsistent with the actual plan. Let's check this early. + PhysicalVerifier.LOCAL_INSTANCE.verify(plan, plan.output()); + } var isCoordPlan = new Holder<>(Boolean.TRUE); Set lookupJoinExecRightChildren = plan.collect(LookupJoinExec.class::isInstance) .stream() @@ -281,13 +286,6 @@ public static PhysicalPlan localPlan( }); PhysicalPlan resultPlan = isCoordPlan.get() ? plan : localPhysicalPlan; - // This check is needed because in test code we sometimes invoke localPlan with a non-ExchangeSinkExec root. - if (resultPlan instanceof ExchangeSinkExec sink) { - resultPlan = sink.replaceChildAndUpdateOutput(sink.child()); - } - if (Assertions.ENABLED) { - PhysicalVerifier.LOCAL_INSTANCE.verify(resultPlan, resultPlan.output()); - } return resultPlan; } diff --git a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected index 218985eb5ae40..5aa4ef96b07c0 100644 --- a/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected +++ b/x-pack/plugin/esql/src/test/resources/org/elasticsearch/xpack/esql/plugin/golden_tests/LateMaterializationPlannerGoldenTests/testTopNWithMissingSortField/local_reduce_physical_optimization_data_driver.expected @@ -1,4 +1,4 @@ -ExchangeSinkExec[[_doc{f}, hire_date{r}],false] +ExchangeSinkExec[[_doc{f}, hire_date{f}],false] \_ProjectExec[[_doc{f}, hire_date{r}]] \_EvalExec[[null[DATETIME] AS hire_date]] \_EsQueryExec[employees], indexMode[standard], [_doc{f}], limit[20], sort[] estimatedRowSize[12] queryBuilderAndTags [[QueryBuilderAndTags[query=null, tags=[]]]] \ No newline at end of file From 41137ef14956fd8f70a1457135a58067a43ecf18 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 20:17:54 +0100 Subject: [PATCH 07/18] Revert LateMaterializationPlanner changes The exchange sinks are already being updated in ComputeService.java. Let's keep the exchange sink update constrained to one place. --- .../xpack/esql/plugin/LateMaterializationPlanner.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java index 734a3c991dd6e..38c8bf3584695 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java @@ -128,7 +128,7 @@ public static Optional planReduceDriverTopN( } var updatedFragment = new Project(Source.EMPTY, withAddedDocToRelation, expectedDataOutput); FragmentExec updatedFragmentExec = fragmentExec.withFragment(updatedFragment); - ExchangeSinkExec updatedDataPlan = originalPlan.replaceChildAndUpdateOutput(updatedFragmentExec); + ExchangeSinkExec updatedDataPlan = originalPlan.replaceChild(updatedFragmentExec); // Replace the TopN child with the data driver as the source. PhysicalPlan reductionPlan = toPhysical(fragmentExec.fragment(), context).transformDown(TopNExec.class, t -> { @@ -138,7 +138,7 @@ public static Optional planReduceDriverTopN( return fragmentIsSorted ? t.replaceChild(exchangeExec).withSortedInput() : t.replaceChild(exchangeExec); }); PhysicalPlan sizedReductionPlan = EstimatesRowSize.estimateRowSize(updatedFragmentExec.estimatedRowSize(), reductionPlan); - ExchangeSinkExec reductionPlanWithSize = originalPlan.replaceChildAndUpdateOutput(sizedReductionPlan); + ExchangeSinkExec reductionPlanWithSize = originalPlan.replaceChild(sizedReductionPlan); // The TopN reduction plan should not be further optimized locally on the node reduce driver, since we took great pains to // preplan in advance, including all the necessary field extractions! From 5f5596bb7cbbad695edee3bbe4d3726ae7ddb755 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 20:18:45 +0100 Subject: [PATCH 08/18] Remove added assertion in PlannerUtils#localPlan The consistency checks in PlannerUtils#reductionPlan are already checking that late materialization planning didn't mess up the data and reduce plans' consistency. --- .../org/elasticsearch/xpack/esql/planner/PlannerUtils.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java index 298b602084757..f55e3c98850c6 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java @@ -239,11 +239,6 @@ public static PhysicalPlan localPlan( LocalPhysicalPlanOptimizer physicalOptimizer, PlanTimeProfile planTimeProfile ) { - if (Assertions.ENABLED) { - // The data node plan can be changed significantly due to late materialization on the data node; the exchange sink exec - // can end up being inconsistent with the actual plan. Let's check this early. - PhysicalVerifier.LOCAL_INSTANCE.verify(plan, plan.output()); - } var isCoordPlan = new Holder<>(Boolean.TRUE); Set lookupJoinExecRightChildren = plan.collect(LookupJoinExec.class::isInstance) .stream() From ccc33df7942ea917d38385fd43ef9feb92446707 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 20:46:55 +0100 Subject: [PATCH 09/18] Simplify and harden ComputeService#reductionPlan --- .../xpack/esql/planner/PlannerUtils.java | 2 - .../xpack/esql/plugin/ComputeService.java | 49 ++++++++++++------- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java index f55e3c98850c6..12f2c72f47b7c 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java @@ -14,7 +14,6 @@ import org.elasticsearch.compute.data.BlockFactory; import org.elasticsearch.compute.data.ElementType; import org.elasticsearch.compute.operator.PlanTimeProfile; -import org.elasticsearch.core.Assertions; import org.elasticsearch.core.Nullable; import org.elasticsearch.core.Tuple; import org.elasticsearch.index.IndexMode; @@ -41,7 +40,6 @@ import org.elasticsearch.xpack.esql.optimizer.LocalLogicalPlanOptimizer; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalPlanOptimizer; -import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates; import org.elasticsearch.xpack.esql.plan.QueryPlan; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 39dcf0e743e9c..d30bd740d22b6 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -61,6 +61,7 @@ import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; +import org.elasticsearch.xpack.esql.plan.physical.AggregateExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; import org.elasticsearch.xpack.esql.plan.physical.OutputExec; @@ -801,9 +802,14 @@ public static ReductionPlan reductionPlan( ) { long startTime = planTimeProfile == null ? 0 : System.nanoTime(); PhysicalPlan source = new ExchangeSourceExec(originalPlan.source(), originalPlan.output(), originalPlan.isIntermediateAgg()); - ReductionPlan defaultResult = new ReductionPlan(originalPlan.replaceChild(source), originalPlan, LocalPhysicalOptimization.ENABLED); + // Just send out everything through a single exchange as a fallback + ReductionPlan passThroughReduction = new ReductionPlan( + originalPlan.replaceChild(source), + originalPlan, + LocalPhysicalOptimization.ENABLED + ); if (reduceNodeLateMaterialization == false && runNodeLevelReduction == false) { - return defaultResult; + return passThroughReduction; } Function placePlanBetweenExchanges = p -> new ReductionPlan( @@ -811,6 +817,21 @@ public static ReductionPlan reductionPlan( originalPlan, LocalPhysicalOptimization.ENABLED ); + Function placeAggBetweenExchanges = p -> { + PhysicalPlan reductionSource = new ExchangeSourceExec( + originalPlan.source(), + // For an agg, the data that's sent between exchanges is different because the reduction driver gets intermediate + // attributes. + new ArrayList<>(p.references()), + originalPlan.isIntermediateAgg() + ); + return new ReductionPlan( + originalPlan.replaceChild(p.replaceChildren(List.of(reductionSource))), + originalPlan, + LocalPhysicalOptimization.ENABLED + ); + }; + // The default plan is just the exchange source piped directly into the exchange sink. ReductionPlan reductionPlan = switch (PlannerUtils.reductionPlan(originalPlan)) { case PlannerUtils.TopNReduction topN when reduceNodeLateMaterialization -> @@ -822,21 +843,13 @@ public static ReductionPlan reductionPlan( originalPlan ) // Fallback to the behavior listed below, i.e., a regular top n reduction without loading new fields. - .orElseGet(() -> runNodeLevelReduction ? placePlanBetweenExchanges.apply(topN.plan()) : defaultResult); + .orElseGet(() -> runNodeLevelReduction ? placePlanBetweenExchanges.apply(topN.plan()) : passThroughReduction); case PlannerUtils.TopNReduction topN when runNodeLevelReduction -> placePlanBetweenExchanges.apply(topN.plan()); - case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> { - PhysicalPlan reductionSource = new ExchangeSourceExec( - originalPlan.source(), - new ArrayList<>(rp.plan().references()), - originalPlan.isIntermediateAgg() - ); - yield new ReductionPlan( - originalPlan.replaceChild(rp.plan().replaceChildren(List.of(reductionSource))), - originalPlan, - LocalPhysicalOptimization.ENABLED - ); - } - default -> defaultResult; + // Not a TopN - must be an agg or a limit + case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> rp.plan() instanceof AggregateExec agg + ? placeAggBetweenExchanges.apply(agg) + : placePlanBetweenExchanges.apply(rp.plan()); + default -> passThroughReduction; }; if (planTimeProfile != null) { planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); @@ -848,7 +861,9 @@ yield new ReductionPlan( ); if (Assertions.ENABLED) { PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), reductionPlan.nodeReducePlan().child().output()); - PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionPlan.dataNodePlan().child().output()); + ExchangeSourceExec reductionSource = (ExchangeSourceExec) reductionPlan.nodeReducePlan().collectLeaves().getFirst(); + // The data driver's output is sent to the reduction driver, so the outputs must match up. + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionSource.output()); } return reductionPlan; } From 4a9a4f14f24ac4c7f995bdc7fa9e9883236be010 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 21:14:10 +0100 Subject: [PATCH 10/18] Simplify some more --- .../xpack/esql/plugin/ComputeService.java | 19 +------------------ 1 file changed, 1 insertion(+), 18 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index d30bd740d22b6..b58eb728c6b63 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -61,7 +61,6 @@ import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; -import org.elasticsearch.xpack.esql.plan.physical.AggregateExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; import org.elasticsearch.xpack.esql.plan.physical.OutputExec; @@ -817,20 +816,6 @@ public static ReductionPlan reductionPlan( originalPlan, LocalPhysicalOptimization.ENABLED ); - Function placeAggBetweenExchanges = p -> { - PhysicalPlan reductionSource = new ExchangeSourceExec( - originalPlan.source(), - // For an agg, the data that's sent between exchanges is different because the reduction driver gets intermediate - // attributes. - new ArrayList<>(p.references()), - originalPlan.isIntermediateAgg() - ); - return new ReductionPlan( - originalPlan.replaceChild(p.replaceChildren(List.of(reductionSource))), - originalPlan, - LocalPhysicalOptimization.ENABLED - ); - }; // The default plan is just the exchange source piped directly into the exchange sink. ReductionPlan reductionPlan = switch (PlannerUtils.reductionPlan(originalPlan)) { @@ -846,9 +831,7 @@ public static ReductionPlan reductionPlan( .orElseGet(() -> runNodeLevelReduction ? placePlanBetweenExchanges.apply(topN.plan()) : passThroughReduction); case PlannerUtils.TopNReduction topN when runNodeLevelReduction -> placePlanBetweenExchanges.apply(topN.plan()); // Not a TopN - must be an agg or a limit - case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> rp.plan() instanceof AggregateExec agg - ? placeAggBetweenExchanges.apply(agg) - : placePlanBetweenExchanges.apply(rp.plan()); + case PlannerUtils.ReducedPlan rp when runNodeLevelReduction -> placePlanBetweenExchanges.apply(rp.plan()); default -> passThroughReduction; }; if (planTimeProfile != null) { From f139867f3da7b287bb2e46d1304d230fb16459f5 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 22:21:00 +0100 Subject: [PATCH 11/18] Fix FragmentExec#output --- .../elasticsearch/xpack/esql/plan/physical/FragmentExec.java | 5 +++++ .../elasticsearch/xpack/esql/planner/mapper/MapperUtils.java | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java index c1a642817fad7..1fa59630140a3 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java @@ -15,7 +15,9 @@ import org.elasticsearch.xpack.esql.core.tree.NodeInfo; import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; +import org.elasticsearch.xpack.esql.plan.logical.Aggregate; import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan; +import org.elasticsearch.xpack.esql.planner.mapper.MapperUtils; import java.io.IOException; import java.util.List; @@ -87,6 +89,9 @@ protected NodeInfo info() { @Override public List output() { + if (fragment instanceof Aggregate agg) { + return MapperUtils.intermediateAttributes(agg); + } return fragment.output(); } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java index 209bdfc377ba7..eafe8badf13d5 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java @@ -190,7 +190,7 @@ static PhysicalPlan mapUnary(UnaryPlan p, PhysicalPlan child) { return unsupported(p); } - static List intermediateAttributes(Aggregate aggregate) { + public static List intermediateAttributes(Aggregate aggregate) { List intermediateAttributes = AbstractPhysicalOperationProviders.intermediateAttributes( aggregate.aggregates(), aggregate.groupings() From aea8e5f9ab1d633a53b6159e9f82c55383b8dde1 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 22:27:58 +0100 Subject: [PATCH 12/18] Do not update reduce plan output That actually genuinely needs to stay untouched. --- .../org/elasticsearch/xpack/esql/plugin/ComputeService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index b58eb728c6b63..95cea67641ac1 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -838,7 +838,7 @@ public static ReductionPlan reductionPlan( planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); } reductionPlan = new ReductionPlan( - reductionPlan.nodeReducePlan().replaceChildAndUpdateOutput(reductionPlan.nodeReducePlan().child()), + reductionPlan.nodeReducePlan(), reductionPlan.dataNodePlan().replaceChildAndUpdateOutput(reductionPlan.dataNodePlan().child()), reductionPlan.localPhysicalOptimization() ); From 747f5a389f6ae208b112eec66553c23fa61106d8 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 22:40:57 +0100 Subject: [PATCH 13/18] Revert "Fix FragmentExec#output" This reverts commit f139867f3da7b287bb2e46d1304d230fb16459f5. --- .../elasticsearch/xpack/esql/plan/physical/FragmentExec.java | 5 ----- .../elasticsearch/xpack/esql/planner/mapper/MapperUtils.java | 2 +- 2 files changed, 1 insertion(+), 6 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java index 1fa59630140a3..c1a642817fad7 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/physical/FragmentExec.java @@ -15,9 +15,7 @@ import org.elasticsearch.xpack.esql.core.tree.NodeInfo; import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; -import org.elasticsearch.xpack.esql.plan.logical.Aggregate; import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan; -import org.elasticsearch.xpack.esql.planner.mapper.MapperUtils; import java.io.IOException; import java.util.List; @@ -89,9 +87,6 @@ protected NodeInfo info() { @Override public List output() { - if (fragment instanceof Aggregate agg) { - return MapperUtils.intermediateAttributes(agg); - } return fragment.output(); } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java index eafe8badf13d5..209bdfc377ba7 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/MapperUtils.java @@ -190,7 +190,7 @@ static PhysicalPlan mapUnary(UnaryPlan p, PhysicalPlan child) { return unsupported(p); } - public static List intermediateAttributes(Aggregate aggregate) { + static List intermediateAttributes(Aggregate aggregate) { List intermediateAttributes = AbstractPhysicalOperationProviders.intermediateAttributes( aggregate.aggregates(), aggregate.groupings() From a5c9b37c055b4c9d4cd671301c9e0bfecfedc0b9 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 23:01:20 +0100 Subject: [PATCH 14/18] Reduce scope Only fix late materialization planning. Only check consistency for non-agg plans; aggs remain inconsistent for now. --- .../xpack/esql/plugin/ComputeService.java | 14 +++++++------- .../esql/plugin/LateMaterializationPlanner.java | 4 +++- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 95cea67641ac1..681f7727152d9 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -61,6 +61,7 @@ import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; +import org.elasticsearch.xpack.esql.plan.physical.AggregateExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; import org.elasticsearch.xpack.esql.plan.physical.OutputExec; @@ -837,13 +838,12 @@ public static ReductionPlan reductionPlan( if (planTimeProfile != null) { planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); } - reductionPlan = new ReductionPlan( - reductionPlan.nodeReducePlan(), - reductionPlan.dataNodePlan().replaceChildAndUpdateOutput(reductionPlan.dataNodePlan().child()), - reductionPlan.localPhysicalOptimization() - ); - if (Assertions.ENABLED) { - PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), reductionPlan.nodeReducePlan().child().output()); + + // TODO: How we generate intermediate attributes prevents us from cleanly checking dependencies here. + // FragmentExec.output() doesn't take into account intermediate attributes of aggs, and time series aggs + // have some peculiarities due to implicit dimensions. We should clean this up and add a proper check here. + if (Assertions.ENABLED && reductionPlan.nodeReducePlan().child() instanceof AggregateExec == false) { + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), originalPlan.output()); ExchangeSourceExec reductionSource = (ExchangeSourceExec) reductionPlan.nodeReducePlan().collectLeaves().getFirst(); // The data driver's output is sent to the reduction driver, so the outputs must match up. PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionSource.output()); diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java index 38c8bf3584695..bf9f619a37d09 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java @@ -7,6 +7,7 @@ package org.elasticsearch.xpack.esql.plugin; +import org.elasticsearch.core.Assertions; import org.elasticsearch.index.IndexMode; import org.elasticsearch.xpack.esql.core.expression.Attribute; import org.elasticsearch.xpack.esql.core.expression.AttributeSet; @@ -14,6 +15,7 @@ import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.core.util.CollectionUtils; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; +import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.InsertFieldExtraction; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushTopNToSource; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.ReplaceSourceAttributes; @@ -128,7 +130,7 @@ public static Optional planReduceDriverTopN( } var updatedFragment = new Project(Source.EMPTY, withAddedDocToRelation, expectedDataOutput); FragmentExec updatedFragmentExec = fragmentExec.withFragment(updatedFragment); - ExchangeSinkExec updatedDataPlan = originalPlan.replaceChild(updatedFragmentExec); + ExchangeSinkExec updatedDataPlan = originalPlan.replaceChildAndUpdateOutput(updatedFragmentExec); // Replace the TopN child with the data driver as the source. PhysicalPlan reductionPlan = toPhysical(fragmentExec.fragment(), context).transformDown(TopNExec.class, t -> { From 27ac3a2b2b9624c74c57f76cfd96e291e0d3af23 Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Mon, 23 Feb 2026 22:08:54 +0000 Subject: [PATCH 15/18] [CI] Auto commit changes from spotless --- .../xpack/esql/plugin/LateMaterializationPlanner.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java index bf9f619a37d09..4f24528b6692c 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlanner.java @@ -7,7 +7,6 @@ package org.elasticsearch.xpack.esql.plugin; -import org.elasticsearch.core.Assertions; import org.elasticsearch.index.IndexMode; import org.elasticsearch.xpack.esql.core.expression.Attribute; import org.elasticsearch.xpack.esql.core.expression.AttributeSet; @@ -15,7 +14,6 @@ import org.elasticsearch.xpack.esql.core.tree.Source; import org.elasticsearch.xpack.esql.core.util.CollectionUtils; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; -import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.InsertFieldExtraction; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.PushTopNToSource; import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.ReplaceSourceAttributes; From 34bf0b54eb7a6ab287f15c6a1b17938f420c4085 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Mon, 23 Feb 2026 23:15:27 +0100 Subject: [PATCH 16/18] Also exclude TimeSeriesAggregates --- .../xpack/esql/plugin/ComputeService.java | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 681f7727152d9..866d2e90b806f 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -60,10 +60,11 @@ import org.elasticsearch.xpack.esql.inference.InferenceService; import org.elasticsearch.xpack.esql.optimizer.LocalPhysicalOptimizerContext; import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; +import org.elasticsearch.xpack.esql.plan.logical.Aggregate; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; -import org.elasticsearch.xpack.esql.plan.physical.AggregateExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; +import org.elasticsearch.xpack.esql.plan.physical.FragmentExec; import org.elasticsearch.xpack.esql.plan.physical.OutputExec; import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan; import org.elasticsearch.xpack.esql.planner.EsPhysicalOperationProviders; @@ -842,12 +843,16 @@ public static ReductionPlan reductionPlan( // TODO: How we generate intermediate attributes prevents us from cleanly checking dependencies here. // FragmentExec.output() doesn't take into account intermediate attributes of aggs, and time series aggs // have some peculiarities due to implicit dimensions. We should clean this up and add a proper check here. - if (Assertions.ENABLED && reductionPlan.nodeReducePlan().child() instanceof AggregateExec == false) { - PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), originalPlan.output()); - ExchangeSourceExec reductionSource = (ExchangeSourceExec) reductionPlan.nodeReducePlan().collectLeaves().getFirst(); - // The data driver's output is sent to the reduction driver, so the outputs must match up. - PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionSource.output()); + if (Assertions.ENABLED == false + || (reductionPlan.dataNodePlan().child() instanceof FragmentExec fragment && fragment.fragment() instanceof Aggregate)) { + return reductionPlan; } + + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.nodeReducePlan(), originalPlan.output()); + ExchangeSourceExec reductionSource = (ExchangeSourceExec) reductionPlan.nodeReducePlan().collectLeaves().getFirst(); + // The data driver's output is sent to the reduction driver, so the outputs must match up. + PhysicalVerifier.LOCAL_INSTANCE.verify(reductionPlan.dataNodePlan(), reductionSource.output()); + return reductionPlan; } From c80a7a8cf46c5795688189330fb044ef88fe00d7 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Wed, 25 Feb 2026 13:34:47 +0100 Subject: [PATCH 17/18] Skip check for MetricsInfo --- .../xpack/esql/plugin/ComputeService.java | 20 +++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 866d2e90b806f..4d74123951b39 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -62,6 +62,8 @@ import org.elasticsearch.xpack.esql.optimizer.PhysicalVerifier; import org.elasticsearch.xpack.esql.plan.logical.Aggregate; import org.elasticsearch.xpack.esql.plan.logical.EsRelation; +import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan; +import org.elasticsearch.xpack.esql.plan.logical.MetricsInfo; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec; import org.elasticsearch.xpack.esql.plan.physical.ExchangeSourceExec; import org.elasticsearch.xpack.esql.plan.physical.FragmentExec; @@ -840,11 +842,11 @@ public static ReductionPlan reductionPlan( planTimeProfile.addReductionPlanNanos(System.nanoTime() - startTime); } - // TODO: How we generate intermediate attributes prevents us from cleanly checking dependencies here. - // FragmentExec.output() doesn't take into account intermediate attributes of aggs, and time series aggs - // have some peculiarities due to implicit dimensions. We should clean this up and add a proper check here. + // TODO: How we generate intermediate attributes prevents us from cleanly checking dependencies here. We should always be + // able to perform this check. if (Assertions.ENABLED == false - || (reductionPlan.dataNodePlan().child() instanceof FragmentExec fragment && fragment.fragment() instanceof Aggregate)) { + || (reductionPlan.dataNodePlan().child() instanceof FragmentExec fragment + && skipConsistencyCheckAfterReductionPlanning(fragment.fragment()))) { return reductionPlan; } @@ -856,6 +858,16 @@ public static ReductionPlan reductionPlan( return reductionPlan; } + private static boolean skipConsistencyCheckAfterReductionPlanning(LogicalPlan fragment) { + // FragmentExec.output() doesn't take into account intermediate attributes of aggs, and time series aggs + // have some peculiarities due to implicit dimensions. We should clean this up and add a proper check here. + return fragment instanceof Aggregate + // MetricsInfo does not serialize its output attributes (they are generated automatically and do not depend on the input). + // After de-serializing the data node plan, the output attributes have different NameIds than the ExchangeSink of the + // data node plan. + || fragment instanceof MetricsInfo; + } + String newChildSession(String session) { return session + "/" + childSessionIdGenerator.incrementAndGet(); } From 633e8c762346908837c6b0f2ef006805da3630e3 Mon Sep 17 00:00:00 2001 From: Alexander Spies Date: Wed, 25 Feb 2026 13:37:04 +0100 Subject: [PATCH 18/18] Address Ievgen's comments --- .../plugin/LateMaterializationPlannerGoldenTests.java | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java index 7af74c57d3f37..ed1bcb613e5e6 100644 --- a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java +++ b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/LateMaterializationPlannerGoldenTests.java @@ -12,6 +12,7 @@ import org.elasticsearch.xpack.esql.optimizer.GoldenTestCase; import java.util.EnumSet; +import java.util.Objects; public class LateMaterializationPlannerGoldenTests extends GoldenTestCase { private static final EnumSet STAGES = EnumSet.of( @@ -152,18 +153,12 @@ private static EsqlTestUtils.TestSearchStats missingFieldStats(String missingFie return new EsqlTestUtils.TestSearchStats() { @Override public boolean exists(FieldAttribute.FieldName field) { - if (field.string().equals(missingField)) { - return false; - } - return true; + return Objects.equals(field.string(), missingField) == false; } @Override public boolean isIndexed(FieldAttribute.FieldName field) { - if (field.string().equals(missingField)) { - return false; - } - return true; + return exists(field); } }; }