Revert "Implement inferred join predicate handling to maintain filters during optimization"

kosiew · kosiew · commit 9f9f0ac17a0b · 2025-08-09T18:13:28.000+08:00
This reverts commit 8337f80.
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
@@ -18,7 +18,7 @@
 //! Planner for [`LogicalPlan`] to [`ExecutionPlan`]
 
 use std::borrow::Cow;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use crate::datasource::file_format::file_type_to_format;
@@ -151,46 +151,6 @@ pub trait ExtensionPlanner {
     ) -> Result<Option<Arc<dyn ExecutionPlan>>>;
 }
 
-fn expr_canonical_string(expr: &Arc<dyn PhysicalExpr>) -> String {
-    format!("{}", expr)
-}
-
-fn dedupe_join_key_pairs(
-    left: Vec<Arc<dyn PhysicalExpr>>,
-    right: Vec<Arc<dyn PhysicalExpr>>,
-) -> (Vec<Arc<dyn PhysicalExpr>>, Vec<Arc<dyn PhysicalExpr>>) {
-    assert_eq!(left.len(), right.len());
-    let mut seen = HashSet::new();
-    let mut out_left = Vec::new();
-    let mut out_right = Vec::new();
-    for (l, r) in left.into_iter().zip(right.into_iter()) {
-        let l_s = expr_canonical_string(&l);
-        let r_s = expr_canonical_string(&r);
-        let key = if l_s <= r_s {
-            format!("{}={}", l_s, r_s)
-        } else {
-            format!("{}={}", r_s, l_s)
-        };
-        if seen.insert(key) {
-            out_left.push(l);
-            out_right.push(r);
-        }
-    }
-    (out_left, out_right)
-}
-
-fn dedupe_partition_exprs(exprs: Vec<Arc<dyn PhysicalExpr>>) -> Vec<Arc<dyn PhysicalExpr>> {
-    let mut seen = HashSet::new();
-    let mut out = Vec::new();
-    for e in exprs.into_iter() {
-        let key = expr_canonical_string(&e);
-        if seen.insert(key) {
-            out.push(e);
-        }
-    }
-    out
-}
-
 /// Default single node physical query planner that converts a
 /// `LogicalPlan` to an `ExecutionPlan` suitable for execution.
 ///
@@ -898,7 +858,6 @@ impl DefaultPhysicalPlanner {
                                 )
                             })
                             .collect::<Result<Vec<_>>>()?;
-                        let runtime_expr = dedupe_partition_exprs(runtime_expr);
                         Partitioning::Hash(runtime_expr, *n)
                     }
                     LogicalPartitioning::DistributeBy(_) => {
@@ -1129,10 +1088,6 @@ impl DefaultPhysicalPlanner {
                         Ok((l, r))
                     })
                     .collect::<Result<join_utils::JoinOn>>()?;
-                let (left_keys, right_keys): (Vec<_>, Vec<_>) = join_on.into_iter().unzip();
-                let (left_keys, right_keys) = dedupe_join_key_pairs(left_keys, right_keys);
-                let join_on: join_utils::JoinOn =
-                    left_keys.into_iter().zip(right_keys).collect();
 
                 let join_filter = match filter {
                     Some(expr) => {
@@ -2390,7 +2345,6 @@ impl<'a> OptimizationInvariantChecker<'a> {
 
         Ok(())
     }
-
 }
 
 impl<'n> TreeNodeVisitor<'n> for OptimizationInvariantChecker<'_> {
@@ -3566,29 +3520,4 @@ digraph {
 
         Ok(())
     }
-
-    #[test]
-    fn test_dedupe_join_key_pairs() {
-        let l1: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
-        let r1: Arc<dyn PhysicalExpr> = Arc::new(Column::new("b", 0));
-        let l2: Arc<dyn PhysicalExpr> = Arc::new(Column::new("b", 0));
-        let r2: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
-        let (left, right) = dedupe_join_key_pairs(vec![l1.clone(), l2], vec![r1.clone(), r2]);
-        assert_eq!(left.len(), 1);
-        assert_eq!(format!("{}", left[0]), "a@0");
-        assert_eq!(format!("{}", right[0]), "b@0");
-    }
-
-    #[test]
-    fn test_dedupe_partition_exprs() {
-        let exprs: Vec<Arc<dyn PhysicalExpr>> = vec![
-            Arc::new(Column::new("a", 0)),
-            Arc::new(Column::new("a", 0)),
-            Arc::new(Column::new("b", 1)),
-        ];
-        let result = dedupe_partition_exprs(exprs);
-        assert_eq!(result.len(), 2);
-        assert_eq!(format!("{}", result[0]), "a@0");
-        assert_eq!(format!("{}", result[1]), "b@1");
-    }
 }
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
@@ -38,7 +38,6 @@ use datafusion_common::tree_node::{
 use datafusion_common::{
     Column, DFSchema, HashMap, Result, ScalarValue, Spans, TableReference,
 };
-use crate::utils::INFERRED_PREDICATE_ALIAS;
 use datafusion_functions_window_common::field::WindowUDFFieldArgs;
 use sqlparser::ast::{
     display_comma_separated, ExceptSelectItem, ExcludeSelectItem, IlikeSelectItem,
@@ -3184,13 +3183,7 @@ pub const UNNEST_COLUMN_PREFIX: &str = "UNNEST";
 impl Display for Expr {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         match self {
-            Expr::Alias(Alias { expr, name, .. }) => {
-                if name == INFERRED_PREDICATE_ALIAS {
-                    write!(f, "{expr}")
-                } else {
-                    write!(f, "{expr} AS {name}")
-                }
-            }
+            Expr::Alias(Alias { expr, name, .. }) => write!(f, "{expr} AS {name}"),
             Expr::Column(c) => write!(f, "{c}"),
             Expr::OuterReferenceColumn(_, c) => {
                 write!(f, "{OUTER_REFERENCE_COLUMN_PREFIX}({c})")
diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs
@@ -28,15 +28,6 @@ use crate::{
 };
 use datafusion_expr_common::signature::{Signature, TypeSignature};
 
-/// Alias used to mark inferred join predicates that should remain join filters
-/// and not be converted into additional equijoin keys during optimization.
-pub const INFERRED_PREDICATE_ALIAS: &str = "__datafusion_inferred_join_predicate";
-
-/// Returns true if expr is an alias used to mark inferred join predicates.
-pub fn is_inferred_alias(expr: &Expr) -> bool {
-    matches!(expr, Expr::Alias(Alias { name, .. }) if name == INFERRED_PREDICATE_ALIAS)
-}
-
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion_common::tree_node::{
     Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
@@ -955,9 +946,7 @@ fn split_conjunction_impl<'a>(expr: &'a Expr, mut exprs: Vec<&'a Expr>) -> Vec<&
             let exprs = split_conjunction_impl(left, exprs);
             split_conjunction_impl(right, exprs)
         }
-        Expr::Alias(alias) if !is_inferred_alias(expr) => {
-            split_conjunction_impl(&alias.expr, exprs)
-        }
+        Expr::Alias(Alias { expr, .. }) => split_conjunction_impl(expr, exprs),
         other => {
             exprs.push(other);
             exprs
@@ -1074,13 +1063,8 @@ fn split_binary_owned_impl(
             let exprs = split_binary_owned_impl(*left, operator, exprs);
             split_binary_owned_impl(*right, operator, exprs)
         }
-        Expr::Alias(alias) => {
-            if is_inferred_alias(&Expr::Alias(alias.clone())) {
-                exprs.push(Expr::Alias(alias));
-                exprs
-            } else {
-                split_binary_owned_impl(*alias.expr, operator, exprs)
-            }
+        Expr::Alias(Alias { expr, .. }) => {
+            split_binary_owned_impl(*expr, operator, exprs)
         }
         other => {
             exprs.push(other);
@@ -1106,9 +1090,7 @@ fn split_binary_impl<'a>(
             let exprs = split_binary_impl(left, operator, exprs);
             split_binary_impl(right, operator, exprs)
         }
-        Expr::Alias(alias) if !is_inferred_alias(expr) => {
-            split_binary_impl(&alias.expr, operator, exprs)
-        }
+        Expr::Alias(Alias { expr, .. }) => split_binary_impl(expr, operator, exprs),
         other => {
             exprs.push(other);
             exprs
@@ -1666,17 +1648,6 @@ mod tests {
         assert_eq!(result, vec![&expr1, &expr2]);
     }
 
-    #[test]
-    fn test_split_conjunction_inferred_alias() {
-        let expr = col("a").eq(col("b")).alias(INFERRED_PREDICATE_ALIAS);
-        let result = split_conjunction(&expr);
-        assert_eq!(result.len(), 1);
-        match result[0] {
-            Expr::Alias(alias) => assert_eq!(alias.name, INFERRED_PREDICATE_ALIAS),
-            _ => panic!("expected alias"),
-        }
-    }
-
     #[test]
     fn test_split_conjunction_or() {
         let expr = col("a").eq(lit(5)).or(col("b"));
diff --git a/datafusion/optimizer/src/extract_equijoin_predicate.rs b/datafusion/optimizer/src/extract_equijoin_predicate.rs
@@ -22,9 +22,8 @@ use datafusion_common::tree_node::Transformed;
 use datafusion_common::DFSchema;
 use datafusion_common::Result;
 use datafusion_expr::utils::split_conjunction_owned;
-use datafusion_expr::utils::{can_hash, find_valid_equijoin_key_pair, is_inferred_alias};
+use datafusion_expr::utils::{can_hash, find_valid_equijoin_key_pair};
 use datafusion_expr::{BinaryExpr, Expr, ExprSchemable, Join, LogicalPlan, Operator};
-use std::collections::HashSet;
 // equijoin predicate
 type EquijoinPredicate = (Expr, Expr);
 
@@ -82,17 +81,14 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                 let right_schema = right.schema();
                 let (equijoin_predicates, non_equijoin_expr) =
                     split_eq_and_noneq_join_predicate(expr, left_schema, right_schema)?;
-                let has_new_keys = !equijoin_predicates.is_empty();
 
-                on.extend(equijoin_predicates);
-                let filter = residual_minus_on(non_equijoin_expr, &on);
-
-                if has_new_keys {
+                if !equijoin_predicates.is_empty() {
+                    on.extend(equijoin_predicates);
                     Ok(Transformed::yes(LogicalPlan::Join(Join {
                         left,
                         right,
                         on,
-                        filter,
+                        filter: non_equijoin_expr,
                         join_type,
                         join_constraint,
                         schema,
@@ -103,7 +99,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                         left,
                         right,
                         on,
-                        filter,
+                        filter: non_equijoin_expr,
                         join_type,
                         join_constraint,
                         schema,
@@ -126,10 +122,6 @@ fn split_eq_and_noneq_join_predicate(
     let mut accum_join_keys: Vec<(Expr, Expr)> = vec![];
     let mut accum_filters: Vec<Expr> = vec![];
     for expr in exprs {
-        if is_inferred_alias(&expr) {
-            accum_filters.push(expr);
-            continue;
-        }
         match expr {
             Expr::BinaryExpr(BinaryExpr {
                 ref left,
@@ -160,77 +152,17 @@ fn split_eq_and_noneq_join_predicate(
     Ok((accum_join_keys, result_filter))
 }
 
-fn residual_minus_on(filter: Option<Expr>, on: &[(Expr, Expr)]) -> Option<Expr> {
-    let filter = filter?;
-    let exprs = split_conjunction_owned(filter);
-
-    let mut on_set: HashSet<(String, String)> = HashSet::new();
-    for (l, r) in on {
-        let (a, b) = canonical_pair(l, r);
-        on_set.insert((a.clone(), b.clone()));
-    }
-
-    let remaining: Vec<Expr> = exprs
-        .into_iter()
-        .filter(|e| !is_self_equality(e))
-        .filter(|e| {
-            let inner = match e {
-                Expr::Alias(alias) => alias.expr.as_ref(),
-                _ => e,
-            };
-            if let Expr::BinaryExpr(BinaryExpr { left, op: Operator::Eq, right }) = inner {
-                let (a, b) = canonical_pair(left, right);
-                !on_set.contains(&(a, b))
-            } else {
-                true
-            }
-        })
-        .collect();
-
-    remaining.into_iter().reduce(Expr::and)
-}
-
-fn canonical_pair(left: &Expr, right: &Expr) -> (String, String) {
-    let l = canonical_str(left);
-    let r = canonical_str(right);
-    if l <= r { (l, r) } else { (r, l) }
-}
-
-fn canonical_str(expr: &Expr) -> String {
-    match expr {
-        Expr::Alias(alias) => canonical_str(&alias.expr),
-        _ => format!("{}", expr),
-    }
-}
-
-fn is_self_equality(expr: &Expr) -> bool {
-    let inner = match expr {
-        Expr::Alias(alias) => alias.expr.as_ref(),
-        _ => expr,
-    };
-    match inner {
-        Expr::BinaryExpr(BinaryExpr { left, op: Operator::Eq, right }) => left.as_ref() == right.as_ref(),
-        _ => false,
-    }
-}
-
-
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::assert_optimized_plan_eq_display_indent_snapshot;
     use crate::test::*;
-    use crate::{Optimizer, OptimizerContext};
     use arrow::datatypes::DataType;
-    use datafusion_common::NullEquality;
     use datafusion_expr::{
         col, lit, logical_plan::builder::LogicalPlanBuilder, JoinType,
-        logical_plan::JoinConstraint,
     };
     use std::sync::Arc;
 
-    fn observe(_plan: &LogicalPlan, _rule: &dyn OptimizerRule) {}
-
     macro_rules! assert_optimized_plan_equal {
         (
             $plan:expr,
@@ -287,31 +219,6 @@ mod tests {
         )
     }
 
-    #[test]
-    fn residual_minus_on_removes_symmetric_dup() -> Result<()> {
-        let left = test_table_scan_with_name("l")?;
-        let right = test_table_scan_with_name("r")?;
-        let on = vec![(col("l.a"), col("r.a"))];
-        let filter = Some(col("r.a").eq(col("l.a")).and(col("l.a").eq(col("l.a"))));
-        let join = Join::try_new(
-            Arc::new(left),
-            Arc::new(right),
-            on,
-            filter,
-            JoinType::Inner,
-            JoinConstraint::On,
-            NullEquality::NullEqualsNull,
-        )?;
-        let optimizer = Optimizer::with_rules(vec![Arc::new(ExtractEquijoinPredicate::new())]);
-        let optimized = optimizer.optimize(LogicalPlan::Join(join), &OptimizerContext::new(), observe)?;
-        if let LogicalPlan::Join(j) = optimized {
-            assert!(j.filter.is_none());
-        } else {
-            panic!("expected join");
-        }
-        Ok(())
-    }
-
     #[test]
     fn join_with_only_none_equi_predicate() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
