Refactor utility layers and simplify null handling

Tighten scalar-subquery guard, remove redundant empty-loop
scaffolding, and share more of the null-evaluation flow.
Simplify small null-restriction helpers and match arms.

Author: kosiew

datafusion/optimizer/src/push_down_filter.rs

@@ -286,21 +286,26 @@ fn can_evaluate_as_join_condition(predicate: &Expr) -> Result<bool> {
 }
 
 fn is_scalar_subquery_cross_join(join: &Join) -> bool {
-    fn classify(plan: &LogicalPlan) -> (bool, bool) {
+    fn is_scalar_aggregate_or_derived_relation(plan: &LogicalPlan) -> (bool, bool) {
         match plan {
             LogicalPlan::SubqueryAlias(subquery_alias) => {
-                let (is_scalar_aggregate, _) = classify(subquery_alias.input.as_ref());
+                let (is_scalar_aggregate, _) = is_scalar_aggregate_or_derived_relation(
+                    subquery_alias.input.as_ref(),
+                );
                 (is_scalar_aggregate, true)
             }
-            LogicalPlan::Projection(projection) => classify(projection.input.as_ref()),
+            LogicalPlan::Projection(projection) => {
+                is_scalar_aggregate_or_derived_relation(projection.input.as_ref())
+            }
             LogicalPlan::Aggregate(aggregate) => (aggregate.group_expr.is_empty(), false),
             _ => (false, false),
         }
     }
 
-    let (left_scalar_aggregate, left_is_derived_relation) = classify(join.left.as_ref());
+    let (left_scalar_aggregate, left_is_derived_relation) =
+        is_scalar_aggregate_or_derived_relation(join.left.as_ref());
     let (right_scalar_aggregate, right_is_derived_relation) =
-        classify(join.right.as_ref());
+        is_scalar_aggregate_or_derived_relation(join.right.as_ref());
     join.on.is_empty()
         && join.filter.is_none()
         && ((left_scalar_aggregate && right_is_derived_relation)
@@ -451,9 +456,12 @@ fn push_down_all_join(
     let keep_mixed_scalar_subquery_filters =
         is_inner_join && is_scalar_subquery_cross_join(&join);
     for predicate in predicates {
-        if left_preserved && checker.is_left_only(&predicate) {
+        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 {
             left_push.push(predicate);
-        } else if right_preserved && checker.is_right_only(&predicate) {
+        } else if right_only {
             right_push.push(predicate);
         } else if is_inner_join
             && !keep_mixed_scalar_subquery_filters
@@ -479,43 +487,63 @@ 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);
 
-    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)
-            }
+    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.
-    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));
-    }
+    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,
+    );
 
     // For predicates from join filter, we should check with if a join side is preserved
     // in term of join filtering.
-    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,
-        ));
-    }
+    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 let Some(predicate) = conjunction(left_push) {
         join.left = Arc::new(LogicalPlan::Filter(Filter::try_new(predicate, join.left)?));

datafusion/optimizer/src/utils.rs

@@ -38,9 +38,6 @@ 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::*;
-
 /// 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()
@@ -76,7 +73,7 @@ pub fn is_restrict_null_predicate<'a>(
     predicate: Expr,
     join_cols_of_predicate: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    if matches!(predicate, Expr::Column(_)) {
+    if is_column_expr(&predicate) {
         return Ok(true);
     }
 
@@ -96,8 +93,8 @@ pub fn is_restrict_null_predicate<'a>(
 
     #[cfg(test)]
     if matches!(
-        null_restriction_eval_mode(),
-        NullRestrictionEvalMode::AuthoritativeOnly
+        test_eval_mode::null_restriction_eval_mode(),
+        test_eval_mode::NullRestrictionEvalMode::AuthoritativeOnly
     ) {
         return authoritative_restrict_null_predicate(predicate, join_cols);
     }
@@ -130,16 +127,16 @@ pub fn evaluates_to_null<'a>(
     predicate: Expr,
     null_columns: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    if matches!(predicate, Expr::Column(_)) {
+    if is_column_expr(&predicate) {
         return Ok(true);
     }
 
-    Ok(
-        match evaluate_expr_with_null_column(predicate, null_columns)? {
+    evaluate_with_null_columns(predicate, null_columns, |result| {
+        Ok(match result {
             ColumnarValue::Array(_) => false,
             ColumnarValue::Scalar(scalar) => scalar.is_null(),
-        },
-    )
+        })
+    })
 }
 
 fn evaluate_expr_with_null_column<'a>(
@@ -173,8 +170,8 @@ fn authoritative_restrict_null_predicate<'a>(
     predicate: Expr,
     join_cols_of_predicate: impl IntoIterator<Item = &'a Column>,
 ) -> Result<bool> {
-    Ok(
-        match evaluate_expr_with_null_column(predicate, join_cols_of_predicate)? {
+    evaluate_with_null_columns(predicate, join_cols_of_predicate, |result| {
+        Ok(match result {
             ColumnarValue::Array(array) if array.len() == 1 => {
                 let boolean_array = as_boolean_array(&array)?;
                 boolean_array.is_null(0) || !boolean_array.value(0)
@@ -186,15 +183,27 @@ 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;
@@ -243,6 +252,11 @@ 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

@@ -36,13 +36,11 @@ pub(super) fn syntactic_restrict_null_predicate(
     predicate: &Expr,
     join_cols: &HashSet<&Column>,
 ) -> Option<bool> {
-    syntactic_null_substitution_value(predicate, join_cols).and_then(
-        |value| match value {
-            NullSubstitutionValue::Boolean(value) => Some(!value),
-            NullSubstitutionValue::Null => Some(true),
-            NullSubstitutionValue::NonNull => None,
-        },
-    )
+    match syntactic_null_substitution_value(predicate, join_cols) {
+        Some(NullSubstitutionValue::Boolean(value)) => Some(!value),
+        Some(NullSubstitutionValue::Null) => Some(true),
+        _ => None,
+    }
 }
 
 fn not(value: Option<NullSubstitutionValue>) -> Option<NullSubstitutionValue> {
@@ -88,13 +86,10 @@ fn null_check_value(
     value: Option<NullSubstitutionValue>,
     is_not_null: bool,
 ) -> Option<NullSubstitutionValue> {
-    match value {
-        Some(NullSubstitutionValue::Null) => {
-            Some(NullSubstitutionValue::Boolean(!is_not_null))
-        }
-        Some(_) => Some(NullSubstitutionValue::Boolean(is_not_null)),
-        None => None,
-    }
+    value.map(|value| match value {
+        NullSubstitutionValue::Null => NullSubstitutionValue::Boolean(!is_not_null),
+        _ => NullSubstitutionValue::Boolean(is_not_null),
+    })
 }
 
 fn null_if_contains_null(
@@ -106,12 +101,6 @@ 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>,
@@ -150,16 +139,17 @@ 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) => 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::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::Negative(expr) => {
-            strict_null_only(syntactic_null_substitution_value(expr.as_ref(), join_cols))
+            syntactic_null_substitution_value(expr.as_ref(), join_cols)
+                .filter(|value| matches!(value, NullSubstitutionValue::Null))
         }
         Expr::Like(like) | Expr::SimilarTo(like) => null_if_contains_null([
             syntactic_null_substitution_value(like.expr.as_ref(), join_cols),