Refactor utility layers and improve code compactness

Flatten single-use helper layers and tighten small utility helpers.
In push_down_filter, fold the join-input classifier and bind
predicate.column_refs() once for join inference. Shorten test-only
import surface in utils.rs and compact mapping helpers in
null_restriction.rs without changing semantics.

Author: kosiew

datafusion/optimizer/src/push_down_filter.rs

@@ -285,26 +285,22 @@ fn can_evaluate_as_join_condition(predicate: &Expr) -> Result<bool> {
     Ok(is_evaluate)
 }
 
-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())
+fn is_scalar_subquery_cross_join(join: &Join) -> bool {
+    fn classify(plan: &LogicalPlan) -> (bool, bool) {
+        match plan {
+            LogicalPlan::SubqueryAlias(subquery_alias) => {
+                let (is_scalar_aggregate, _) = classify(subquery_alias.input.as_ref());
+                (is_scalar_aggregate, true)
+            }
+            LogicalPlan::Projection(projection) => classify(projection.input.as_ref()),
+            LogicalPlan::Aggregate(aggregate) => (aggregate.group_expr.is_empty(), false),
+            _ => (false, false),
         }
-        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) =
-        classify_join_input(join.left.as_ref());
+    let (left_scalar_aggregate, left_is_derived_relation) = classify(join.left.as_ref());
     let (right_scalar_aggregate, right_is_derived_relation) =
-        classify_join_input(join.right.as_ref());
+        classify(join.right.as_ref());
     join.on.is_empty()
         && join.filter.is_none()
         && ((left_scalar_aggregate && right_is_derived_relation)
@@ -754,10 +750,11 @@ 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 &predicate.column_refs() {
+        for &col in &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

@@ -39,10 +39,7 @@ use log::{debug, trace};
 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,
-};
+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 {

datafusion/optimizer/src/utils/null_restriction.rs

@@ -36,11 +36,13 @@ pub(super) fn syntactic_restrict_null_predicate(
     predicate: &Expr,
     join_cols: &HashSet<&Column>,
 ) -> Option<bool> {
-    match syntactic_null_substitution_value(predicate, join_cols) {
-        Some(NullSubstitutionValue::Boolean(value)) => Some(!value),
-        Some(NullSubstitutionValue::Null) => Some(true),
-        Some(NullSubstitutionValue::NonNull) | None => None,
-    }
+    syntactic_null_substitution_value(predicate, join_cols).and_then(
+        |value| match value {
+            NullSubstitutionValue::Boolean(value) => Some(!value),
+            NullSubstitutionValue::Null => Some(true),
+            NullSubstitutionValue::NonNull => None,
+        },
+    )
 }
 
 fn not(value: Option<NullSubstitutionValue>) -> Option<NullSubstitutionValue> {
@@ -49,7 +51,7 @@ fn not(value: Option<NullSubstitutionValue>) -> Option<NullSubstitutionValue> {
             Some(NullSubstitutionValue::Boolean(!value))
         }
         Some(NullSubstitutionValue::Null) => Some(NullSubstitutionValue::Null),
-        Some(NullSubstitutionValue::NonNull) | None => None,
+        _ => None,
     }
 }
 
@@ -90,9 +92,7 @@ fn null_check_value(
         Some(NullSubstitutionValue::Null) => {
             Some(NullSubstitutionValue::Boolean(!is_not_null))
         }
-        Some(NullSubstitutionValue::NonNull | NullSubstitutionValue::Boolean(_)) => {
-            Some(NullSubstitutionValue::Boolean(is_not_null))
-        }
+        Some(_) => Some(NullSubstitutionValue::Boolean(is_not_null)),
         None => None,
     }
 }