From ebec98f43bccfd548fe307c45bdd26c2e7d2ae85 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Thu, 17 Sep 2026 01:51:54 -0500 Subject: [PATCH 1/4] fix: resolve pass-through columns against the input when merging an extraction projection `build_extraction_projection_impl` merges an extraction projection into the projection below it, and adds the pass-through columns that the merged projection does not already carry. It compared the columns it was about to add against the projection's own expressions without putting the two in the same space. A projection can list bare column names over a qualified input. Eliminating the empty side of a union leaves exactly that. The comparison then misses, the column is added a second time under its bare name, and `Projection::try_new` rejects the result: Optimizer rule 'push_down_leaf_projections' failed caused by Schema error: Schema contains qualified field name samples.env and unqualified field name env which would be ambiguous Resolve both sides against the input schema, and push the column under the name the input gives it. --- .../optimizer/src/extract_leaf_expressions.rs | 37 ++++++++++++++++--- datafusion/sqllogictest/test_files/struct.slt | 25 +++++++++++++ 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index 8a4abfcb48e56..b5ecf56768b1a 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -637,6 +637,24 @@ impl<'a> LeafExpressionExtractor<'a> { } } +/// The way `schema` names `col`, or `None` when it does not hold it unambiguously. +/// +/// A qualified column is taken as it stands. An unqualified one is matched on name alone +/// and comes back carrying the qualifier the schema gives that field, so a column pushed +/// into a projection reads as the input spells it. A name the schema holds more than once +/// resolves to nothing rather than to an arbitrary one of them. +fn resolve_against(schema: &DFSchema, col: &Column) -> Option { + match &col.relation { + Some(relation) => schema + .has_column_with_qualified_name(relation, &col.name) + .then(|| col.clone()), + None => schema + .qualified_field_with_unqualified_name(&col.name) + .ok() + .map(|(qualifier, field)| Column::new(qualifier.cloned(), field.name())), + } +} + /// Build an extraction projection above the target node (shared by both passes). /// /// If the target is an existing projection, merges into it. This requires @@ -702,27 +720,36 @@ fn build_extraction_projection_impl( // than target_schema (the projection's output) because columns produced // by alias expressions (e.g., CSE's __common_expr_N) exist in the output but // not the input, and cannot be added as pass-through Column references. + // + // Both sides of that check are read in the input's own spelling. A column can + // arrive here unqualified while the input names it `t.c`, and the other way round: + // a projection of bare names over a qualified input is what eliminating one side + // of a union leaves behind. Resolving both sides lets a pass-through the + // projection already carries match the one about to be added, and pushes the one + // that is genuinely new under the name the input gives it. Left unresolved, the + // merged projection would hold `t.c` and a bare `c` together, which + // `Projection::try_new` rejects as ambiguous. + let input_schema = existing.input.schema(); let existing_cols: IndexSet = existing .expr .iter() .filter_map(|e| { if let Expr::Column(c) = e { - Some(c.clone()) + resolve_against(input_schema, c) } else { None } }) .collect(); - let input_schema = existing.input.schema(); for col in columns_needed { 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) + && let Some(input_col) = resolve_against(input_schema, resolved_col) + && !existing_cols.contains(&input_col) { - proj_exprs.push(Expr::Column(resolved_col.clone())); + proj_exprs.push(Expr::Column(input_col)); } // If resolved to non-column expr, it's already computed by existing projection } diff --git a/datafusion/sqllogictest/test_files/struct.slt b/datafusion/sqllogictest/test_files/struct.slt index 87bbd11c986a4..10ecbf0885211 100644 --- a/datafusion/sqllogictest/test_files/struct.slt +++ b/datafusion/sqllogictest/test_files/struct.slt @@ -1803,3 +1803,28 @@ drop view struct_ctor_view; statement ok drop table struct_ctor_null; + +# Merging an extraction projection into a projection whose output lost its +# qualifier. Eliminating the empty side of the union leaves a projection of +# bare column names over a qualified input, and the merge used to add the +# pass-through columns under those bare names beside the qualified ones the +# projection already carried, which is an ambiguous schema. +statement ok +create table leaf_merge_source(v int, s struct, env varchar) as values (1, {a: 10}, 'prod'), (2, {a: 20}, 'dev'); + +query TI +with samples as ( + select v, s, env from leaf_merge_source +), +expanded as ( + select v, s, env from samples + union all + select v, s, env from samples where 1 = 2 +) +select env, sum(s['a']) from expanded group by env order by env; +---- +dev 20 +prod 10 + +statement ok +drop table leaf_merge_source; From 96b6af4f2bf2028987051cdafbebc0efde859559 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Thu, 17 Sep 2026 09:40:45 -0500 Subject: [PATCH 2/4] fix: count same-name aliases as pass-throughs when merging an extraction projection A projection can spell a pass-through column as `t.c AS c`. The merge only counted bare column expressions as existing pass-throughs, so it added `t.c` beside the `c` the alias already outputs, which is an ambiguous schema. Use `passthrough_column` to collect them. Co-Authored-By: Claude Opus 5 --- .../optimizer/src/extract_leaf_expressions.rs | 47 +++++++++++++++---- 1 file changed, 39 insertions(+), 8 deletions(-) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index b5ecf56768b1a..f54d71ba1b18a 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -728,18 +728,13 @@ fn build_extraction_projection_impl( // projection already carries match the one about to be added, and pushes the one // that is genuinely new under the name the input gives it. Left unresolved, the // merged projection would hold `t.c` and a bare `c` together, which - // `Projection::try_new` rejects as ambiguous. + // `Projection::try_new` rejects as ambiguous. A same-name alias such as + // `t.c AS c` is a pass-through as well, and counts the same as a bare `t.c`. let input_schema = existing.input.schema(); let existing_cols: IndexSet = existing .expr .iter() - .filter_map(|e| { - if let Expr::Column(c) = e { - resolve_against(input_schema, c) - } else { - None - } - }) + .filter_map(|e| resolve_against(input_schema, passthrough_column(e)?)) .collect(); for col in columns_needed { @@ -2377,6 +2372,42 @@ mod tests { "#) } + /// A projection can spell a pass-through column as a same-name alias + /// (`test.a AS a`). Merging an extraction into it must treat that alias as + /// the pass-through it is, and not add `test.a` beside the `a` it outputs, + /// which is an ambiguous schema. + #[test] + fn test_merge_into_projection_with_same_name_alias() -> Result<()> { + let table_scan = test_table_scan()?; + let plan = LogicalPlanBuilder::from(table_scan) + .project(vec![ + col("test.a").alias("a"), + col("test.b").alias("b"), + col("test.c").alias("c"), + ])? + .filter(leaf_udf(col("a"), "x").eq(lit(1)))? + .build()?; + + assert_stages!(plan, @r#" + ## Original Plan + Filter: leaf_udf(a, Utf8("x")) = Int32(1) + Projection: test.a AS a, test.b AS b, test.c AS c + TableScan: test projection=[a, b, c] + + ## After Extraction + Projection: a, b, c + Filter: __datafusion_extracted_1 = Int32(1) + Projection: test.a AS a, test.b AS b, test.c AS c, leaf_udf(test.a, Utf8("x")) AS __datafusion_extracted_1 + TableScan: test projection=[a, b, c] + + ## After Pushdown + (same as after extraction) + + ## Optimized + (same as after pushdown) + "#) + } + // ========================================================================= // Join extraction tests // ========================================================================= From 61923e2f0fabf121d0ef20ee6d0b0ffd9cef61e8 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Thu, 17 Sep 2026 09:54:09 -0500 Subject: [PATCH 3/4] test: cover merging a bare column into a qualified projection Co-Authored-By: Claude Opus 5 --- .../optimizer/src/extract_leaf_expressions.rs | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index f54d71ba1b18a..e267a9d5b8751 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -2372,6 +2372,47 @@ mod tests { "#) } + /// A filter can name a column bare (`a`) while the projection below it + /// outputs the qualified `test.a`, as eliminating the empty side of a union + /// leaves behind. The merge must match the two, and not add a bare `a` + /// beside `test.a`, which is an ambiguous schema. + #[test] + fn test_merge_bare_column_into_qualified_projection() -> Result<()> { + let table_scan = test_table_scan()?; + let projection = LogicalPlanBuilder::from(table_scan) + .project(vec![ + col("test.a"), + col("test.b"), + (col("test.c") + lit(1)).alias("d"), + ])? + .build()?; + let predicate = + leaf_udf(Expr::Column(Column::new_unqualified("a")), "x").eq(lit(1)); + let plan = LogicalPlan::Filter(datafusion_expr::Filter::try_new( + predicate, + Arc::new(projection), + )?); + + assert_stages!(plan, @r#" + ## Original Plan + Filter: leaf_udf(a, Utf8("x")) = Int32(1) + Projection: test.a, test.b, test.c + Int32(1) AS d + TableScan: test projection=[a, b, c] + + ## After Extraction + Projection: test.a, test.b, d + Filter: __datafusion_extracted_1 = Int32(1) + Projection: test.a, test.b, test.c + Int32(1) AS d, leaf_udf(a, Utf8("x")) AS __datafusion_extracted_1 + TableScan: test projection=[a, b, c] + + ## After Pushdown + (same as after extraction) + + ## Optimized + (same as after pushdown) + "#) + } + /// A projection can spell a pass-through column as a same-name alias /// (`test.a AS a`). Merging an extraction into it must treat that alias as /// the pass-through it is, and not add `test.a` beside the `a` it outputs, From 53a845898a0d68c8dde18cb1f1f129102b1a537b Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 23 Sep 2026 20:02:00 -0500 Subject: [PATCH 4/4] refactor: simplify resolve_against and lock the plan shape in struct.slt - Resolve both qualified and unqualified columns with one `qualified_field_from_column` lookup. The result always carries the input schema's qualifier, so two spellings of the same column compare equal. Correct the doc: a duplicate name held by exactly one unqualified field resolves to that field, not to `None`. - Make the merge comment shorter. - Add a logical EXPLAIN to the struct.slt case, so the test shows when union elimination stops producing the shape it covers. Co-Authored-By: Claude Opus 5.5 --- .../optimizer/src/extract_leaf_expressions.rs | 37 +++++++------------ datafusion/sqllogictest/test_files/struct.slt | 26 +++++++++++++ 2 files changed, 40 insertions(+), 23 deletions(-) diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index e267a9d5b8751..76c008640855f 100644 --- a/datafusion/optimizer/src/extract_leaf_expressions.rs +++ b/datafusion/optimizer/src/extract_leaf_expressions.rs @@ -637,22 +637,17 @@ impl<'a> LeafExpressionExtractor<'a> { } } -/// The way `schema` names `col`, or `None` when it does not hold it unambiguously. +/// The way `schema` names `col`, or `None` when `schema` does not hold it +/// or the name is ambiguous. /// -/// A qualified column is taken as it stands. An unqualified one is matched on name alone -/// and comes back carrying the qualifier the schema gives that field, so a column pushed -/// into a projection reads as the input spells it. A name the schema holds more than once -/// resolves to nothing rather than to an arbitrary one of them. +/// The result always carries the qualifier the schema gives the field, so +/// two spellings of the same input column compare equal, and a column pushed +/// into a projection reads as the input spells it. fn resolve_against(schema: &DFSchema, col: &Column) -> Option { - match &col.relation { - Some(relation) => schema - .has_column_with_qualified_name(relation, &col.name) - .then(|| col.clone()), - None => schema - .qualified_field_with_unqualified_name(&col.name) - .ok() - .map(|(qualifier, field)| Column::new(qualifier.cloned(), field.name())), - } + schema + .qualified_field_from_column(col) + .ok() + .map(Column::from) } /// Build an extraction projection above the target node (shared by both passes). @@ -721,15 +716,11 @@ fn build_extraction_projection_impl( // by alias expressions (e.g., CSE's __common_expr_N) exist in the output but // not the input, and cannot be added as pass-through Column references. // - // Both sides of that check are read in the input's own spelling. A column can - // arrive here unqualified while the input names it `t.c`, and the other way round: - // a projection of bare names over a qualified input is what eliminating one side - // of a union leaves behind. Resolving both sides lets a pass-through the - // projection already carries match the one about to be added, and pushes the one - // that is genuinely new under the name the input gives it. Left unresolved, the - // merged projection would hold `t.c` and a bare `c` together, which - // `Projection::try_new` rejects as ambiguous. A same-name alias such as - // `t.c AS c` is a pass-through as well, and counts the same as a bare `t.c`. + // Compare both sides in the input's spelling (see `resolve_against`). + // Without this, a bare `c` and a qualified `t.c` do not match, and the + // merged projection holds both, which `Projection::try_new` rejects as + // ambiguous. Eliminating the empty side of a union makes this shape. + // A same-name alias (`t.c AS c`) counts as a pass-through of `t.c`. let input_schema = existing.input.schema(); let existing_cols: IndexSet = existing .expr diff --git a/datafusion/sqllogictest/test_files/struct.slt b/datafusion/sqllogictest/test_files/struct.slt index 10ecbf0885211..eaa0a3f20c832 100644 --- a/datafusion/sqllogictest/test_files/struct.slt +++ b/datafusion/sqllogictest/test_files/struct.slt @@ -1812,6 +1812,32 @@ drop table struct_ctor_null; statement ok create table leaf_merge_source(v int, s struct, env varchar) as values (1, {a: 10}, 'prod'), (2, {a: 20}, 'dev'); +statement ok +set datafusion.explain.logical_plan_only = true; + +query TT +explain with samples as ( + select v, s, env from leaf_merge_source +), +expanded as ( + select v, s, env from samples + union all + select v, s, env from samples where 1 = 2 +) +select env, sum(s['a']) from expanded group by env order by env; +---- +logical_plan +01)Sort: expanded.env ASC NULLS LAST +02)--Projection: expanded.env, sum(__datafusion_extracted_1) AS sum(expanded.s[a]) +03)----Aggregate: groupBy=[[expanded.env]], aggr=[[sum(CAST(__datafusion_extracted_1 AS Int64))]] +04)------SubqueryAlias: expanded +05)--------SubqueryAlias: samples +06)----------Projection: leaf_merge_source.env, get_field(s, Utf8("a")) AS __datafusion_extracted_1 +07)------------TableScan: leaf_merge_source projection=[s, env] + +statement ok +set datafusion.explain.logical_plan_only = false; + query TI with samples as ( select v, s, env from leaf_merge_source