Use dedicated NullEquality enum instead of null_equals_null boolean (#16419)

* Use dedicated NullEquality enum instead of null_equals_null boolean

* Fix wrong operator mapping in hash_join

* Add an example to the documentation

Author: tobixdev

datafusion/common/src/lib.rs

@@ -46,6 +46,7 @@ pub mod file_options;
 pub mod format;
 pub mod hash_utils;
 pub mod instant;
+mod null_equality;
 pub mod parsers;
 pub mod pruning;
 pub mod rounding;
@@ -79,6 +80,7 @@ pub use functional_dependencies::{
 };
 use hashbrown::hash_map::DefaultHashBuilder;
 pub use join_type::{JoinConstraint, JoinSide, JoinType};
+pub use null_equality::NullEquality;
 pub use param_value::ParamValues;
 pub use scalar::{ScalarType, ScalarValue};
 pub use schema_reference::SchemaReference;

datafusion/common/src/null_equality.rs

@@ -0,0 +1,46 @@
+// 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 behavior for null values when evaluating equality. Currently, its primary use
+/// case is to define the behavior of joins for null values.
+///
+/// # Examples
+///
+/// The following table shows the expected equality behavior for `NullEquality`.
+///
+/// | A    | B    | NullEqualsNothing | NullEqualsNull |
+/// |------|------|-------------------|----------------|
+/// | NULL | NULL | false             | true           |
+/// | NULL | 'b'  | false             | false          |
+/// | 'a'  | NULL | false             | false          |
+/// | 'a'  | 'b'  | false             | false          |
+///
+/// # 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.
+/// [NullEquality::NullEqualsNothing] represents the most restrictive behavior.
+///
+/// This mirrors the old order with `null_equals_null` booleans, as `false` indicated that
+/// `null != null`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
+pub enum NullEquality {
+    /// Null is *not* equal to anything (`null != null`)
+    NullEqualsNothing,
+    /// Null is equal to null (`null == null`)
+    NullEqualsNull,
+}

datafusion/core/src/physical_planner.rs

@@ -901,12 +901,10 @@ impl DefaultPhysicalPlanner {
                 on: keys,
                 filter,
                 join_type,
-                null_equals_null,
+                null_equality,
                 schema: join_schema,
                 ..
             }) => {
-                let null_equals_null = *null_equals_null;
-
                 let [physical_left, physical_right] = children.two()?;
 
                 // If join has expression equijoin keys, add physical projection.
@@ -1127,7 +1125,7 @@ impl DefaultPhysicalPlanner {
                         join_filter,
                         *join_type,
                         vec![SortOptions::default(); join_on_len],
-                        null_equals_null,
+                        *null_equality,
                     )?)
                 } else if session_state.config().target_partitions() > 1
                     && session_state.config().repartition_joins()
@@ -1141,7 +1139,7 @@ impl DefaultPhysicalPlanner {
                         join_type,
                         None,
                         PartitionMode::Auto,
-                        null_equals_null,
+                        *null_equality,
                     )?)
                 } else {
                     Arc::new(HashJoinExec::try_new(
@@ -1152,7 +1150,7 @@ impl DefaultPhysicalPlanner {
                         join_type,
                         None,
                         PartitionMode::CollectLeft,
-                        null_equals_null,
+                        *null_equality,
                     )?)
                 };
 

datafusion/core/tests/execution/infinite_cancel.rs

@@ -33,7 +33,7 @@ use datafusion::physical_plan::execution_plan::Boundedness;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::{JoinType, ScalarValue};
+use datafusion_common::{JoinType, NullEquality, ScalarValue};
 use datafusion_expr_common::operator::Operator::Gt;
 use datafusion_physical_expr::expressions::{col, BinaryExpr, Column, Literal};
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
@@ -467,7 +467,7 @@ async fn test_infinite_join_cancel(
         &JoinType::Inner,
         None,
         PartitionMode::CollectLeft,
-        true,
+        NullEquality::NullEqualsNull,
     )?);
 
     // 3) Wrap yields under each infinite leaf
@@ -550,7 +550,7 @@ async fn test_infinite_join_agg_cancel(
         &JoinType::Inner,
         None,
         PartitionMode::CollectLeft,
-        true,
+        NullEquality::NullEqualsNull,
     )?);
 
     // 3) Project only one column (“value” from the left side) because we just want to sum that
@@ -714,7 +714,7 @@ async fn test_infinite_hash_join_without_repartition_and_no_agg(
         /* output64 */ None,
         // Using CollectLeft is fine—just avoid RepartitionExec’s partitioned channels.
         PartitionMode::CollectLeft,
-        /* build_left */ true,
+        /* build_left */ NullEquality::NullEqualsNull,
     )?);
 
     // 3) Do not apply InsertYieldExec—since there is no aggregation, InsertYieldExec would
@@ -796,7 +796,7 @@ async fn test_infinite_sort_merge_join_without_repartition_and_no_agg(
         /* filter */ None,
         JoinType::Inner,
         vec![SortOptions::new(true, false)], // ascending, nulls last
-        /* null_equal */ true,
+        /* null_equality */ NullEquality::NullEqualsNull,
     )?);
 
     // 3) Do not apply InsertYieldExec (no aggregation, no repartition → no built-in yields).

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -37,7 +37,7 @@ use datafusion::physical_plan::joins::{
     HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
 };
 use datafusion::prelude::{SessionConfig, SessionContext};
-use datafusion_common::ScalarValue;
+use datafusion_common::{NullEquality, ScalarValue};
 use datafusion_physical_expr::expressions::Literal;
 use datafusion_physical_expr::PhysicalExprRef;
 
@@ -504,7 +504,7 @@ impl JoinFuzzTestCase {
                 self.join_filter(),
                 self.join_type,
                 vec![SortOptions::default(); self.on_columns().len()],
-                false,
+                NullEquality::NullEqualsNothing,
             )
             .unwrap(),
         )
@@ -521,7 +521,7 @@ impl JoinFuzzTestCase {
                 &self.join_type,
                 None,
                 PartitionMode::Partitioned,
-                false,
+                NullEquality::NullEqualsNothing,
             )
             .unwrap(),
         )

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -25,8 +25,8 @@ use std::{
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::JoinSide;
 use datafusion_common::{stats::Precision, ColumnStatistics, JoinType, ScalarValue};
+use datafusion_common::{JoinSide, NullEquality};
 use datafusion_common::{Result, Statistics};
 use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
 use datafusion_expr::Operator;
@@ -222,7 +222,7 @@ async fn test_join_with_swap() {
             &JoinType::Left,
             None,
             PartitionMode::CollectLeft,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     );
@@ -284,7 +284,7 @@ async fn test_left_join_no_swap() {
             &JoinType::Left,
             None,
             PartitionMode::CollectLeft,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     );
@@ -333,7 +333,7 @@ async fn test_join_with_swap_semi() {
             &join_type,
             None,
             PartitionMode::Partitioned,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap();
 
@@ -408,7 +408,7 @@ async fn test_nested_join_swap() {
         &JoinType::Inner,
         None,
         PartitionMode::CollectLeft,
-        false,
+        NullEquality::NullEqualsNothing,
     )
     .unwrap();
     let child_schema = child_join.schema();
@@ -425,7 +425,7 @@ async fn test_nested_join_swap() {
         &JoinType::Left,
         None,
         PartitionMode::CollectLeft,
-        false,
+        NullEquality::NullEqualsNothing,
     )
     .unwrap();
 
@@ -464,7 +464,7 @@ async fn test_join_no_swap() {
             &JoinType::Inner,
             None,
             PartitionMode::CollectLeft,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     );
@@ -690,7 +690,7 @@ async fn test_hash_join_swap_on_joins_with_projections(
         &join_type,
         Some(projection),
         PartitionMode::Partitioned,
-        false,
+        NullEquality::NullEqualsNothing,
     )?);
 
     let swapped = join
@@ -851,7 +851,7 @@ fn check_join_partition_mode(
             &JoinType::Inner,
             None,
             PartitionMode::Auto,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     );
@@ -1498,7 +1498,7 @@ async fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
         &t.initial_join_type,
         None,
         t.initial_mode,
-        false,
+        NullEquality::NullEqualsNothing,
     )?) as _;
 
     let optimized_join_plan = hash_join_swap_subrule(join, &ConfigOptions::new())?;

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -25,7 +25,7 @@ use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::physical_plan::CsvSource;
 use datafusion::datasource::source::DataSourceExec;
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::{JoinSide, JoinType, Result, ScalarValue};
+use datafusion_common::{JoinSide, JoinType, NullEquality, Result, ScalarValue};
 use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
@@ -883,7 +883,7 @@ fn test_join_after_projection() -> Result<()> {
             ])),
         )),
         &JoinType::Inner,
-        true,
+        NullEquality::NullEqualsNull,
         None,
         None,
         StreamJoinPartitionMode::SinglePartition,
@@ -997,7 +997,7 @@ fn test_join_after_required_projection() -> Result<()> {
             ])),
         )),
         &JoinType::Inner,
-        true,
+        NullEquality::NullEqualsNull,
         None,
         None,
         StreamJoinPartitionMode::SinglePartition,
@@ -1158,7 +1158,7 @@ fn test_hash_join_after_projection() -> Result<()> {
         &JoinType::Inner,
         None,
         PartitionMode::Auto,
-        true,
+        NullEquality::NullEqualsNull,
     )?);
     let projection = Arc::new(ProjectionExec::try_new(
         vec![

datafusion/core/tests/physical_optimizer/replace_with_order_preserving_variants.rs

@@ -30,7 +30,7 @@ use arrow::compute::SortOptions;
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::tree_node::{TransformedResult, TreeNode};
-use datafusion_common::{assert_contains, Result};
+use datafusion_common::{assert_contains, NullEquality, Result};
 use datafusion_common::config::ConfigOptions;
 use datafusion_datasource::source::DataSourceExec;
 use datafusion_execution::TaskContext;
@@ -1171,7 +1171,7 @@ fn hash_join_exec(
             &JoinType::Inner,
             None,
             PartitionMode::Partitioned,
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     )

datafusion/core/tests/physical_optimizer/test_utils.rs

@@ -33,7 +33,7 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_common::stats::Precision;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::utils::expr::COUNT_STAR_EXPANSION;
-use datafusion_common::{ColumnStatistics, JoinType, Result, Statistics};
+use datafusion_common::{ColumnStatistics, JoinType, NullEquality, Result, Statistics};
 use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
@@ -190,7 +190,7 @@ pub fn sort_merge_join_exec(
             None,
             *join_type,
             vec![SortOptions::default(); join_on.len()],
-            false,
+            NullEquality::NullEqualsNothing,
         )
         .unwrap(),
     )
@@ -236,7 +236,7 @@ pub fn hash_join_exec(
         join_type,
         None,
         PartitionMode::Partitioned,
-        true,
+        NullEquality::NullEqualsNull,
     )?))
 }