fix: dedup commutative AND/OR operands when canonicalize is disabled

datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs

@@ -2526,6 +2526,48 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_simplify_swapped_operands_in_and_or_no_canonicalize() {
+        // Regression test for https://github.com/apache/datafusion/issues/14943
+        //
+        // `SimplifyExpressions` disables canonicalization for `LogicalPlan::Join`
+        // (see https://github.com/apache/datafusion/pull/8780), so commutative
+        // operands like `A = B` and `B = A` cannot be normalized to a single
+        // form before AND/OR dedup runs. The dedup itself must therefore
+        // recognize commutative equivalence directly.
+
+        // c3 = 5 AND 5 = c3 --> c3 = 5
+        let expr = col("c3_non_null")
+            .eq(lit(5_i64))
+            .and(lit(5_i64).eq(col("c3_non_null")));
+        let expected = col("c3_non_null").eq(lit(5_i64));
+        assert_eq!(simplify_no_canonicalize(expr), expected);
+
+        // 5 = c3 AND c3 = 5 --> 5 = c3
+        let expr = lit(5_i64)
+            .eq(col("c3_non_null"))
+            .and(col("c3_non_null").eq(lit(5_i64)));
+        let expected = lit(5_i64).eq(col("c3_non_null"));
+        assert_eq!(simplify_no_canonicalize(expr), expected);
+
+        // c3 = 5 OR 5 = c3 --> c3 = 5
+        let expr = col("c3_non_null")
+            .eq(lit(5_i64))
+            .or(lit(5_i64).eq(col("c3_non_null")));
+        let expected = col("c3_non_null").eq(lit(5_i64));
+        assert_eq!(simplify_no_canonicalize(expr), expected);
+
+        // (c3 = 5 AND c4 > 0) AND (5 = c3) --> c3 = 5 AND c4 > 0
+        let expr = col("c3_non_null")
+            .eq(lit(5_i64))
+            .and(col("c4_non_null").gt(lit(0_u32)))
+            .and(lit(5_i64).eq(col("c3_non_null")));
+        let expected = col("c3_non_null")
+            .eq(lit(5_i64))
+            .and(col("c4_non_null").gt(lit(0_u32)));
+        assert_eq!(simplify_no_canonicalize(expr), expected);
+    }
+
     #[test]
     fn test_simplify_eq_not_self() {
         // `expr_a`: column `c2` is nullable, so `c2 = c2` simplifies to `c2 IS NOT NULL OR NULL`
@@ -3648,6 +3690,14 @@ mod tests {
         try_simplify(expr).unwrap()
     }
 
+    fn simplify_no_canonicalize(expr: Expr) -> Expr {
+        let schema = expr_test_schema();
+        ExprSimplifier::new(SimplifyContext::builder().with_schema(schema).build())
+            .with_canonicalize(false)
+            .simplify(expr)
+            .unwrap()
+    }
+
     fn try_simplify_with_cycle_count(expr: Expr) -> Result<(Expr, u32)> {
         let schema = expr_test_schema();
         let simplifier =

datafusion/optimizer/src/simplify_expressions/utils.rs

@@ -18,6 +18,7 @@
 //! Utility functions for expression simplification
 
 use arrow::datatypes::i256;
+use datafusion_common::cse::NormalizeEq;
 use datafusion_common::{Result, ScalarValue, internal_err};
 use datafusion_expr::{
     Case, Expr, Like, Operator,
@@ -68,13 +69,18 @@ pub static POWS_OF_TEN: [i128; 38] = [
 
 /// returns true if `needle` is found in a chain of search_op
 /// expressions. Such as: (A AND B) AND C
+///
+/// Equality uses [`NormalizeEq`] so commutative operands (`A = B` vs `B = A`)
+/// are recognized as the same predicate even when the canonicalizer is
+/// disabled, notably for `LogicalPlan::Join` filters (see
+/// <https://github.com/apache/datafusion/pull/8780>).
 fn expr_contains_inner(expr: &Expr, needle: &Expr, search_op: Operator) -> bool {
     match expr {
         Expr::BinaryExpr(BinaryExpr { left, op, right }) if *op == search_op => {
             expr_contains_inner(left, needle, search_op)
                 || expr_contains_inner(right, needle, search_op)
         }
-        _ => expr == needle,
+        _ => expr.normalize_eq(needle),
     }
 }
 

datafusion/sqllogictest/test_files/simplify_expr.slt

@@ -146,3 +146,57 @@ logical_plan
 physical_plan
 01)ProjectionExec: expr=[column1@0 = 1 as opt1, column1@0 = 2 AND column1@0 != 2 as noopt1, column1@0 = 4 as opt2, column1@0 != 5 AND column1@0 = 5 as noopt2]
 02)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Dedup commutative AND/OR operands in join conditions
+# (canonicalization is disabled for joins, so `A = B` vs `B = A` must be
+# recognized directly). See https://github.com/apache/datafusion/issues/14943
+
+statement ok
+create table t1(a int) as values (1), (2);
+
+statement ok
+create table t2(b int) as values (1), (3);
+
+# `t1.a = t2.b OR t2.b = t1.a` simplifies to `t1.a = t2.b`, allowing a
+# HashJoin instead of a NestedLoopJoin
+query TT
+explain select * from t1 join t2 on t1.a = t2.b or t2.b = t1.a;
+----
+logical_plan
+01)Inner Join: t1.a = t2.b
+02)--TableScan: t1 projection=[a]
+03)--TableScan: t2 projection=[b]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+query II
+select * from t1 join t2 on t1.a = t2.b or t2.b = t1.a;
+----
+1 1
+
+# swapped commutative operands (`t1.a + t2.b` vs `t2.b + t1.a`) dedup within
+# a non-equi join filter
+query TT
+explain select * from t1 join t2 on t1.a > t2.b and (t1.a + t2.b > 1 or t2.b + t1.a > 1);
+----
+logical_plan
+01)Inner Join:  Filter: t1.a > t2.b AND t1.a + t2.b > Int32(1)
+02)--TableScan: t1 projection=[a]
+03)--TableScan: t2 projection=[b]
+physical_plan
+01)NestedLoopJoinExec: join_type=Inner, filter=a@0 > b@1 AND a@0 + b@1 > 1
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+query II
+select * from t1 join t2 on t1.a > t2.b and (t1.a + t2.b > 1 or t2.b + t1.a > 1);
+----
+2 1
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;