fix: requalify LATERAL filters through derived-table aliases

datafusion/optimizer/src/decorrelate.rs

@@ -31,7 +31,7 @@ use datafusion_common::tree_node::{
 use datafusion_common::utils::has_float_leaf;
 use datafusion_common::{
     Column, DFSchema, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err,
-    plan_err,
+    internal_err, plan_err,
 };
 use datafusion_expr::expr::{Alias, GroupingSet};
 use datafusion_expr::expr_rewriter::strip_outer_reference;
@@ -102,6 +102,12 @@ pub struct PullUpCorrelatedExpr {
     /// The list is cleared when the pull up passes a node that can put a NULL
     /// back into such a column: an outer join, a union or a grouping set.
     pub correlated_filters: Vec<Expr>,
+    /// `join_filters.len()` at each `SubqueryAlias` on the way down.
+    ///
+    /// Filters appended while that alias is being rewritten belong to its
+    /// subtree. On the way up, only those filters are requalified to the
+    /// alias, so a sibling filter that names the same column is left alone.
+    alias_filter_base: Vec<usize>,
     /// Columns each node on the path from the subquery's root to the node
     /// [`Self::f_down`]/[`Self::f_up`] is currently visiting reads in its own
     /// expressions: one entry per node on that path, pushed in `f_down` and
@@ -133,6 +139,7 @@ impl PullUpCorrelatedExpr {
             pull_up_having_expr: None,
             pulled_up_scalar_agg: false,
             correlated_filters: Vec::new(),
+            alias_filter_base: Vec::new(),
             column_refs_above: Vec::new(),
         }
     }
@@ -309,6 +316,17 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
                     _ => Ok(self.stop_pull_up(plan)),
                 }
             }
+            // Remember how many join filters exist before this alias's
+            // children are rewritten. `f_up` requalifies only the filters
+            // those children add. See #25837.
+            LogicalPlan::SubqueryAlias(_) => {
+                self.alias_filter_base.push(self.join_filters.len());
+                if plan.contains_outer_reference() {
+                    Ok(self.stop_pull_up(plan))
+                } else {
+                    Ok(Transformed::no(plan))
+                }
+            }
             // A correlated filter below these nodes can move above them. The
             // node itself must not hold an outer reference, because only a
             // Filter gives its outer references to the join.
@@ -337,7 +355,6 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
             | LogicalPlan::Join(_)
             | LogicalPlan::AsOfJoin(_)
             | LogicalPlan::Repartition(_)
-            | LogicalPlan::SubqueryAlias(_)
             | LogicalPlan::TableScan(_)
             | LogicalPlan::EmptyRelation(_)
             | LogicalPlan::MaterializedCte(MaterializedCtePlan::Scan(_))
@@ -671,22 +688,49 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
                     &self.correlated_subquery_cols_map,
                     &mut local_correlated_cols,
                 );
+                // A filter pulled up from below this alias still names the
+                // inner table (`l.id`). The alias's output names that column
+                // `l2.id`, and the join condition has to match.
+                // https://github.com/apache/datafusion/issues/25837
+                //
+                // Pull-up can add a column whose name another projected
+                // column already has. `SubqueryAlias::try_new` suffixes that
+                // duplicate (`id` becomes `id:1`), so the output name is the
+                // field at the same position, not the inner name.
+                let Some(base) = self.alias_filter_base.pop() else {
+                    return internal_err!(
+                        "SubqueryAlias f_down pushes alias_filter_base"
+                    );
+                };
+                // The projection and the duplicate-name suffixes come from
+                // `SubqueryAlias::try_new`. Read this schema before rebuilding:
+                // the projection that constructor inserts already uses the
+                // suffixed names.
+                let input_schema = Arc::clone(alias.input.schema());
+                let new_plan =
+                    if input_schema.fields().len() != alias.schema.fields().len() {
+                        LogicalPlanBuilder::from((*alias.input).clone())
+                            .alias(alias.alias.clone())?
+                            .build()?
+                    } else {
+                        plan.clone()
+                    };
+                let output_schema = Arc::clone(new_plan.schema());
+                requalify_join_filter_columns(
+                    &mut self.join_filters[base..],
+                    &local_correlated_cols,
+                    &input_schema,
+                    &output_schema,
+                )?;
                 let mut new_correlated_cols = BTreeSet::new();
-                for col in local_correlated_cols.iter() {
-                    new_correlated_cols
-                        .insert(Column::new(Some(alias.alias.clone()), col.name.clone()));
+                for col in &local_correlated_cols {
+                    new_correlated_cols.insert(alias_output_column(
+                        col,
+                        &input_schema,
+                        &output_schema,
+                    )?);
                 }
 
-                let new_plan = if alias.input.schema().fields().len()
-                    != alias.schema.fields().len()
-                {
-                    LogicalPlanBuilder::from((*alias.input).clone())
-                        .alias(alias.alias.clone())?
-                        .build()?
-                } else {
-                    plan.clone()
-                };
-
                 self.correlated_subquery_cols_map
                     .insert(new_plan.clone(), new_correlated_cols);
                 if let Some(input_map) = self.collected_count_expr_map.get(&*alias.input)
@@ -1103,6 +1147,55 @@ fn split_column_equality(expr: &Expr) -> Option<(&Expr, &Expr)> {
         })
 }
 
+/// Rewrite columns in `filters` that are listed in `cols` so they use the
+/// alias's output column at the same position.
+///
+/// `cols` are the correlated columns as named below the alias. Outer
+/// references are not in that set, so they stay as they are. The caller
+/// passes only the filters added under this alias.
+///
+/// `input_schema` is the schema passed to `SubqueryAlias::try_new`.
+/// `output_schema` is the schema it built. Fields line up by position,
+/// including the dedup suffixes `try_new` applies.
+fn requalify_join_filter_columns(
+    filters: &mut [Expr],
+    cols: &BTreeSet<Column>,
+    input_schema: &DFSchema,
+    output_schema: &DFSchema,
+) -> Result<()> {
+    if cols.is_empty() || filters.is_empty() {
+        return Ok(());
+    }
+    for filter in filters {
+        *filter =
+            replace_qualified_name(filter.clone(), cols, input_schema, output_schema)?;
+    }
+    Ok(())
+}
+
+/// The alias output column that corresponds to `col` in `input_schema`.
+///
+/// `SubqueryAlias::try_new` keeps field order. When two input fields share a
+/// name, the later one's output name is suffixed (`id`, then `id:1`).
+///
+/// Returns an error if `col` is not in `input_schema`, or if that position
+/// is past the end of `output_schema`. Callers must not skip the column:
+/// leaving it out drops the predicate that referenced it.
+pub(crate) fn alias_output_column(
+    col: &Column,
+    input_schema: &DFSchema,
+    output_schema: &DFSchema,
+) -> Result<Column> {
+    let idx = input_schema.index_of_column(col)?;
+    assert_or_internal_err!(
+        idx < output_schema.fields().len(),
+        "SubqueryAlias output schema has {} fields, input column {col} is at {idx}",
+        output_schema.fields().len(),
+    );
+    let (qualifier, field) = output_schema.qualified_field(idx);
+    Ok(Column::new(qualifier.cloned(), field.name()))
+}
+
 fn collect_local_correlated_cols(
     plan: &LogicalPlan,
     all_cols_map: &HashMap<LogicalPlan, BTreeSet<Column>>,
@@ -1159,15 +1252,16 @@ pub(crate) fn find_join_exprs(exprs: Vec<&Expr>) -> (Vec<Expr>, Vec<Expr>) {
 pub(crate) fn build_join_filter<'a>(
     join_filters: Vec<Expr>,
     correlated_cols: impl IntoIterator<Item = &'a Column>,
-    subquery_alias: &str,
+    input_schema: &DFSchema,
+    output_schema: &DFSchema,
 ) -> Result<Option<Expr>> {
     let correlated_cols = correlated_cols
         .into_iter()
         .cloned()
         .collect::<BTreeSet<_>>();
     conjunction(join_filters)
         .map(|filter| {
-            replace_qualified_name(filter, &correlated_cols, subquery_alias)
+            replace_qualified_name(filter, &correlated_cols, input_schema, output_schema)
                 .map(strip_outer_reference)
         })
         .transpose()
@@ -1359,7 +1453,7 @@ fn filter_exprs_evaluation_result_on_empty_batch(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow::datatypes::DataType;
+    use arrow::datatypes::{DataType, Field, Schema};
     use datafusion_expr::{col, out_ref_col};
 
     #[test]
@@ -1377,14 +1471,30 @@ mod tests {
 
     #[test]
     fn test_build_join_filter() -> Result<()> {
-        assert_eq!(build_join_filter(vec![], [], "sq")?, None);
+        let input_schema = DFSchema::try_from_qualified_schema(
+            "t",
+            &Schema::new(vec![Field::new("a", DataType::Int32, true)]),
+        )?;
+        let output_schema = DFSchema::try_from_qualified_schema(
+            "sq",
+            &Schema::new(vec![Field::new("a", DataType::Int32, true)]),
+        )?;
+        assert_eq!(
+            build_join_filter(vec![], [], &input_schema, &output_schema)?,
+            None
+        );
 
         let column = Column::from("t.a");
         let outer = out_ref_col(DataType::Int32, "t.a");
         let filters = vec![col("t.a").eq(outer.clone()), outer.gt(lit(1))];
 
         assert_eq!(
-            build_join_filter(filters, [&column, &column], "sq")?,
+            build_join_filter(
+                filters,
+                [&column, &column],
+                &input_schema,
+                &output_schema
+            )?,
             Some(col("sq.a").eq(col("t.a")).and(col("t.a").gt(lit(1))))
         );
         Ok(())

datafusion/optimizer/src/decorrelate_lateral_join.rs

@@ -19,7 +19,9 @@
 
 use std::sync::Arc;
 
-use crate::decorrelate::{PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR};
+use crate::decorrelate::{
+    PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR, alias_output_column,
+};
 use crate::optimizer::ApplyOrder;
 use crate::utils::evaluates_to_null;
 use crate::{OptimizerConfig, OptimizerRule};
@@ -130,22 +132,27 @@ fn rewrite_internal(join: Join) -> Result<Transformed<LogicalPlan>> {
         .cloned();
 
     // Re-wrap in SubqueryAlias if the original had one, preserving the alias name.
-    // The SubqueryAlias re-qualifies all columns with the alias, so we must also
-    // rewrite column references in both the correlation and ON-clause filters.
+    // The correlation filter still names the inner tables. `try_new` may suffix
+    // a duplicate, so those columns are mapped by position.
+    //
+    // The user's ON clause already names the alias output. Pull-up only appends
+    // columns, and `unique_field_aliases` only suffixes those new columns, so
+    // the names ON uses do not change. Mapping ON through the pre-alias schema
+    // is wrong when an inner column shares the alias qualifier (`l AS s` inside
+    // `LATERAL (...) s`): `s.id` would move from the first output `id` to a
+    // later `id:1`.
     let (right_plan, correlation_filter, original_join_filter) =
         if let Some(ref alias) = alias {
-            let inner_schema = Arc::clone(rewritten_subquery.schema());
+            let input_schema = Arc::clone(rewritten_subquery.schema());
             let right = LogicalPlan::SubqueryAlias(SubqueryAlias::try_new(
                 Arc::new(rewritten_subquery),
                 alias.clone(),
             )?);
+            let output_schema = Arc::clone(right.schema());
             let corr = correlation_filter
-                .map(|f| requalify_filter(f, &inner_schema, alias))
+                .map(|f| requalify_filter(f, &input_schema, &output_schema))
                 .transpose()?;
-            let on = original_join_filter
-                .map(|f| requalify_filter(f, &inner_schema, alias))
-                .transpose()?;
-            (right, corr, on)
+            (right, corr, original_join_filter)
         } else {
             (rewritten_subquery, correlation_filter, original_join_filter)
         };
@@ -346,24 +353,31 @@ fn extract_lateral_subquery(
 }
 
 /// Rewrite column references in a join filter expression so that columns
-/// belonging to the inner (right) side use the SubqueryAlias qualifier.
+/// belonging to the inner (right) side use the `SubqueryAlias` output column
+/// at the same position.
 ///
 /// The `PullUpCorrelatedExpr` pass extracts join filters with the inner
 /// columns qualified by their original table names (e.g., `t2.t1_id`).
 /// When the inner plan is wrapped in a `SubqueryAlias("sub")`, those
-/// columns are re-qualified as `sub.t1_id`. This function applies the
-/// same requalification to the filter so it matches the aliased schema.
+/// columns are re-qualified from `input_schema` (the schema passed to
+/// `SubqueryAlias::try_new`) to `output_schema` (the schema it built).
+/// A duplicate name is suffixed there (`id`, then `id:1`), so keeping
+/// `col.name` would point at the earlier column.
+///
+/// Columns that are not in `input_schema` are outer references and stay as
+/// they are. A column that is in `input_schema` but has no output field is
+/// an error: skipping it would drop that predicate.
 fn requalify_filter(
     filter: Expr,
-    inner_schema: &DFSchema,
-    alias: &TableReference,
+    input_schema: &DFSchema,
+    output_schema: &DFSchema,
 ) -> Result<Expr> {
     filter
         .transform(|expr| {
             if let Expr::Column(col) = &expr
-                && inner_schema.has_column(col)
+                && input_schema.has_column(col)
             {
-                let new_col = Column::new(Some(alias.clone()), col.name.clone());
+                let new_col = alias_output_column(col, input_schema, output_schema)?;
                 return Ok(Transformed::yes(Expr::Column(new_col)));
             }
             Ok(Transformed::no(expr))

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -884,13 +884,15 @@ fn build_join(
         }
     }
 
+    let input_schema = Arc::clone(new_plan.schema());
     let sub_query_alias = LogicalPlanBuilder::from(new_plan)
         .alias(alias.to_string())?
         .build()?;
     let join_filter_opt = build_join_filter(
         pull_up.join_filters,
         pull_up.correlated_subquery_cols_map.values().flatten(),
-        alias,
+        &input_schema,
+        sub_query_alias.schema(),
     )?;
 
     if let Some(expr_map) = count_bug_compensation

datafusion/optimizer/src/scalar_subquery_to_join.rs

@@ -358,14 +358,16 @@ fn build_join(
         .collected_count_expr_map
         .get(&decorrelated_subquery)
         .cloned();
+    let input_schema = Arc::clone(decorrelated_subquery.schema());
     let aliased_subquery = LogicalPlanBuilder::from(decorrelated_subquery)
         .alias(subquery_alias.to_string())?
         .build()?;
 
     let join_filter_opt = build_join_filter(
         pull_up.join_filters,
         pull_up.correlated_subquery_cols_map.values().flatten(),
-        subquery_alias,
+        &input_schema,
+        aliased_subquery.schema(),
     )?;
 
     // When pull-up did not extract any usable join keys (a correlated subquery