Revert to 787d0a4

Author: kosiew

datafusion/optimizer/src/push_down_filter.rs

@@ -285,27 +285,26 @@ fn can_evaluate_as_join_condition(predicate: &Expr) -> Result<bool> {
     Ok(is_evaluate)
 }
 
-fn is_scalar_subquery_cross_join(join: &Join) -> bool {
-    fn is_scalar_aggregate_or_derived_relation(plan: &LogicalPlan) -> (bool, bool) {
-        match plan {
-            LogicalPlan::SubqueryAlias(subquery_alias) => {
-                let (is_scalar_aggregate, _) = is_scalar_aggregate_or_derived_relation(
-                    subquery_alias.input.as_ref(),
-                );
-                (is_scalar_aggregate, true)
-            }
-            LogicalPlan::Projection(projection) => {
-                is_scalar_aggregate_or_derived_relation(projection.input.as_ref())
-            }
-            LogicalPlan::Aggregate(aggregate) => (aggregate.group_expr.is_empty(), false),
-            _ => (false, false),
+fn classify_join_input(plan: &LogicalPlan) -> (bool, bool) {
+    match plan {
+        LogicalPlan::SubqueryAlias(subquery_alias) => {
+            let (is_scalar_aggregate, _) =
+                classify_join_input(subquery_alias.input.as_ref());
+            (is_scalar_aggregate, true)
+        }
+        LogicalPlan::Projection(projection) => {
+            classify_join_input(projection.input.as_ref())
         }
+        LogicalPlan::Aggregate(aggregate) => (aggregate.group_expr.is_empty(), false),
+        _ => (false, false),
     }
+}
 
+fn is_scalar_subquery_cross_join(join: &Join) -> bool {
     let (left_scalar_aggregate, left_is_derived_relation) =
-        is_scalar_aggregate_or_derived_relation(join.left.as_ref());
+        classify_join_input(join.left.as_ref());
     let (right_scalar_aggregate, right_is_derived_relation) =
-        is_scalar_aggregate_or_derived_relation(join.right.as_ref());
+        classify_join_input(join.right.as_ref());
     join.on.is_empty()
         && join.filter.is_none()
         && ((left_scalar_aggregate && right_is_derived_relation)
@@ -456,12 +455,9 @@ fn push_down_all_join(
     let keep_mixed_scalar_subquery_filters =
         is_inner_join && is_scalar_subquery_cross_join(&join);
     for predicate in predicates {
-        let left_only = left_preserved && checker.is_left_only(&predicate);
-        let right_only =
-            !left_only && right_preserved && checker.is_right_only(&predicate);
-        if left_only {
+        if left_preserved && checker.is_left_only(&predicate) {
             left_push.push(predicate);
-        } else if right_only {
+        } else if right_preserved && checker.is_right_only(&predicate) {
             right_push.push(predicate);
         } else if is_inner_join
             && !keep_mixed_scalar_subquery_filters
@@ -487,63 +483,43 @@ fn push_down_all_join(
     let mut on_filter_join_conditions = vec![];
     let (on_left_preserved, on_right_preserved) = on_lr_is_preserved(join.join_type);
 
-    for on in on_filter {
-        if on_left_preserved && checker.is_left_only(&on) {
-            left_push.push(on)
-        } else if on_right_preserved && checker.is_right_only(&on) {
-            right_push.push(on)
-        } else {
-            on_filter_join_conditions.push(on)
+    if !on_filter.is_empty() {
+        for on in on_filter {
+            if on_left_preserved && checker.is_left_only(&on) {
+                left_push.push(on)
+            } else if on_right_preserved && checker.is_right_only(&on) {
+                right_push.push(on)
+            } else {
+                on_filter_join_conditions.push(on)
+            }
         }
     }
 
     // Extract from OR clause, generate new predicates for both side of join if possible.
     // We only track the unpushable predicates above.
-    let extend_or_clauses =
-        |target: &mut Vec<Expr>, filters: &[Expr], schema: &DFSchema, preserved| {
-            if preserved {
-                target.extend(extract_or_clauses_for_join(filters, schema));
-            }
-        };
-    extend_or_clauses(
-        &mut left_push,
-        &keep_predicates,
-        left_schema,
-        left_preserved,
-    );
-    extend_or_clauses(
-        &mut left_push,
-        &join_conditions,
-        left_schema,
-        left_preserved,
-    );
-    extend_or_clauses(
-        &mut right_push,
-        &keep_predicates,
-        right_schema,
-        right_preserved,
-    );
-    extend_or_clauses(
-        &mut right_push,
-        &join_conditions,
-        right_schema,
-        right_preserved,
-    );
+    if left_preserved {
+        left_push.extend(extract_or_clauses_for_join(&keep_predicates, left_schema));
+        left_push.extend(extract_or_clauses_for_join(&join_conditions, left_schema));
+    }
+    if right_preserved {
+        right_push.extend(extract_or_clauses_for_join(&keep_predicates, right_schema));
+        right_push.extend(extract_or_clauses_for_join(&join_conditions, right_schema));
+    }
 
     // For predicates from join filter, we should check with if a join side is preserved
     // in term of join filtering.
-    extend_or_clauses(
-        &mut left_push,
-        &on_filter_join_conditions,
-        left_schema,
-        on_left_preserved,
-    );
-    extend_or_clauses(
-        &mut right_push,
-        &on_filter_join_conditions,
-        right_schema,
-        on_right_preserved,
-    );
+    if on_left_preserved {
+        left_push.extend(extract_or_clauses_for_join(
+            &on_filter_join_conditions,
+            left_schema,
+        ));
+    }
+    if on_right_preserved {
+        right_push.extend(extract_or_clauses_for_join(
+            &on_filter_join_conditions,
+            right_schema,
+        ));
+    }
 
     if let Some(predicate) = conjunction(left_push) {
         join.left = Arc::new(LogicalPlan::Filter(Filter::try_new(predicate, join.left)?));
@@ -778,11 +754,10 @@ fn infer_join_predicates_impl<
     inferred_predicates: &mut InferredPredicates,
 ) -> Result<()> {
     for predicate in input_predicates {
-        let column_refs = predicate.column_refs();
         let mut join_cols_to_replace = HashMap::new();
         let mut saw_non_replaceable_ref = false;
 
-        for &col in &column_refs {
+        for &col in &predicate.column_refs() {
             let replacement = join_col_keys.iter().find_map(|(l, r)| {
                 if ENABLE_LEFT_TO_RIGHT && col == *l {
                     Some((col, *r))

datafusion/optimizer/src/utils.rs

@@ -38,6 +38,12 @@ use log::{debug, trace};
 /// as it was initially placed here and then moved elsewhere.
 pub use datafusion_expr::expr_rewriter::NamePreserver;
 
+#[cfg(test)]
+use self::test_eval_mode::{
+    NullRestrictionEvalMode, null_restriction_eval_mode,
+    set_null_restriction_eval_mode_for_test, with_null_restriction_eval_mode_for_test,
+};
+
 /// Returns true if `expr` contains all columns in `schema_cols`
 pub(crate) fn has_all_column_refs(expr: &Expr, schema_cols: &HashSet<Column>) -> bool {
     expr.column_refs()
@@ -73,7 +79,7 @@ pub fn is_restrict_null_predicate<'a>(
     predicate: Expr,
     join_cols_of_predicate: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    if is_column_expr(&predicate) {
+    if matches!(predicate, Expr::Column(_)) {
         return Ok(true);
     }
 
@@ -93,8 +99,8 @@ pub fn is_restrict_null_predicate<'a>(
 
     #[cfg(test)]
     if matches!(
-        test_eval_mode::null_restriction_eval_mode(),
-        test_eval_mode::NullRestrictionEvalMode::AuthoritativeOnly
+        null_restriction_eval_mode(),
+        NullRestrictionEvalMode::AuthoritativeOnly
     ) {
         return authoritative_restrict_null_predicate(predicate, join_cols);
     }
@@ -127,16 +133,16 @@ pub fn evaluates_to_null<'a>(
     predicate: Expr,
     null_columns: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    if is_column_expr(&predicate) {
+    if matches!(predicate, Expr::Column(_)) {
         return Ok(true);
     }
 
-    evaluate_with_null_columns(predicate, null_columns, |result| {
-        Ok(match result {
+    Ok(
+        match evaluate_expr_with_null_column(predicate, null_columns)? {
             ColumnarValue::Array(_) => false,
             ColumnarValue::Scalar(scalar) => scalar.is_null(),
-        })
-    })
+        },
+    )
 }
 
 fn evaluate_expr_with_null_column<'a>(
@@ -170,8 +176,8 @@ fn authoritative_restrict_null_predicate<'a>(
     predicate: Expr,
     join_cols_of_predicate: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    evaluate_with_null_columns(predicate, join_cols_of_predicate, |result| {
-        Ok(match result {
+    Ok(
+        match evaluate_expr_with_null_column(predicate, join_cols_of_predicate)? {
             ColumnarValue::Array(array) if array.len() == 1 => {
                 let boolean_array = as_boolean_array(&array)?;
                 boolean_array.is_null(0) || !boolean_array.value(0)
@@ -183,27 +189,15 @@ fn authoritative_restrict_null_predicate<'a>(
                     | ScalarValue::Boolean(Some(false))
                     | ScalarValue::Null
             ),
-        })
-    })
+        },
+    )
 }
 
 fn coerce(expr: Expr, schema: &DFSchema) -> Result<Expr> {
     let mut expr_rewrite = TypeCoercionRewriter { schema };
     expr.rewrite(&mut expr_rewrite).data()
 }
 
-fn is_column_expr(expr: &Expr) -> bool {
-    matches!(expr, Expr::Column(_))
-}
-
-fn evaluate_with_null_columns<'a, T>(
-    predicate: Expr,
-    null_columns: impl IntoIterator<Item = &'a Column>,
-    f: impl FnOnce(ColumnarValue) -> Result<T>,
-) -> Result<T> {
-    f(evaluate_expr_with_null_column(predicate, null_columns)?)
-}
-
 #[cfg(test)]
 mod test_eval_mode {
     use std::cell::Cell;
@@ -252,11 +246,6 @@ mod tests {
     use super::*;
     use std::panic::{AssertUnwindSafe, catch_unwind};
 
-    use crate::utils::test_eval_mode::{
-        NullRestrictionEvalMode, null_restriction_eval_mode,
-        set_null_restriction_eval_mode_for_test,
-        with_null_restriction_eval_mode_for_test,
-    };
     use datafusion_expr::{
         Operator, binary_expr, case, col, in_list, is_null, lit, when,
     };

datafusion/optimizer/src/utils/null_restriction.rs

@@ -39,7 +39,7 @@ pub(super) fn syntactic_restrict_null_predicate(
     match syntactic_null_substitution_value(predicate, join_cols) {
         Some(NullSubstitutionValue::Boolean(value)) => Some(!value),
         Some(NullSubstitutionValue::Null) => Some(true),
-        _ => None,
+        Some(NullSubstitutionValue::NonNull) | None => None,
     }
 }
 
@@ -49,7 +49,7 @@ fn not(value: Option<NullSubstitutionValue>) -> Option<NullSubstitutionValue> {
             Some(NullSubstitutionValue::Boolean(!value))
         }
         Some(NullSubstitutionValue::Null) => Some(NullSubstitutionValue::Null),
-        _ => None,
+        Some(NullSubstitutionValue::NonNull) | None => None,
     }
 }
 
@@ -86,10 +86,15 @@ fn null_check_value(
     value: Option<NullSubstitutionValue>,
     is_not_null: bool,
 ) -> Option<NullSubstitutionValue> {
-    value.map(|value| match value {
-        NullSubstitutionValue::Null => NullSubstitutionValue::Boolean(!is_not_null),
-        _ => NullSubstitutionValue::Boolean(is_not_null),
-    })
+    match value {
+        Some(NullSubstitutionValue::Null) => {
+            Some(NullSubstitutionValue::Boolean(!is_not_null))
+        }
+        Some(NullSubstitutionValue::NonNull | NullSubstitutionValue::Boolean(_)) => {
+            Some(NullSubstitutionValue::Boolean(is_not_null))
+        }
+        None => None,
+    }
 }
 
 fn null_if_contains_null(
@@ -101,6 +106,12 @@ fn null_if_contains_null(
         .then_some(NullSubstitutionValue::Null)
 }
 
+fn strict_null_only(
+    value: Option<NullSubstitutionValue>,
+) -> Option<NullSubstitutionValue> {
+    value.filter(|value| matches!(value, NullSubstitutionValue::Null))
+}
+
 fn syntactic_null_substitution_value(
     expr: &Expr,
     join_cols: &HashSet<&Column>,
@@ -139,17 +150,16 @@ fn syntactic_null_substitution_value(
             syntactic_null_substitution_value(between.low.as_ref(), join_cols),
             syntactic_null_substitution_value(between.high.as_ref(), join_cols),
         ]),
-        Expr::Cast(cast) => {
-            syntactic_null_substitution_value(cast.expr.as_ref(), join_cols)
-                .filter(|value| matches!(value, NullSubstitutionValue::Null))
-        }
-        Expr::TryCast(try_cast) => {
-            syntactic_null_substitution_value(try_cast.expr.as_ref(), join_cols)
-                .filter(|value| matches!(value, NullSubstitutionValue::Null))
-        }
+        Expr::Cast(cast) => strict_null_only(syntactic_null_substitution_value(
+            cast.expr.as_ref(),
+            join_cols,
+        )),
+        Expr::TryCast(try_cast) => strict_null_only(syntactic_null_substitution_value(
+            try_cast.expr.as_ref(),
+            join_cols,
+        )),
         Expr::Negative(expr) => {
-            syntactic_null_substitution_value(expr.as_ref(), join_cols)
-                .filter(|value| matches!(value, NullSubstitutionValue::Null))
+            strict_null_only(syntactic_null_substitution_value(expr.as_ref(), join_cols))
         }
         Expr::Like(like) | Expr::SimilarTo(like) => null_if_contains_null([
             syntactic_null_substitution_value(like.expr.as_ref(), join_cols),