Use EqualityNullBehavior instead of boolean for logical plan

Author: tschwarzinger

datafusion/common/src/equality_null_behavior.rs

@@ -0,0 +1,33 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/// Represents the null-handling behavior when evaluating equality.
+///
+/// # Order
+///
+/// The order on this type represents the "restrictiveness" of the behavior. The more restrictive
+/// a behavior is, the fewer elements are considered to be equal to `null`.
+/// [EqualityNullBehavior::NullEqualsNothing] represents the most restrictive behavior.
+///
+/// This mirrors the old order with booleans, as `false` indicated that `null != null`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
+pub enum EqualityNullBehavior {
+    /// Null is *not* equal to null while joining (`null != null`)
+    NullEqualsNothing,
+    /// Null is equal to null while joining (`null == null`)
+    NullEqualsNull,
+}

datafusion/common/src/lib.rs

@@ -41,6 +41,7 @@ pub mod config;
 pub mod cse;
 pub mod diagnostic;
 pub mod display;
+mod equality_null_behavior;
 pub mod error;
 pub mod file_options;
 pub mod format;
@@ -64,6 +65,7 @@ pub use dfschema::{
     qualified_name, DFSchema, DFSchemaRef, ExprSchema, SchemaExt, ToDFSchema,
 };
 pub use diagnostic::Diagnostic;
+pub use equality_null_behavior::EqualityNullBehavior;
 pub use error::{
     field_not_found, unqualified_field_not_found, DataFusionError, Result, SchemaError,
     SharedResult,

datafusion/core/src/physical_planner.rs

@@ -67,7 +67,7 @@ use datafusion_common::tree_node::{
 };
 use datafusion_common::{
     exec_err, internal_datafusion_err, internal_err, not_impl_err, plan_err, DFSchema,
-    ScalarValue,
+    EqualityNullBehavior, ScalarValue,
 };
 use datafusion_datasource::memory::MemorySourceConfig;
 use datafusion_expr::dml::{CopyTo, InsertOp};
@@ -895,11 +895,12 @@ impl DefaultPhysicalPlanner {
                 on: keys,
                 filter,
                 join_type,
-                null_equals_null,
+                equality_null_behavior,
                 schema: join_schema,
                 ..
             }) => {
-                let null_equals_null = *null_equals_null;
+                let null_equals_null =
+                    *equality_null_behavior == EqualityNullBehavior::NullEqualsNull;
 
                 let [physical_left, physical_right] = children.two()?;
 

datafusion/expr/src/logical_plan/builder.rs

@@ -56,7 +56,8 @@ use datafusion_common::file_options::file_type::FileType;
 use datafusion_common::{
     exec_err, get_target_functional_dependencies, internal_err, not_impl_err,
     plan_datafusion_err, plan_err, Column, Constraints, DFSchema, DFSchemaRef,
-    DataFusionError, Result, ScalarValue, TableReference, ToDFSchema, UnnestOptions,
+    DataFusionError, EqualityNullBehavior, Result, ScalarValue, TableReference,
+    ToDFSchema, UnnestOptions,
 };
 use datafusion_expr_common::type_coercion::binary::type_union_resolution;
 
@@ -903,7 +904,13 @@ impl LogicalPlanBuilder {
         join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
         filter: Option<Expr>,
     ) -> Result<Self> {
-        self.join_detailed(right, join_type, join_keys, filter, false)
+        self.join_detailed(
+            right,
+            join_type,
+            join_keys,
+            filter,
+            EqualityNullBehavior::NullEqualsNothing,
+        )
     }
 
     /// Apply a join using the specified expressions.
@@ -959,7 +966,7 @@ impl LogicalPlanBuilder {
             join_type,
             (Vec::<Column>::new(), Vec::<Column>::new()),
             filter,
-            false,
+            EqualityNullBehavior::NullEqualsNothing,
         )
     }
 
@@ -996,7 +1003,7 @@ impl LogicalPlanBuilder {
         join_type: JoinType,
         join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
         filter: Option<Expr>,
-        null_equals_null: bool,
+        equality_null_behavior: EqualityNullBehavior,
     ) -> Result<Self> {
         if join_keys.0.len() != join_keys.1.len() {
             return plan_err!("left_keys and right_keys were not the same length");
@@ -1113,7 +1120,7 @@ impl LogicalPlanBuilder {
             join_type,
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
-            null_equals_null,
+            equality_null_behavior,
         })))
     }
 
@@ -1186,7 +1193,7 @@ impl LogicalPlanBuilder {
                 filters,
                 join_type,
                 JoinConstraint::Using,
-                false,
+                EqualityNullBehavior::NullEqualsNothing,
             )?;
 
             Ok(Self::new(LogicalPlan::Join(join)))
@@ -1202,7 +1209,7 @@ impl LogicalPlanBuilder {
             None,
             JoinType::Inner,
             JoinConstraint::On,
-            false,
+            EqualityNullBehavior::NullEqualsNothing,
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))
@@ -1340,12 +1347,24 @@ impl LogicalPlanBuilder {
             .unzip();
         if is_all {
             LogicalPlanBuilder::from(left_plan)
-                .join_detailed(right_plan, join_type, join_keys, None, true)?
+                .join_detailed(
+                    right_plan,
+                    join_type,
+                    join_keys,
+                    None,
+                    EqualityNullBehavior::NullEqualsNull,
+                )?
                 .build()
         } else {
             LogicalPlanBuilder::from(left_plan)
                 .distinct()?
-                .join_detailed(right_plan, join_type, join_keys, None, true)?
+                .join_detailed(
+                    right_plan,
+                    join_type,
+                    join_keys,
+                    None,
+                    EqualityNullBehavior::NullEqualsNull,
+                )?
                 .build()
         }
     }
@@ -1423,7 +1442,7 @@ impl LogicalPlanBuilder {
             filter,
             join_type,
             JoinConstraint::On,
-            false,
+            EqualityNullBehavior::NullEqualsNothing,
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))

datafusion/expr/src/logical_plan/plan.rs

@@ -55,9 +55,9 @@ use datafusion_common::tree_node::{
 };
 use datafusion_common::{
     aggregate_functional_dependencies, internal_err, plan_err, Column, Constraints,
-    DFSchema, DFSchemaRef, DataFusionError, Dependency, FunctionalDependence,
-    FunctionalDependencies, ParamValues, Result, ScalarValue, Spans, TableReference,
-    UnnestOptions,
+    DFSchema, DFSchemaRef, DataFusionError, Dependency, EqualityNullBehavior,
+    FunctionalDependence, FunctionalDependencies, ParamValues, Result, ScalarValue,
+    Spans, TableReference, UnnestOptions,
 };
 use indexmap::IndexSet;
 
@@ -655,7 +655,7 @@ impl LogicalPlan {
                 join_constraint,
                 on,
                 schema: _,
-                null_equals_null,
+                equality_null_behavior: null_equals_null,
             }) => {
                 let schema =
                     build_join_schema(left.schema(), right.schema(), &join_type)?;
@@ -676,7 +676,7 @@ impl LogicalPlan {
                     on: new_on,
                     filter,
                     schema: DFSchemaRef::new(schema),
-                    null_equals_null,
+                    equality_null_behavior: null_equals_null,
                 }))
             }
             LogicalPlan::Subquery(_) => Ok(self),
@@ -894,7 +894,7 @@ impl LogicalPlan {
                 join_type,
                 join_constraint,
                 on,
-                null_equals_null,
+                equality_null_behavior: null_equals_null,
                 ..
             }) => {
                 let (left, right) = self.only_two_inputs(inputs)?;
@@ -933,7 +933,7 @@ impl LogicalPlan {
                     on: new_on,
                     filter: filter_expr,
                     schema: DFSchemaRef::new(schema),
-                    null_equals_null: *null_equals_null,
+                    equality_null_behavior: *null_equals_null,
                 }))
             }
             LogicalPlan::Subquery(Subquery {
@@ -3704,8 +3704,8 @@ pub struct Join {
     pub join_constraint: JoinConstraint,
     /// The output schema, containing fields from the left and right inputs
     pub schema: DFSchemaRef,
-    /// If null_equals_null is true, null == null else null != null
-    pub null_equals_null: bool,
+    /// The null handling behavior for equalities
+    pub equality_null_behavior: EqualityNullBehavior,
 }
 
 impl Join {
@@ -3734,7 +3734,7 @@ impl Join {
         filter: Option<Expr>,
         join_type: JoinType,
         join_constraint: JoinConstraint,
-        null_equals_null: bool,
+        equality_null_behavior: EqualityNullBehavior,
     ) -> Result<Self> {
         let join_schema = build_join_schema(left.schema(), right.schema(), &join_type)?;
 
@@ -3746,7 +3746,7 @@ impl Join {
             join_type,
             join_constraint,
             schema: Arc::new(join_schema),
-            null_equals_null,
+            equality_null_behavior,
         })
     }
 
@@ -3779,7 +3779,7 @@ impl Join {
             join_type: original_join.join_type,
             join_constraint: original_join.join_constraint,
             schema: Arc::new(join_schema),
-            null_equals_null: original_join.null_equals_null,
+            equality_null_behavior: original_join.equality_null_behavior,
         })
     }
 }
@@ -3801,8 +3801,8 @@ impl PartialOrd for Join {
             pub join_type: &'a JoinType,
             /// Join constraint
             pub join_constraint: &'a JoinConstraint,
-            /// If null_equals_null is true, null == null else null != null
-            pub null_equals_null: &'a bool,
+            /// The null handling behavior for equalities
+            pub equality_null_behavior: &'a EqualityNullBehavior,
         }
         let comparable_self = ComparableJoin {
             left: &self.left,
@@ -3811,7 +3811,7 @@ impl PartialOrd for Join {
             filter: &self.filter,
             join_type: &self.join_type,
             join_constraint: &self.join_constraint,
-            null_equals_null: &self.null_equals_null,
+            equality_null_behavior: &self.equality_null_behavior,
         };
         let comparable_other = ComparableJoin {
             left: &other.left,
@@ -3820,7 +3820,7 @@ impl PartialOrd for Join {
             filter: &other.filter,
             join_type: &other.join_type,
             join_constraint: &other.join_constraint,
-            null_equals_null: &other.null_equals_null,
+            equality_null_behavior: &other.equality_null_behavior,
         };
         comparable_self.partial_cmp(&comparable_other)
     }
@@ -4891,7 +4891,7 @@ mod tests {
                 join_type: JoinType::Inner,
                 join_constraint: JoinConstraint::On,
                 schema: Arc::new(left_schema.join(&right_schema)?),
-                null_equals_null: false,
+                equality_null_behavior: EqualityNullBehavior::NullEqualsNothing,
             }))
         }
 
@@ -5002,7 +5002,7 @@ mod tests {
                 Some(col("t1.b").gt(col("t2.b"))),
                 join_type,
                 JoinConstraint::On,
-                false,
+                EqualityNullBehavior::NullEqualsNothing,
             )?;
 
             match join_type {
@@ -5112,7 +5112,10 @@ mod tests {
             assert_eq!(join.filter, Some(col("t1.b").gt(col("t2.b"))));
             assert_eq!(join.join_type, join_type);
             assert_eq!(join.join_constraint, JoinConstraint::On);
-            assert!(!join.null_equals_null);
+            assert_eq!(
+                join.equality_null_behavior,
+                EqualityNullBehavior::NullEqualsNothing
+            );
         }
 
         Ok(())
@@ -5147,7 +5150,7 @@ mod tests {
                 None,
                 JoinType::Inner,
                 JoinConstraint::Using,
-                false,
+                EqualityNullBehavior::NullEqualsNothing,
             )?;
 
             let fields = join.schema.fields();
@@ -5198,7 +5201,7 @@ mod tests {
                 Some(col("t1.value").lt(col("t2.value"))), // Non-equi filter condition
                 JoinType::Inner,
                 JoinConstraint::On,
-                false,
+                EqualityNullBehavior::NullEqualsNothing,
             )?;
 
             let fields = join.schema.fields();
@@ -5247,10 +5250,13 @@ mod tests {
                 None,
                 JoinType::Inner,
                 JoinConstraint::On,
-                true,
+                EqualityNullBehavior::NullEqualsNull,
             )?;
 
-            assert!(join.null_equals_null);
+            assert_eq!(
+                join.equality_null_behavior,
+                EqualityNullBehavior::NullEqualsNull
+            );
         }
 
         Ok(())
@@ -5289,7 +5295,7 @@ mod tests {
                 Some(col("t1.value").gt(lit(5.0))),
                 join_type,
                 JoinConstraint::On,
-                false,
+                EqualityNullBehavior::NullEqualsNothing,
             )?;
 
             let fields = join.schema.fields();
@@ -5328,7 +5334,7 @@ mod tests {
             None,
             JoinType::Inner,
             JoinConstraint::Using,
-            false,
+            EqualityNullBehavior::NullEqualsNothing,
         )?;
 
         assert_eq!(