fix: Fix three bugs in query decorrelation (apache#21208)

## Which issue does this PR close?

- Closes apache#21205.
- Closes apache#21206.
- Closes apache#20315.

## Rationale for this change

This PR fixes three separate bugs in query decorrelation:

1. When removing duplicate filters as part of pulling up `IN`
subqueries, an operator precedence error meant that we would consider
two filters to be duplicates even if they involved different operators
(e.g., `=` and `>`).
2. When generating the `CASE` used to implement "count bug" handling, we
referenced the subquery output column without qualifying it by the
subquery alias. This could result in name-collisions with unrelated
identifiers in the parent query.
3. After generating the `CASE` used for "count bug" handling, we rewrote
the parent query to replace references to the subquery output column
with the generated `CASE` expression. This rewrite only matched on
unqualified column name only, which meant that unrelated parent query
identifiers that happened to share the same column name as subquery
aggregate aliases could been rewritten by mistake.

The first and third issues could result in incorrect query results; the
second would only cause spurious errors, as far as I can see.

## What changes are included in this PR?

* Fix all three bugs
* Add SLT tests
* Add a `debug_assert!` to document/check that
`remove_duplicated_filter` is called with a commutative operator

## Are these changes tested?

Yes.

## Are there any user-facing changes?

No, except in the sense that they fix user-visible bugs.

Author: neilconway

datafusion/optimizer/src/decorrelate.rs

@@ -26,7 +26,9 @@ use crate::simplify_expressions::ExprSimplifier;
 use datafusion_common::tree_node::{
     Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter,
 };
-use datafusion_common::{Column, DFSchemaRef, HashMap, Result, ScalarValue, plan_err};
+use datafusion_common::{
+    Column, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err, plan_err,
+};
 use datafusion_expr::expr::Alias;
 use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::utils::{
@@ -179,7 +181,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
                     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);
+                    join_filters = remove_duplicated_filter(join_filters, in_predicate)?;
                 }
                 let correlated_subquery_cols =
                     collect_subquery_cols(&join_filters, subquery_schema)?;
@@ -460,25 +462,39 @@ fn collect_local_correlated_cols(
     }
 }
 
-fn remove_duplicated_filter(filters: Vec<Expr>, in_predicate: &Expr) -> Vec<Expr> {
-    filters
+fn remove_duplicated_filter(
+    filters: Vec<Expr>,
+    in_predicate: &Expr,
+) -> Result<Vec<Expr>> {
+    // We assume below that swapping the order of operands to an operator does
+    // not change behavior, which is only true if the operator is commutative.
+    assert_or_internal_err!(
+        match in_predicate {
+            Expr::BinaryExpr(b) => b.op.swap() == Some(b.op),
+            _ => true,
+        },
+        "remove_duplicated_filter: in_predicate must use a commutative operator"
+    );
+
+    Ok(filters
         .into_iter()
         .filter(|filter| {
             if filter == in_predicate {
                 return false;
             }
 
-            // ignore the binary order
+            // Treat swapped operand order to a binary operator as equivalent
             !match (filter, in_predicate) {
                 (Expr::BinaryExpr(a_expr), Expr::BinaryExpr(b_expr)) => {
-                    (a_expr.op == b_expr.op)
-                        && (a_expr.left == b_expr.left && a_expr.right == b_expr.right)
-                        || (a_expr.left == b_expr.right && a_expr.right == b_expr.left)
+                    a_expr.op == b_expr.op
+                        && ((a_expr.left == b_expr.left && a_expr.right == b_expr.right)
+                            || (a_expr.left == b_expr.right
+                                && a_expr.right == b_expr.left))
                 }
                 _ => false,
             }
         })
-        .collect::<Vec<_>>()
+        .collect::<Vec<_>>())
 }
 
 fn agg_exprs_evaluation_result_on_empty_batch(

datafusion/optimizer/src/scalar_subquery_to_join.rs

@@ -111,7 +111,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
                                     // replace column references with entry in map, if it exists
                                     if let Some(map_expr) = expr
                                         .try_as_col()
-                                        .and_then(|col| expr_check_map.get(&col.name))
+                                        .and_then(|col| expr_check_map.get(col))
                                     {
                                         Ok(Transformed::yes(map_expr.clone()))
                                     } else {
@@ -176,7 +176,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
                                     // replace column references with entry in map, if it exists
                                     if let Some(map_expr) = expr
                                         .try_as_col()
-                                        .and_then(|col| expr_check_map.get(&col.name))
+                                        .and_then(|col| expr_check_map.get(col))
                                     {
                                         Ok(Transformed::yes(map_expr.clone()))
                                     } else {
@@ -301,7 +301,7 @@ fn build_join(
     subquery: &Subquery,
     filter_input: &LogicalPlan,
     subquery_alias: &str,
-) -> Result<Option<(LogicalPlan, HashMap<String, Expr>)>> {
+) -> Result<Option<(LogicalPlan, HashMap<Column, Expr>)>> {
     let subquery_plan = subquery.subquery.as_ref();
     let mut pull_up = PullUpCorrelatedExpr::new().with_need_handle_count_bug(true);
     let new_plan = subquery_plan.clone().rewrite(&mut pull_up).data()?;
@@ -358,44 +358,43 @@ fn build_join(
                 // If expr always returns null when column is null, skip processing
                 continue;
             }
+
+            let indicator_col =
+                Column::new(Some(subquery_alias), UN_MATCHED_ROW_INDICATOR);
+            // Qualify with the subquery alias to avoid ambiguity when the
+            // outer table has a column with the same name as the aggregate.
+            let value_col = Column::new(Some(subquery_alias), name.clone());
+
             let computer_expr = if let Some(filter) = &pull_up.pull_up_having_expr {
                 Expr::Case(expr::Case {
                     expr: None,
                     when_then_expr: vec![
                         (
-                            Box::new(Expr::IsNull(Box::new(Expr::Column(
-                                Column::new_unqualified(UN_MATCHED_ROW_INDICATOR),
-                            )))),
+                            Box::new(Expr::IsNull(Box::new(Expr::Column(indicator_col)))),
                             Box::new(result),
                         ),
                         (
                             Box::new(Expr::Not(Box::new(filter.clone()))),
                             Box::new(Expr::Literal(ScalarValue::Null, None)),
                         ),
                     ],
-                    else_expr: Some(Box::new(Expr::Column(Column::new_unqualified(
-                        name.clone(),
-                    )))),
+                    else_expr: Some(Box::new(Expr::Column(value_col.clone()))),
                 })
             } else {
                 Expr::Case(expr::Case {
                     expr: None,
                     when_then_expr: vec![(
-                        Box::new(Expr::IsNull(Box::new(Expr::Column(
-                            Column::new_unqualified(UN_MATCHED_ROW_INDICATOR),
-                        )))),
+                        Box::new(Expr::IsNull(Box::new(Expr::Column(indicator_col)))),
                         Box::new(result),
                     )],
-                    else_expr: Some(Box::new(Expr::Column(Column::new_unqualified(
-                        name.clone(),
-                    )))),
+                    else_expr: Some(Box::new(Expr::Column(value_col.clone()))),
                 })
             };
             let mut expr_rewrite = TypeCoercionRewriter {
                 schema: new_plan.schema(),
             };
             computation_project_expr
-                .insert(name, computer_expr.rewrite(&mut expr_rewrite).data()?);
+                .insert(value_col, computer_expr.rewrite(&mut expr_rewrite).data()?);
         }
     }
 

datafusion/sqllogictest/test_files/subquery.slt

@@ -1671,3 +1671,71 @@ drop table employees;
 
 statement count 0
 drop table project_assignments;
+
+# https://github.com/apache/datafusion/issues/21205
+statement ok
+CREATE TABLE dup_filter_t1(id INTEGER) AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE dup_filter_t2(id INTEGER) AS VALUES (1), (2), (3);
+
+query I
+SELECT * FROM dup_filter_t1 WHERE dup_filter_t1.id IN (
+    SELECT dup_filter_t2.id FROM dup_filter_t2 WHERE dup_filter_t2.id > dup_filter_t1.id
+);
+----
+
+statement ok
+DROP TABLE dup_filter_t1;
+
+statement ok
+DROP TABLE dup_filter_t2;
+
+# https://github.com/apache/datafusion/issues/21206
+statement ok
+CREATE TABLE sq_name_t1(id INTEGER) AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE sq_name_t2(id INTEGER, outer_id INTEGER) AS VALUES (10, 1), (20, 1), (30, 2);
+
+query II
+SELECT sq_name_t1.id,
+  (SELECT count(*) AS id FROM sq_name_t2 WHERE sq_name_t2.outer_id = sq_name_t1.id) AS cnt
+FROM sq_name_t1
+ORDER BY sq_name_t1.id;
+----
+1 2
+2 1
+3 0
+
+query I
+SELECT sq_name_t1.id
+FROM sq_name_t1
+WHERE sq_name_t1.id > (
+  SELECT count(*) AS id
+  FROM sq_name_t2
+  WHERE sq_name_t2.outer_id = sq_name_t1.id
+)
+ORDER BY sq_name_t1.id;
+----
+2
+3
+
+query I
+SELECT sq_name_t1.id * 10 + (
+  SELECT count(*) AS id
+  FROM sq_name_t2
+  WHERE sq_name_t2.outer_id = sq_name_t1.id
+) AS total
+FROM sq_name_t1
+ORDER BY sq_name_t1.id;
+----
+12
+21
+30
+
+statement ok
+DROP TABLE sq_name_t1;
+
+statement ok
+DROP TABLE sq_name_t2;