fix: keep the recovery projection when leaf pushdown would drop a computed same-name column

datafusion/optimizer/src/extract_leaf_expressions.rs

@@ -1111,6 +1111,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 +1134,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 {
@@ -3601,4 +3616,43 @@ mod tests {
 
         Ok(())
     }
+
+    /// Regression test for <https://github.com/apache/datafusion/issues/25414>.
+    ///
+    /// `(- test.id) AS id` computes a new value under the same name as its input
+    /// column `test.id`. Pushing the extraction below that projection makes
+    /// `test.id` visible again under the name `id`. The recovery projection must
+    /// stay, or the computed column is silently replaced by the table column.
+    ///
+    /// The two leaf rules run alone here, in their production order.
+    /// `optimize_projections` merges the two projections into one and hides the
+    /// shape, and it only runs after both leaf rules.
+    #[test]
+    fn test_recovery_kept_for_same_name_computed_column() -> Result<()> {
+        let table_scan = test_table_scan_with_struct()?;
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .filter(col("id").gt(lit(0u32)))?
+            .project(vec![
+                Expr::Negative(Box::new(col("id"))).alias("id"),
+                col("user"),
+            ])?
+            .project(vec![col("id"), leaf_udf(col("user"), "name")])?
+            .build()?;
+
+        let ctx = OptimizerContext::new().with_max_passes(1);
+        let optimizer = Optimizer::with_rules(vec![
+            Arc::new(ExtractLeafExpressions::new()),
+            Arc::new(PushDownLeafProjections::new()),
+        ]);
+        let optimized = optimizer.optimize(plan, &ctx, |_, _| {})?;
+
+        insta::assert_snapshot!(format!("{optimized}"), @r#"
+        Projection: id, __datafusion_extracted_1 AS leaf_udf(test.user,Utf8("name"))
+          Projection: (- test.id) 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
+        "#);
+        Ok(())
+    }
 }

datafusion/sqllogictest/test_files/struct.slt

@@ -1803,3 +1803,72 @@ drop view struct_ctor_view;
 
 statement ok
 drop table struct_ctor_null;
+
+# Regression test for https://github.com/apache/datafusion/issues/25414.
+# `-a AS a` computes a new value under the same name as its input column.
+# Leaf expression pushdown moves `s['b']` below that projection. The projection
+# that computes `-a` must stay, or the query returns the input column `a`.
+
+statement ok
+create table leaf_same_name(a int, s struct<b varchar>) as values (1, {b: 'x'}), (2, {b: 'y'});
+
+# The computed column must survive the pushdown. Line 02 holds `(- t.a) AS a`.
+query TT
+explain select a, s['b'] from (select -a as a, s from leaf_same_name where a > 0);
+----
+logical_plan
+01)Projection: (- leaf_same_name.a) AS a, __datafusion_extracted_1 AS leaf_same_name.s[b]
+02)--Filter: leaf_same_name.a > Int32(0)
+03)----Projection: get_field(leaf_same_name.s, Utf8("b")) AS __datafusion_extracted_1, leaf_same_name.a
+04)------TableScan: leaf_same_name projection=[a, s]
+physical_plan
+01)ProjectionExec: expr=[(- a@1) as a, __datafusion_extracted_1@0 as leaf_same_name.s[b]]
+02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+03)----FilterExec: a@1 > 0
+04)------ProjectionExec: expr=[get_field(s@1, b) as __datafusion_extracted_1, a@0 as a]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Through a Filter
+query IT rowsort
+select a, s['b'] from (select -a as a, s from leaf_same_name where a > 0);
+----
+-1 x
+-2 y
+
+# Through a Limit
+query IT rowsort
+select a, s['b'] from (select -a as a, s from leaf_same_name limit 10);
+----
+-1 x
+-2 y
+
+# A computed column whose type differs from its input column
+query IT rowsort
+select a, s['b'] from (select a * 10 as a, s from leaf_same_name limit 10);
+----
+10 x
+20 y
+
+# The outer filter must see the computed value
+query IT rowsort
+select a, s['b'] from (select -a as a, s from leaf_same_name) where a < 0;
+----
+-1 x
+-2 y
+
+# The group key must be the computed value
+query II rowsort
+select a, count(s['b']) from (select -a as a, s from leaf_same_name where a > 0) group by a;
+----
+-1 1
+-2 1
+
+# Through a Sort
+query IT rowsort
+select a, s['b'] from (select -a as a, s from leaf_same_name order by a);
+----
+-1 x
+-2 y
+
+statement ok
+drop table leaf_same_name;