fix: Correct ANY/ALL results for unaliased subqueries on the same table

datafusion/expr/src/utils.rs

@@ -1427,6 +1427,10 @@ pub fn add_filter(plan: LogicalPlan, predicates: &[&Expr]) -> Result<LogicalPlan
 /// # Return value
 ///
 /// Tuple of (expressions containing joins, remaining non-join expressions)
+#[deprecated(
+    since = "56.0.0",
+    note = "This decorrelation helper is intended for internal optimizer use and has no public replacement"
+)]
 pub fn find_join_exprs(exprs: Vec<&Expr>) -> Result<(Vec<Expr>, Vec<Expr>)> {
     let mut joins = vec![];
     let mut others = vec![];

datafusion/optimizer/src/decorrelate.rs

@@ -21,6 +21,7 @@ use std::collections::BTreeSet;
 use std::sync::Arc;
 
 use crate::simplify_expressions::ExprSimplifier;
+use crate::utils::replace_qualified_name;
 
 use datafusion_common::tree_node::{
     Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
@@ -29,10 +30,9 @@ use datafusion_common::{
     Column, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err, plan_err,
 };
 use datafusion_expr::expr::Alias;
+use datafusion_expr::expr_rewriter::strip_outer_reference;
 use datafusion_expr::simplify::SimplifyContext;
-use datafusion_expr::utils::{
-    collect_subquery_cols, conjunction, find_join_exprs, split_conjunction,
-};
+use datafusion_expr::utils::{collect_subquery_cols, conjunction, split_conjunction};
 use datafusion_expr::{
     BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, LogicalPlan,
     LogicalPlanBuilder, Operator, expr, lit,
@@ -183,7 +183,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
                         .filter(|e| e.contains_outer())
                         .all(|&e| can_pullup_over_aggregation(e));
                 let (mut join_filters, subquery_filters) =
-                    find_join_exprs(subquery_filter_exprs)?;
+                    find_join_exprs(subquery_filter_exprs);
                 if let Some(in_predicate) = &self.in_predicate_opt {
                     // in_predicate may be already included in the join filters, remove it from the join filters first.
                     join_filters = remove_duplicated_filter(join_filters, in_predicate)?;
@@ -480,6 +480,60 @@ fn collect_local_correlated_cols(
     }
 }
 
+/// Extracts correlated predicates, such as a comparison between a column from the
+/// subquery and a column from the outer scope.
+///
+/// Preserves [`Expr::OuterReferenceColumn`] markers so callers can distinguish
+/// inner and outer columns when qualifying join filters (see [`build_join_filter`]).
+///
+/// # Arguments
+///
+/// * `exprs` - Subquery filter predicates to classify.
+///
+/// # Return value
+///
+/// Tuple of (correlated join filters, remaining subquery filters). Correlated
+/// self-equalities are discarded from both lists.
+pub(crate) fn find_join_exprs(exprs: Vec<&Expr>) -> (Vec<Expr>, Vec<Expr>) {
+    let mut joins = vec![];
+    let mut others = vec![];
+    for filter in exprs {
+        // Predicates containing outer references become join filters.
+        if filter.contains_outer() {
+            // Check equality before stripping markers: inner and outer columns
+            // with the same qualified name still belong to different scopes.
+            if !matches!(filter, Expr::BinaryExpr(BinaryExpr { left, op: Operator::Eq, right }) if left.eq(right))
+            {
+                joins.push(filter.clone());
+            }
+        } else {
+            others.push(filter.clone());
+        }
+    }
+    (joins, others)
+}
+
+/// Combines join filters and qualifies their inner columns with the subquery alias.
+///
+/// Aliasing must precede stripping outer references so inner and outer columns
+/// with the same qualified name remain distinguishable.
+pub(crate) fn build_join_filter<'a>(
+    join_filters: Vec<Expr>,
+    correlated_cols: impl IntoIterator<Item = &'a Column>,
+    subquery_alias: &str,
+) -> 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)
+                .map(strip_outer_reference)
+        })
+        .transpose()
+}
+
 fn remove_duplicated_filter(
     filters: Vec<Expr>,
     in_predicate: &Expr,
@@ -496,6 +550,7 @@ fn remove_duplicated_filter(
 
     Ok(filters
         .into_iter()
+        .map(strip_outer_reference)
         .filter(|filter| {
             if filter == in_predicate {
                 return false;
@@ -652,3 +707,38 @@ fn filter_exprs_evaluation_result_on_empty_batch(
     };
     Ok(pull_up_expr)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::datatypes::DataType;
+    use datafusion_expr::{col, out_ref_col};
+
+    #[test]
+    fn test_find_join_exprs_preserves_outer_references() {
+        let outer = out_ref_col(DataType::Int32, "t.a");
+        let correlated = col("t.a").eq(outer.clone());
+        let self_equality = outer.clone().eq(outer);
+        let local = col("t.b").gt(lit(2));
+
+        assert_eq!(
+            find_join_exprs(vec![&correlated, &self_equality, &local]),
+            (vec![correlated], vec![local])
+        );
+    }
+
+    #[test]
+    fn test_build_join_filter() -> Result<()> {
+        assert_eq!(build_join_filter(vec![], [], "sq")?, 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")?,
+            Some(col("sq.a").eq(col("t.a")).and(col("t.a").gt(lit(1))))
+        );
+        Ok(())
+    }
+}

datafusion/optimizer/src/decorrelate_lateral_join.rs

@@ -23,6 +23,7 @@ use crate::decorrelate::{PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR};
 use crate::optimizer::ApplyOrder;
 use crate::utils::evaluates_to_null;
 use crate::{OptimizerConfig, OptimizerRule};
+use datafusion_expr::expr_rewriter::strip_outer_reference;
 use datafusion_expr::{Expr, Join, expr};
 
 use datafusion_common::tree_node::{
@@ -153,6 +154,8 @@ fn rewrite_internal(join: Join) -> Result<Transformed<LogicalPlan>> {
             (rewritten_subquery, correlation_filter, original_join_filter)
         };
 
+    let correlation_filter = correlation_filter.map(strip_outer_reference);
+
     // For LEFT lateral joins, verify that all column references in the
     // correlation filter are resolvable within the join's left and right
     // schemas. If the lateral subquery references columns from an outer scope,

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -16,14 +16,12 @@
 // under the License.
 
 //! [`DecorrelatePredicateSubquery`] converts `IN`/`EXISTS` subquery predicates to `SEMI`/`ANTI` joins
-use std::collections::BTreeSet;
 use std::ops::Deref;
 use std::sync::Arc;
 
-use crate::decorrelate::PullUpCorrelatedExpr;
+use crate::decorrelate::{PullUpCorrelatedExpr, build_join_filter};
 use crate::extract_equijoin_predicate::split_eq_and_noneq_join_predicate;
 use crate::optimizer::ApplyOrder;
-use crate::utils::replace_qualified_name;
 use crate::{OptimizerConfig, OptimizerRule};
 
 use datafusion_common::alias::AliasGenerator;
@@ -374,17 +372,11 @@ fn build_join(
     let sub_query_alias = LogicalPlanBuilder::from(new_plan)
         .alias(alias.to_string())?
         .build()?;
-    let mut all_correlated_cols = BTreeSet::new();
-    pull_up
-        .correlated_subquery_cols_map
-        .values()
-        .for_each(|cols| all_correlated_cols.extend(cols.clone()));
-
-    // alias the join filter
-    let join_filter_opt = conjunction(pull_up.join_filters)
-        .map_or(Ok(None), |filter| {
-            replace_qualified_name(filter, &all_correlated_cols, &alias).map(Some)
-        })?;
+    let join_filter_opt = build_join_filter(
+        pull_up.join_filters,
+        pull_up.correlated_subquery_cols_map.values().flatten(),
+        &alias,
+    )?;
 
     let join_filter = match (join_filter_opt, in_predicate_opt.cloned()) {
         (
@@ -765,7 +757,7 @@ mod tests {
             SubqueryAlias: __correlated_sq_2 [o_custkey:Int64]
               Projection: orders.o_custkey [o_custkey:Int64]
                 TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]
-        "    
+        "
         )
     }
 

datafusion/optimizer/src/scalar_subquery_to_join.rs

@@ -17,12 +17,14 @@
 
 //! [`ScalarSubqueryToJoin`] rewriting scalar subquery filters to `JOIN`s
 
-use std::collections::{BTreeSet, HashMap};
+use std::collections::HashMap;
 use std::sync::Arc;
 
-use crate::decorrelate::{PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR};
+use crate::decorrelate::{
+    PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR, build_join_filter,
+};
 use crate::optimizer::ApplyOrder;
-use crate::utils::{evaluates_to_null, replace_qualified_name};
+use crate::utils::evaluates_to_null;
 use crate::{OptimizerConfig, OptimizerRule};
 
 use crate::analyzer::type_coercion::TypeCoercionRewriter;
@@ -33,7 +35,6 @@ use datafusion_common::tree_node::{
 use datafusion_common::{Column, Result, ScalarValue, assert_or_internal_err, plan_err};
 use datafusion_expr::expr_rewriter::create_col_from_scalar_expr;
 use datafusion_expr::logical_plan::{JoinType, Subquery};
-use datafusion_expr::utils::conjunction;
 use datafusion_expr::{Expr, LogicalPlan, LogicalPlanBuilder, lit, not, when};
 
 /// Optimizer rule that rewrites scalar subquery filters to joins and places an
@@ -365,19 +366,11 @@ fn build_join(
         .alias(subquery_alias.to_string())?
         .build()?;
 
-    let all_correlated_cols: BTreeSet<Column> = pull_up
-        .correlated_subquery_cols_map
-        .values()
-        .flatten()
-        .cloned()
-        .collect();
-
-    // Correlated columns now live in the decorrelated subquery's output,
-    // so re-qualify them with the subquery alias.
-    let join_filter_opt =
-        conjunction(pull_up.join_filters).map_or(Ok(None), |filter| {
-            replace_qualified_name(filter, &all_correlated_cols, subquery_alias).map(Some)
-        })?;
+    let join_filter_opt = build_join_filter(
+        pull_up.join_filters,
+        pull_up.correlated_subquery_cols_map.values().flatten(),
+        subquery_alias,
+    )?;
 
     // When pull-up did not extract any usable join keys (a correlated subquery
     // whose predicate references only outer columns), fall back to `ON true`: