fix: keep renamed columns intact in leaf projection pushdown

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
+    // <https://github.com/apache/datafusion/issues/25414>.
+    //
+    // 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]
         "#)
     }
 

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<x varchar>) 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<b varchar>) 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;