diff --git a/datafusion/optimizer/src/extract_leaf_expressions.rs b/datafusion/optimizer/src/extract_leaf_expressions.rs index 8a4abfcb48e56..8c35a03b0c849 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,92 @@ 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) + "#) + } + + /// 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] "#) } 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;