feat: null aware RightAnti hash join execution + planning support

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -441,6 +441,154 @@ async fn test_join_with_swap_semi() {
     }
 }
 
+#[rstest]
+#[case(PartitionMode::CollectLeft)]
+#[case(PartitionMode::Auto)]
+#[case(PartitionMode::Partitioned)]
+#[tokio::test]
+async fn test_null_aware_left_anti_swaps_to_right_anti(
+    #[case] partition_mode: PartitionMode,
+) -> Result<()> {
+    let (big, small) = create_big_and_small();
+    let join = HashJoinExec::try_new(
+        Arc::clone(&big),
+        Arc::clone(&small),
+        vec![(
+            Arc::new(Column::new_with_schema("big_col", &big.schema())?),
+            Arc::new(Column::new_with_schema("small_col", &small.schema())?),
+        )],
+        None,
+        &JoinType::LeftAnti,
+        None,
+        partition_mode,
+        NullEquality::NullEqualsNothing,
+        true,
+    )?;
+    let original_schema = join.schema();
+
+    let optimized_join =
+        JoinSelection::new().optimize(Arc::new(join), &ConfigOptions::new())?;
+    let swapped_join = optimized_join
+        .downcast_ref::<HashJoinExec>()
+        .expect("anti join swap should not require a projection");
+
+    assert_eq!(*swapped_join.join_type(), JoinType::RightAnti);
+    assert_eq!(*swapped_join.partition_mode(), PartitionMode::CollectLeft);
+    assert!(swapped_join.null_aware);
+    assert_eq!(swapped_join.left().schema().field(0).name(), "small_col");
+    assert_eq!(swapped_join.right().schema().field(0).name(), "big_col");
+    assert_eq!(swapped_join.schema(), original_schema);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_null_aware_auto_large_inputs_swaps_to_collect_left() -> Result<()> {
+    let bigger: Arc<dyn ExecutionPlan> = Arc::new(StatisticsExec::new(
+        bigger_statistics(),
+        Schema::new(vec![Field::new("bigger_col", DataType::Int32, false)]),
+    ));
+    let big: Arc<dyn ExecutionPlan> = Arc::new(StatisticsExec::new(
+        big_statistics(),
+        Schema::new(vec![Field::new("big_col", DataType::Int32, false)]),
+    ));
+    let join = HashJoinExec::try_new(
+        Arc::clone(&bigger),
+        Arc::clone(&big),
+        vec![(
+            Arc::new(Column::new_with_schema("bigger_col", &bigger.schema())?),
+            Arc::new(Column::new_with_schema("big_col", &big.schema())?),
+        )],
+        None,
+        &JoinType::LeftAnti,
+        None,
+        PartitionMode::Auto,
+        NullEquality::NullEqualsNothing,
+        true,
+    )?;
+
+    let optimized_join =
+        JoinSelection::new().optimize(Arc::new(join), &ConfigOptions::new())?;
+    let swapped_join = optimized_join
+        .downcast_ref::<HashJoinExec>()
+        .expect("anti join swap should not require a projection");
+
+    assert_eq!(*swapped_join.join_type(), JoinType::RightAnti);
+    assert_eq!(*swapped_join.partition_mode(), PartitionMode::CollectLeft);
+    assert!(swapped_join.null_aware);
+    assert_eq!(swapped_join.left().schema().field(0).name(), "big_col");
+    assert_eq!(swapped_join.right().schema().field(0).name(), "bigger_col");
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_null_aware_left_anti_does_not_swap_when_left_is_smaller() -> Result<()> {
+    let (big, small) = create_big_and_small();
+    let join = HashJoinExec::try_new(
+        Arc::clone(&small),
+        Arc::clone(&big),
+        vec![(
+            Arc::new(Column::new_with_schema("small_col", &small.schema())?),
+            Arc::new(Column::new_with_schema("big_col", &big.schema())?),
+        )],
+        None,
+        &JoinType::LeftAnti,
+        None,
+        PartitionMode::CollectLeft,
+        NullEquality::NullEqualsNothing,
+        true,
+    )?;
+
+    let optimized_join =
+        JoinSelection::new().optimize(Arc::new(join), &ConfigOptions::new())?;
+    let unswapped_join = optimized_join
+        .downcast_ref::<HashJoinExec>()
+        .expect("join type should remain unchanged");
+
+    assert_eq!(*unswapped_join.join_type(), JoinType::LeftAnti);
+    assert_eq!(*unswapped_join.partition_mode(), PartitionMode::CollectLeft);
+    assert!(unswapped_join.null_aware);
+    assert_eq!(unswapped_join.left().schema().field(0).name(), "small_col");
+    assert_eq!(unswapped_join.right().schema().field(0).name(), "big_col");
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_null_aware_left_anti_respects_disabled_join_reordering() -> Result<()> {
+    let (big, small) = create_big_and_small();
+    let join = HashJoinExec::try_new(
+        Arc::clone(&big),
+        Arc::clone(&small),
+        vec![(
+            Arc::new(Column::new_with_schema("big_col", &big.schema())?),
+            Arc::new(Column::new_with_schema("small_col", &small.schema())?),
+        )],
+        None,
+        &JoinType::LeftAnti,
+        None,
+        PartitionMode::CollectLeft,
+        NullEquality::NullEqualsNothing,
+        true,
+    )?;
+    let mut config = ConfigOptions::new();
+    config.optimizer.join_reordering = false;
+
+    let optimized_join = JoinSelection::new().optimize(Arc::new(join), &config)?;
+    let unswapped_join = optimized_join
+        .downcast_ref::<HashJoinExec>()
+        .expect("join type should remain unchanged");
+
+    assert_eq!(*unswapped_join.join_type(), JoinType::LeftAnti);
+    assert_eq!(*unswapped_join.partition_mode(), PartitionMode::CollectLeft);
+    assert!(unswapped_join.null_aware);
+    assert_eq!(unswapped_join.left().schema().field(0).name(), "big_col");
+    assert_eq!(unswapped_join.right().schema().field(0).name(), "small_col");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_join_with_swap_mark() {
     let join_types = [JoinType::LeftMark, JoinType::RightMark];

datafusion/physical-optimizer/src/join_selection.rs

@@ -186,6 +186,12 @@ impl PhysicalOptimizerRule for JoinSelection {
     }
 }
 
+/// Determines whether it is possible to swap inputs of a hash join - for null-aware joins, we can only swap `LeftAnti`
+fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool {
+    hash_join.join_type().supports_swap()
+        && (!hash_join.null_aware || *hash_join.join_type() == JoinType::LeftAnti)
+}
+
 /// Tries to create a [`HashJoinExec`] in [`PartitionMode::CollectLeft`] when possible.
 ///
 /// This function will first consider the given join type and check whether the
@@ -224,9 +230,8 @@ pub(crate) fn try_collect_left(
 
     match (left_can_collect, right_can_collect) {
         (true, true) => {
-            // Don't swap null-aware anti joins as they have specific side requirements
-            if hash_join.join_type().supports_swap()
-                && !hash_join.null_aware
+            // For null-aware joins, we only swap `LeftAnti` joins where the left side is > right side
+            if can_swap_hash_join(hash_join)
                 && should_swap_join_order(&**left, &**right, config, registry)?
             {
                 Ok(Some(hash_join.swap_inputs(PartitionMode::CollectLeft)?))
@@ -246,11 +251,7 @@ pub(crate) fn try_collect_left(
                 .build()?,
         ))),
         (false, true) => {
-            // Don't swap null-aware anti joins as they have specific side requirements
-            if optimizer_config.join_reordering
-                && hash_join.join_type().supports_swap()
-                && !hash_join.null_aware
-            {
+            if optimizer_config.join_reordering && can_swap_hash_join(hash_join) {
                 hash_join.swap_inputs(PartitionMode::CollectLeft).map(Some)
             } else {
                 Ok(None)
@@ -275,22 +276,20 @@ pub(crate) fn partitioned_hash_join(
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let left = hash_join.left();
     let right = hash_join.right();
-    // Don't swap null-aware anti joins as they have specific side requirements
-    if hash_join.join_type().supports_swap()
-        && !hash_join.null_aware
+    let partition_mode = if hash_join.null_aware {
+        PartitionMode::CollectLeft
+    } else {
+        PartitionMode::Partitioned
+    };
+    if can_swap_hash_join(hash_join)
         && should_swap_join_order(&**left, &**right, config, registry)?
     {
-        hash_join.swap_inputs(PartitionMode::Partitioned)
+        hash_join.swap_inputs(partition_mode)
     } else {
         // Null-aware anti joins must use CollectLeft mode because they track probe-side state
         // (probe_side_non_empty, probe_side_has_null) per-partition, but need global knowledge
         // for correct null handling. With partitioning, a partition might not see probe rows
         // even if the probe side is globally non-empty, leading to incorrect NULL row handling.
-        let partition_mode = if hash_join.null_aware {
-            PartitionMode::CollectLeft
-        } else {
-            PartitionMode::Partitioned
-        };
 
         Ok(Arc::new(
             hash_join
@@ -330,14 +329,16 @@ fn statistical_join_selection_subrule(
             PartitionMode::Partitioned => {
                 let left = hash_join.left();
                 let right = hash_join.right();
-                // Don't swap null-aware anti joins as they have specific side requirements
-                if hash_join.join_type().supports_swap()
-                    && !hash_join.null_aware
+                if can_swap_hash_join(hash_join)
                     && should_swap_join_order(&**left, &**right, config, registry)?
                 {
-                    hash_join
-                        .swap_inputs(PartitionMode::Partitioned)
-                        .map(Some)?
+                    // Null-aware RightAnti only supports CollectLeft
+                    let partition_mode = if hash_join.null_aware {
+                        PartitionMode::CollectLeft
+                    } else {
+                        PartitionMode::Partitioned
+                    };
+                    hash_join.swap_inputs(partition_mode).map(Some)?
                 } else {
                     None
                 }