fix: scan subqueries when advancing the extracted-alias generator

datafusion/optimizer/src/extract_leaf_expressions.rs

@@ -134,15 +134,19 @@ impl OptimizerRule for ExtractLeafExpressions {
     }
 }
 
-/// Scans the current plan node's expressions for pre-existing
-/// `__datafusion_extracted_N` aliases and advances the generator
-/// counter past them to avoid collisions with user-provided aliases.
+/// Scans the plan for pre-existing `__datafusion_extracted_N` aliases and
+/// advances the generator counter past them to avoid collisions with
+/// user-provided aliases.
+///
+/// Subquery plans nested inside expressions are scanned as well: extraction
+/// rewrites with `transform_down_with_subqueries`, so it can generate aliases
+/// *inside* a subquery and would otherwise collide with a user alias there.
 fn advance_generator_past_existing(
     plan: &LogicalPlan,
     alias_generator: &AliasGenerator,
 ) -> Result<()> {
-    plan.apply(|plan| {
-        plan.expressions().iter().try_for_each(|expr| {
+    plan.apply_with_subqueries(|plan| {
+        plan.apply_expressions(|expr| {
             expr.apply(|e| {
                 if let Expr::Alias(alias) = e
                     && let Some(id) = alias
@@ -154,10 +158,8 @@ fn advance_generator_past_existing(
                     alias_generator.update_min_id(id);
                 }
                 Ok(TreeNodeRecursion::Continue)
-            })?;
-            Ok::<(), datafusion_common::error::DataFusionError>(())
-        })?;
-        Ok(TreeNodeRecursion::Continue)
+            })
+        })
     })
     .map(|_| ())
 }
@@ -3253,4 +3255,30 @@ mod tests {
         "#);
         Ok(())
     }
+
+    /// Pre-existing `__datafusion_extracted_N` aliases must advance the alias
+    /// generator even when they live inside a subquery plan, since extraction
+    /// descends into subqueries and would otherwise reuse the same alias.
+    #[test]
+    fn test_advance_generator_past_alias_in_subquery() -> Result<()> {
+        use datafusion_expr::in_subquery;
+
+        let subquery = LogicalPlanBuilder::from(test_table_scan_with_struct()?)
+            .project(vec![
+                leaf_udf(col("user"), "name").alias("__datafusion_extracted_7"),
+            ])?
+            .build()?;
+        let plan = LogicalPlanBuilder::from(test_table_scan_with_struct_named("outer")?)
+            .filter(in_subquery(col("id"), Arc::new(subquery)))?
+            .build()?;
+
+        let alias_generator = AliasGenerator::new();
+        advance_generator_past_existing(&plan, &alias_generator)?;
+
+        assert_eq!(
+            alias_generator.next(EXTRACTED_EXPR_PREFIX),
+            "__datafusion_extracted_8"
+        );
+        Ok(())
+    }
 }

datafusion/sqllogictest/test_files/projection_pushdown.slt

@@ -2183,3 +2183,74 @@ physical_plan
 # Reset the config changed above (the SLT runner expects target_partitions = 4).
 statement ok
 SET datafusion.execution.target_partitions = 4;
+
+#####################
+# Section: extraction aliases inside a subquery advance the alias generator
+#
+# Regression test for `advance_generator_past_existing` in
+# `extract_leaf_expressions.rs`.
+#
+# `ExtractLeafExpressions` names the columns it extracts
+# `__datafusion_extracted_N`, handing out N from a shared `AliasGenerator`. That
+# prefix is reserved for the optimizer, but nothing stops a user from writing it,
+# so before extracting the rule scans the plan for existing
+# `__datafusion_extracted_N` aliases and bumps the generator past the highest one.
+#
+# The bug: that scan used `apply`, which walks plan nodes but does *not* descend
+# into subquery plans held inside expressions, while the extraction itself uses
+# `transform_down_with_subqueries` and *does* rewrite inside subqueries. So an
+# alias living only inside a subquery was invisible to the scan, and extraction
+# then minted the very same name next to it.
+#
+# Each ingredient below is load-bearing:
+#
+# * The `IN (<subquery>)` sits in the SELECT list, not in a WHERE clause, so
+#   `decorrelate_predicate_subquery` (which runs earlier) leaves it alone and it
+#   is still a subquery expression by the time extraction runs. A subquery in
+#   WHERE would be flattened into the main plan, where the old scan could see it.
+# * The alias inside the subquery is literally `__datafusion_extracted_1`. Rename
+#   it to anything outside the reserved prefix and there is nothing to collide
+#   with -- the query then passes with or without the fix and guards nothing.
+# * The inner `WHERE s['value'] > 120` is what forces extraction to *generate* an
+#   alias inside that same subquery. Without a leaf expression there, the
+#   generator is never called where the collision would happen.
+#
+# Without the fix, extraction reuses `__datafusion_extracted_1` and planning
+# aborts with: Optimizer rule 'push_down_leaf_projections' failed Schema error:
+# Schema contains duplicate unqualified field name __datafusion_extracted_1.
+# With the fix, the generator starts at 2, as the plan below shows.
+#
+# This is `EXPLAIN` under `logical_plan_only` rather than an executed query
+# because a surviving `InSubquery` expression has no physical plan; the logical
+# plan is both the observable result and exactly what regressed.
+#####################
+
+statement ok
+set datafusion.explain.logical_plan_only = true;
+
+query TT
+EXPLAIN
+SELECT
+    id,
+    id IN (
+        SELECT id
+        FROM (
+            SELECT id, s['label'] AS __datafusion_extracted_1
+            FROM simple_struct
+            WHERE s['value'] > 120
+        )
+        WHERE __datafusion_extracted_1 <> 'delta'
+    ) AS has_matching_label
+FROM simple_struct;
+----
+logical_plan
+01)Projection: simple_struct.id, simple_struct.id IN (<subquery>) AS has_matching_label
+02)--Subquery:
+03)----Projection: simple_struct.id
+04)------Filter: __datafusion_extracted_2 > Int64(120) AND __datafusion_extracted_1 != Utf8("delta")
+05)--------Projection: get_field(simple_struct.s, Utf8("value")) AS __datafusion_extracted_2, simple_struct.id, get_field(simple_struct.s, Utf8("label")) AS __datafusion_extracted_1
+06)----------TableScan: simple_struct projection=[id, s], partial_filters=[get_field(simple_struct.s, Utf8("value")) > Int64(120)]
+07)--TableScan: simple_struct projection=[id]
+
+statement ok
+set datafusion.explain.logical_plan_only = false;