From c67e79a7f5e70f43228cd0f262652c1b3329cf49 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 18 Sep 2026 00:09:35 -0500 Subject: [PATCH 1/2] fix: keep renamed columns intact in leaf projection pushdown A sub-query projection that renames a column to the name of a different column of the same input made `push_down_leaf_projections` fail when a struct field is read above it: Schema error: Schema contains qualified field name t.a and unqualified field name a which would be ambiguous The merge of the extraction projection into the projection below it resolved every column the parent needs through the projection's rename map, then added the input column that the name resolves to. For `select t.a as b, t.b as a, s from t`, the parent's `b` resolves to `t.a`, and `t.a` lands beside the output field `a`. The merge keeps every expression of the projection below it, so the parent can still read each name that projection produces. Skip those names instead of resolving them, and never add a pass-through column whose name is one of the projection's output names. The recovery check then needs the same care, or the shape gives wrong results instead of an error. Equal sets of field names do not prove that the plan below carries the same values. That hunk is the same change as https://github.com/apache/datafusion/pull/25445, so the two merge without a conflict. Three plan snapshots lose a duplicate pass-through column in the intermediate stage. The optimized plans do not change. Co-Authored-By: Claude Fable 5.1 --- .../optimizer/src/extract_leaf_expressions.rs | 58 ++++++++++++++----- 1 file changed, 42 insertions(+), 16 deletions(-) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index 8a4abfcb48e56..3e500d85cf605 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -714,13 +714,33 @@ fn build_extraction_projection_impl( }) .collect(); + // The names the merged projection already produces. The merge keeps + // every expression of `existing`, so the parent can still read each of + // these names, and a pass-through column that carries one of them makes + // the output schema ambiguous. + let output_names: std::collections::HashSet<&str> = existing + .schema + .fields() + .iter() + .map(|f| f.name().as_str()) + .collect(); + let input_schema = existing.input.schema(); for col in columns_needed { + // The projection produces this name, so the parent reads it from the + // projection's own output. Do not resolve the name through the rename + // map: a rename such as `t.a AS b` is a computed output, and its + // input column `t.a` beside the output field `a` of a second rename + // gives an ambiguous schema (issue #25446). + if output_names.contains(col.name.as_str()) { + continue; + } let col_expr = Expr::Column(col.clone()); let resolved = replace_cols_by_name(col_expr, &replace_map)?; if let Expr::Column(resolved_col) = &resolved && !existing_cols.contains(resolved_col) && input_schema.has_column(resolved_col) + && !output_names.contains(resolved_col.name.as_str()) { proj_exprs.push(Expr::Column(resolved_col.clone())); } @@ -1111,6 +1131,18 @@ fn split_and_push_projection( // `SubqueryAlias` re-qualification (`sub.__datafusion_extracted_1` vs // `__datafusion_extracted_1`) that a qualified/ordered comparison would // spuriously treat as drift, stacking redundant recovery projections. + // + // A name comparison alone is not sufficient. A name says nothing about the + // *value* behind it. Take the projection + // `(- t.a) AS a, t.s, get_field(t.s, "b") AS __datafusion_extracted_1`. When + // the extraction goes below it, the pushed plan keeps every name, but it + // exposes the table column `t.a` where the projection computed `- t.a`. If + // the recovery projection goes away, the computed column becomes its own + // input column and the query gives wrong results. See + // . + // + // So the recovery projection also stays when a recovery expression computes + // a value, that is, when it is not a pass-through of a column. let base_names: BTreeSet<&str> = base_plan .schema() .fields() @@ -1122,7 +1154,10 @@ fn split_and_push_projection( .iter() .map(|f| f.name().as_str()) .collect(); - let needs_recovery = base_names != original_names; + let computes_a_value = recovery_exprs + .iter() + .any(|expr| passthrough_column(expr).is_none()); + let needs_recovery = base_names != original_names || computes_a_value; // Wrap with recovery projection if the output schema changed if needs_recovery { @@ -2777,14 +2812,11 @@ mod tests { ## After Pushdown Projection: __datafusion_extracted_1 AS leaf_udf(x,Utf8("a")) Filter: x IS NOT NULL - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1, test.user + Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 TableScan: test projection=[user] ## Optimized - Projection: __datafusion_extracted_1 AS leaf_udf(x,Utf8("a")) - Filter: x IS NOT NULL - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 - TableScan: test projection=[user] + (same as after pushdown) "#) } @@ -2811,14 +2843,11 @@ mod tests { ## After Pushdown Projection: __datafusion_extracted_1 IS NOT NULL AS leaf_udf(x,Utf8("a")) IS NOT NULL Filter: x IS NOT NULL - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1, test.user + Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 TableScan: test projection=[user] ## Optimized - Projection: __datafusion_extracted_1 IS NOT NULL AS leaf_udf(x,Utf8("a")) IS NOT NULL - Filter: x IS NOT NULL - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 - TableScan: test projection=[user] + (same as after pushdown) "#) } @@ -2840,17 +2869,14 @@ mod tests { ## After Extraction Projection: x Filter: __datafusion_extracted_1 = Utf8("active") - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1, test.user + Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 TableScan: test projection=[user] ## After Pushdown (same as after extraction) ## Optimized - Projection: x - Filter: __datafusion_extracted_1 = Utf8("active") - Projection: test.user AS x, leaf_udf(test.user, Utf8("a")) AS __datafusion_extracted_1 - TableScan: test projection=[user] + (same as after pushdown) "#) } From ba695d451195949f2bfa97b77b3af801c1eb17c2 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 18 Sep 2026 08:19:48 -0500 Subject: [PATCH 2/2] test: cover renamed and swapped columns in leaf projection pushdown Add the statements from the report as sqllogictest cases: a rename beside a same-name alias under a filter, a limit, an order by and a group by, a swap of two column names, and a swap where the struct field name is also a column name. Add two optimizer unit tests: one for the ambiguous schema, one for the recovery projection that must keep a rename alive. Co-Authored-By: Claude Fable 5.1 --- .../optimizer/src/extract_leaf_expressions.rs | 78 +++++++++++++++++++ datafusion/sqllogictest/test_files/struct.slt | 75 ++++++++++++++++++ 2 files changed, 153 insertions(+) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index 3e500d85cf605..8c35a03b0c849 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -2880,6 +2880,84 @@ mod tests { "#) } + /// A projection that swaps two column names, below a filter and a struct + /// field read. The merge must not resolve the parent's column references + /// through the rename map: the input column `test.user` beside the output + /// field `user` of the other rename gives an ambiguous schema (#25446). + #[test] + fn test_extract_above_projection_that_swaps_column_names() -> Result<()> { + let table_scan = test_table_scan_with_struct()?; + let plan = LogicalPlanBuilder::from(table_scan) + .project(vec![col("user").alias("id"), col("id").alias("user")])? + .filter(col("user").gt(lit(0u32)))? + .project(vec![col("id"), col("user"), leaf_udf(col("id"), "name")])? + .build()?; + + assert_stages!(plan, @r#" + ## Original Plan + Projection: id, user, leaf_udf(id, Utf8("name")) + Filter: user > UInt32(0) + Projection: test.user AS id, test.id AS user + TableScan: test projection=[id, user] + + ## After Extraction + (same as original) + + ## After Pushdown + Projection: id, user, __datafusion_extracted_1 AS leaf_udf(id,Utf8("name")) + Filter: user > UInt32(0) + Projection: test.user AS id, test.id AS user, leaf_udf(test.user, Utf8("name")) AS __datafusion_extracted_1 + TableScan: test projection=[id, user] + + ## Optimized + (same as after pushdown) + "#) + } + + /// A projection that gives a computed column the name of its own input + /// column. The extraction projection goes below the filter, which puts a + /// column named `id` under the rename, so the two plans have the same set + /// of field names. The recovery projection must stay, or the rename is lost + /// and the query gives wrong results. + #[test] + fn test_extract_above_projection_that_redefines_column_name() -> Result<()> { + let table_scan = test_table_scan_with_struct()?; + let plan = LogicalPlanBuilder::from(table_scan) + .filter(col("id").gt(lit(0u32)))? + .project(vec![(col("id") * lit(10u32)).alias("id"), col("user")])? + .alias("sub")? + .project(vec![col("sub.id"), leaf_udf(col("sub.user"), "name")])? + .build()?; + + assert_stages!(plan, @r#" + ## Original Plan + Projection: sub.id, leaf_udf(sub.user, Utf8("name")) + SubqueryAlias: sub + Projection: test.id * UInt32(10) AS id, test.user + Filter: test.id > UInt32(0) + TableScan: test projection=[id, user] + + ## After Extraction + (same as original) + + ## After Pushdown + Projection: sub.id, __datafusion_extracted_1 AS leaf_udf(sub.user,Utf8("name")) + SubqueryAlias: sub + Projection: test.id * UInt32(10) AS id, test.user, __datafusion_extracted_1 + Filter: test.id > UInt32(0) + Projection: leaf_udf(test.user, Utf8("name")) AS __datafusion_extracted_1, test.id, test.user + TableScan: test projection=[id, user] + + ## Optimized + Projection: sub.id, __datafusion_extracted_1 AS leaf_udf(sub.user,Utf8("name")) + SubqueryAlias: sub + Projection: test.id * UInt32(10) AS id, __datafusion_extracted_1 + Filter: test.id > UInt32(0) + Projection: leaf_udf(test.user, Utf8("name")) AS __datafusion_extracted_1, test.id + TableScan: test projection=[id, user] + "#) + } + // ========================================================================= // SubqueryAlias extraction tests // ========================================================================= diff --git a/datafusion/sqllogictest/test_files/struct.slt b/datafusion/sqllogictest/test_files/struct.slt index 87bbd11c986a4..085304e3b423d 100644 --- a/datafusion/sqllogictest/test_files/struct.slt +++ b/datafusion/sqllogictest/test_files/struct.slt @@ -1803,3 +1803,78 @@ drop view struct_ctor_view; statement ok drop table struct_ctor_null; + +# A sub-query projection that renames a column to the name of a different +# column of the same input, with a struct field read above it. Leaf projection +# pushdown resolved the parent's column references through the rename map and +# added the renamed input column a second time, which made the output schema +# ambiguous (https://github.com/apache/datafusion/issues/25446). +statement ok +create table rename_swap_struct(a int, b int, s struct) as values (1, 10, {x: 'p'}), (2, 20, {x: 'q'}); + +# a rename beside a same-name alias, under a filter +query IIT +select b, a, s['x'] from (select rename_swap_struct.a as b, rename_swap_struct.b as a, s from rename_swap_struct) where a > 0 order by b; +---- +1 10 p +2 20 q + +# the same shape under a limit +query IIT rowsort +select b, a, s['x'] from (select rename_swap_struct.a as b, rename_swap_struct.b as a, s from rename_swap_struct) limit 10; +---- +1 10 p +2 20 q + +# a swap of two column names +query IIT +select a, c, s['x'] from (select rename_swap_struct.b as a, rename_swap_struct.a as c, s from rename_swap_struct) where a > 0 order by a; +---- +10 1 p +20 2 q + +# the same shape under an order by +query IIT +select b, a, s['x'] from (select rename_swap_struct.a as b, rename_swap_struct.b as a, s from rename_swap_struct) order by b; +---- +1 10 p +2 20 q + +# the same shape under a group by +query III +select b, a, count(s['x']) from (select rename_swap_struct.a as b, rename_swap_struct.b as a, s from rename_swap_struct) group by b, a order by b; +---- +1 10 1 +2 20 1 + +# the rename stays above the extraction projection, and the struct field is +# still read at the scan +query TT +explain select b, a, s['x'] from (select rename_swap_struct.a as b, rename_swap_struct.b as a, s from rename_swap_struct) where a > 0; +---- +logical_plan +01)Projection: rename_swap_struct.a AS b, rename_swap_struct.b AS a, __datafusion_extracted_1 AS rename_swap_struct.s[x] +02)--Filter: rename_swap_struct.b > Int32(0) +03)----Projection: get_field(rename_swap_struct.s, Utf8("x")) AS __datafusion_extracted_1, rename_swap_struct.a, rename_swap_struct.b +04)------TableScan: rename_swap_struct projection=[a, b, s] +physical_plan +01)ProjectionExec: expr=[a@1 as b, b@2 as a, __datafusion_extracted_1@0 as rename_swap_struct.s[x]] +02)--FilterExec: b@2 > 0 +03)----ProjectionExec: expr=[get_field(s@2, x) as __datafusion_extracted_1, a@0 as a, b@1 as b] +04)------DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +drop table rename_swap_struct; + +# The same swap where the struct field name is also a column name of the table. +statement ok +create table rename_swap_struct_field(a int, b int, s struct) as values (1, 10, {b: 'p'}), (2, 20, {b: 'q'}); + +query IIT +select a, b, s['b'] from (select b as a, a as b, s from rename_swap_struct_field limit 100) order by a; +---- +10 1 p +20 2 q + +statement ok +drop table rename_swap_struct_field;