feat: Add null-aware anti join support

datafusion/core/src/physical_planner.rs

@@ -1091,6 +1091,7 @@ impl DefaultPhysicalPlanner {
                 filter,
                 join_type,
                 null_equality,
+                null_aware,
                 schema: join_schema,
                 ..
             }) => {
@@ -1487,6 +1488,8 @@ impl DefaultPhysicalPlanner {
                 } else if session_state.config().target_partitions() > 1
                     && session_state.config().repartition_joins()
                     && prefer_hash_join
+                    && !*null_aware
+                // Null-aware joins must use CollectLeft
                 {
                     Arc::new(HashJoinExec::try_new(
                         physical_left,
@@ -1497,6 +1500,7 @@ impl DefaultPhysicalPlanner {
                         None,
                         PartitionMode::Auto,
                         *null_equality,
+                        *null_aware,
                     )?)
                 } else {
                     Arc::new(HashJoinExec::try_new(
@@ -1508,6 +1512,7 @@ impl DefaultPhysicalPlanner {
                         None,
                         PartitionMode::CollectLeft,
                         *null_equality,
+                        *null_aware,
                     )?)
                 };
 

datafusion/core/tests/execution/coop.rs

@@ -606,6 +606,7 @@ async fn join_yields(
         None,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNull,
+        false,
     )?);
 
     query_yields(join, session_ctx.task_ctx()).await
@@ -655,6 +656,7 @@ async fn join_agg_yields(
         None,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNull,
+        false,
     )?);
 
     // Project only one column (“value” from the left side) because we just want to sum that
@@ -720,6 +722,7 @@ async fn hash_join_yields(
         None,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNull,
+        false,
     )?);
 
     query_yields(join, session_ctx.task_ctx()).await
@@ -751,9 +754,10 @@ async fn hash_join_without_repartition_and_no_agg(
         /* filter */ None,
         &JoinType::Inner,
         /* output64 */ None,
-        // Using CollectLeft is fine—just avoid RepartitionExec’s partitioned channels.
+        // Using CollectLeft is fine—just avoid RepartitionExec's partitioned channels.
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNull,
+        false,
     )?);
 
     query_yields(join, session_ctx.task_ctx()).await

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -849,6 +849,7 @@ impl JoinFuzzTestCase {
                 None,
                 PartitionMode::Partitioned,
                 NullEquality::NullEqualsNothing,
+                false,
             )
             .unwrap(),
         )

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -233,6 +233,7 @@ async fn test_dynamic_filter_pushdown_through_hash_join_with_topk() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -354,6 +355,7 @@ async fn test_static_filter_pushdown_through_hash_join() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -418,6 +420,7 @@ async fn test_static_filter_pushdown_through_hash_join() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -981,6 +984,7 @@ async fn test_hashjoin_dynamic_filter_pushdown() {
             None,
             PartitionMode::CollectLeft,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     ) as Arc<dyn ExecutionPlan>;
@@ -1170,6 +1174,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -1363,6 +1368,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_collect_left() {
             None,
             PartitionMode::CollectLeft,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -1531,6 +1537,7 @@ async fn test_nested_hashjoin_dynamic_filter_pushdown() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -1550,6 +1557,7 @@ async fn test_nested_hashjoin_dynamic_filter_pushdown() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     ) as Arc<dyn ExecutionPlan>;
@@ -1665,6 +1673,7 @@ async fn test_hashjoin_parent_filter_pushdown() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -2771,6 +2780,7 @@ async fn test_hashjoin_dynamic_filter_all_partitions_empty() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -2899,6 +2909,7 @@ async fn test_hashjoin_dynamic_filter_with_nulls() {
             None,
             PartitionMode::CollectLeft,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -3049,6 +3060,7 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() {
             None,
             PartitionMode::Partitioned,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -3199,6 +3211,7 @@ async fn test_hashjoin_hash_table_pushdown_collect_left() {
             None,
             PartitionMode::CollectLeft,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -3333,6 +3346,7 @@ async fn test_hashjoin_hash_table_pushdown_integer_keys() {
             None,
             PartitionMode::CollectLeft,
             datafusion_common::NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -3441,6 +3455,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_is_used() {
                 None,
                 PartitionMode::CollectLeft,
                 datafusion_common::NullEquality::NullEqualsNothing,
+                false,
             )
             .unwrap(),
         ) as Arc<dyn ExecutionPlan>;

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -222,6 +222,7 @@ async fn test_join_with_swap() {
             None,
             PartitionMode::CollectLeft,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -284,6 +285,7 @@ async fn test_left_join_no_swap() {
             None,
             PartitionMode::CollectLeft,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -333,6 +335,7 @@ async fn test_join_with_swap_semi() {
             None,
             PartitionMode::Partitioned,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap();
 
@@ -388,6 +391,7 @@ async fn test_join_with_swap_mark() {
             None,
             PartitionMode::Partitioned,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap();
 
@@ -461,6 +465,7 @@ async fn test_nested_join_swap() {
         None,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNothing,
+        false,
     )
     .unwrap();
     let child_schema = child_join.schema();
@@ -478,6 +483,7 @@ async fn test_nested_join_swap() {
         None,
         PartitionMode::CollectLeft,
         NullEquality::NullEqualsNothing,
+        false,
     )
     .unwrap();
 
@@ -518,6 +524,7 @@ async fn test_join_no_swap() {
             None,
             PartitionMode::CollectLeft,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -745,6 +752,7 @@ async fn test_hash_join_swap_on_joins_with_projections(
         Some(projection),
         PartitionMode::Partitioned,
         NullEquality::NullEqualsNothing,
+        false,
     )?);
 
     let swapped = join
@@ -906,6 +914,7 @@ fn check_join_partition_mode(
             None,
             PartitionMode::Auto,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     );
@@ -1554,6 +1563,7 @@ async fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
         None,
         t.initial_mode,
         NullEquality::NullEqualsNothing,
+        false,
     )?) as _;
 
     let optimized_join_plan =

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -1284,6 +1284,7 @@ fn test_hash_join_after_projection() -> Result<()> {
         None,
         PartitionMode::Auto,
         NullEquality::NullEqualsNothing,
+        false,
     )?);
     let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
         vec![

datafusion/core/tests/physical_optimizer/replace_with_order_preserving_variants.rs

@@ -1139,6 +1139,7 @@ fn hash_join_exec(
             None,
             PartitionMode::Partitioned,
             NullEquality::NullEqualsNothing,
+            false,
         )
         .unwrap(),
     )

datafusion/core/tests/physical_optimizer/test_utils.rs

@@ -247,6 +247,7 @@ pub fn hash_join_exec(
         None,
         PartitionMode::Partitioned,
         NullEquality::NullEqualsNothing,
+        false,
     )?))
 }
 

datafusion/expr/src/logical_plan/builder.rs

@@ -1011,6 +1011,25 @@ impl LogicalPlanBuilder {
         join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
         filter: Option<Expr>,
         null_equality: NullEquality,
+    ) -> Result<Self> {
+        self.join_detailed_with_options(
+            right,
+            join_type,
+            join_keys,
+            filter,
+            null_equality,
+            false,
+        )
+    }
+
+    pub fn join_detailed_with_options(
+        self,
+        right: LogicalPlan,
+        join_type: JoinType,
+        join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
+        filter: Option<Expr>,
+        null_equality: NullEquality,
+        null_aware: bool,
     ) -> Result<Self> {
         if join_keys.0.len() != join_keys.1.len() {
             return plan_err!("left_keys and right_keys were not the same length");
@@ -1128,6 +1147,7 @@ impl LogicalPlanBuilder {
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
             null_equality,
+            null_aware,
         })))
     }
 
@@ -1201,6 +1221,7 @@ impl LogicalPlanBuilder {
                 join_type,
                 JoinConstraint::Using,
                 NullEquality::NullEqualsNothing,
+                false, // null_aware
             )?;
 
             Ok(Self::new(LogicalPlan::Join(join)))
@@ -1217,6 +1238,7 @@ impl LogicalPlanBuilder {
             JoinType::Inner,
             JoinConstraint::On,
             NullEquality::NullEqualsNothing,
+            false, // null_aware
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))
@@ -1471,6 +1493,7 @@ impl LogicalPlanBuilder {
             join_type,
             JoinConstraint::On,
             NullEquality::NullEqualsNothing,
+            false, // null_aware
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))