From 54be73d074d66ca435180097af3afee87b8c95b0 Mon Sep 17 00:00:00 2001 From: saadtajwar Date: Tue, 28 Jul 2026 18:43:20 -0400 Subject: [PATCH 1/4] feat: supporting null-aware RightAnti hash join for NOT IN --- .../physical-plan/src/joins/hash_join/exec.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index ccdb050d168e0..0f618d501f1f2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -217,6 +217,11 @@ pub(super) struct JoinLeftData { pub(super) probe_side_non_empty: AtomicBool, /// Shared atomic flag indicating if any probe partition saw NULL in join keys (for null-aware anti joins) pub(super) probe_side_has_null: AtomicBool, + + // For RightAnti joins, where the build side is a smaller subquery, truthy if empty + build_is_empty: bool, + // For RightAnti joins, where the build side is a smaller subquery, truthy if has null for tjhe single join key + build_has_null: bool, } impl JoinLeftData { @@ -422,9 +427,14 @@ impl HashJoinExecBuilder { // Validate null_aware flag if exec.null_aware { let join_type = exec.join_type(); - if !matches!(join_type, JoinType::LeftAnti) { + let partition_mode = exec.partition_mode(); + if !matches!( + (join_type, partition_mode), + (JoinType::LeftAnti, _) + | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RigthAnti` is probe-driven + ) { return plan_err!( - "null_aware can only be true for LeftAnti joins, got {join_type}" + "null_aware can only be true for LeftAnti and RightAnti joins, got {join_type}" ); } let on = exec.on(); From be053ca6deee9df47fec3135d156deeecb13b907 Mon Sep 17 00:00:00 2001 From: saadtajwar Date: Tue, 28 Jul 2026 19:52:32 -0400 Subject: [PATCH 2/4] modifying hash join streaming for changes --- .../physical-plan/src/joins/hash_join/exec.rs | 11 +-- .../src/joins/hash_join/stream.rs | 89 +++++++++++++------ 2 files changed, 66 insertions(+), 34 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 0f618d501f1f2..24ee28c0bd1bf 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -218,10 +218,8 @@ pub(super) struct JoinLeftData { /// Shared atomic flag indicating if any probe partition saw NULL in join keys (for null-aware anti joins) pub(super) probe_side_has_null: AtomicBool, - // For RightAnti joins, where the build side is a smaller subquery, truthy if empty - build_is_empty: bool, - // For RightAnti joins, where the build side is a smaller subquery, truthy if has null for tjhe single join key - build_has_null: bool, + // For RightAnti joins, where the build side is a smaller subquery, truthy if has null for the single join key + pub(super) build_side_has_null: bool, } impl JoinLeftData { @@ -434,7 +432,7 @@ impl HashJoinExecBuilder { | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RigthAnti` is probe-driven ) { return plan_err!( - "null_aware can only be true for LeftAnti and RightAnti joins, got {join_type}" + "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, got {join_type} with {partition_mode}" ); } let on = exec.on(); @@ -2334,6 +2332,8 @@ async fn collect_left_input( bounds = None; } + let build_has_null = left_values[0].null_count() == 0; + let data = JoinLeftData { map, batch, @@ -2345,6 +2345,7 @@ async fn collect_left_input( membership, probe_side_non_empty: AtomicBool::new(false), probe_side_has_null: AtomicBool::new(false), + build_side_has_null: build_has_null, }; Ok(data) diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 2aa6e69dff807..45d39e8748302 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -54,6 +54,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::{ JoinSide, JoinType, NullEquality, Result, internal_datafusion_err, internal_err, }; +use datafusion_expr::JoinType::RightAnti; use datafusion_physical_expr::PhysicalExprRef; use datafusion_common::hash_utils::RandomState; @@ -742,40 +743,58 @@ impl HashJoinStream { let timer = self.join_metrics.join_time.timer(); // Null-aware anti join semantics: + // For LeftAnti: output LEFT (build) rows where LEFT.key NOT IN RIGHT.key // 1. If RIGHT (probe) contains NULL in any batch, no LEFT rows should be output // 2. LEFT rows with NULL keys should not be output (handled in final stage) + + // For RightAnti: output RIGHT (probe) rows where RIGHT.key NOT IN LEFT.key + // 1. If LEFT (build) contains NULL, no RIGHT rows should be output + // 2. RIGHT rows with NULL keys should not be output + // 3. If LEFT (build) is empty, all RIGHT rows should be output if self.null_aware { - // Mark that we've seen a probe batch with actual rows (probe side is non-empty) - // Only set this if batch has rows - empty batches don't count - // Use shared atomic state so all partitions can see this global information - if state.batch.num_rows() > 0 { - build_side - .left_data - .probe_side_non_empty - .store(true, Ordering::Relaxed); - } + match self.join_type { + JoinType::RightAnti => { + if build_side.left_data.build_side_has_null { + timer.done(); + self.state = HashJoinStreamState::FetchProbeBatch; + return Ok(StatefulStreamResult::Continue); + } + } + JoinType::LeftAnti => { + // Mark that we've seen a probe batch with actual rows (probe side is non-empty) + // Only set this if batch has rows - empty batches don't count + // Use shared atomic state so all partitions can see this global information + if state.batch.num_rows() > 0 { + build_side + .left_data + .probe_side_non_empty + .store(true, Ordering::Relaxed); + } - // Check if probe side (RIGHT) contains NULL - // Since null_aware validation ensures single column join, we only check the first column - let probe_key_column = &state.values[0]; - if probe_key_column.null_count() > 0 { - // Found NULL in probe side - set shared flag to prevent any output - build_side - .left_data - .probe_side_has_null - .store(true, Ordering::Relaxed); - } + // Check if probe side (RIGHT) contains NULL + // Since null_aware validation ensures single column join, we only check the first column + let probe_key_column = &state.values[0]; + if probe_key_column.null_count() > 0 { + // Found NULL in probe side - set shared flag to prevent any output + build_side + .left_data + .probe_side_has_null + .store(true, Ordering::Relaxed); + } - // If probe side has NULL (detected in this or any other partition), return empty result - if build_side - .left_data - .probe_side_has_null - .load(Ordering::Relaxed) - { - timer.done(); - self.state = HashJoinStreamState::FetchProbeBatch; - return Ok(StatefulStreamResult::Continue); + // If probe side has NULL (detected in this or any other partition), return empty result + if build_side + .left_data + .probe_side_has_null + .load(Ordering::Relaxed) + { + timer.done(); + self.state = HashJoinStreamState::FetchProbeBatch; + return Ok(StatefulStreamResult::Continue); + } + } + _ => {} } } @@ -894,7 +913,7 @@ impl HashJoinStream { last_joined_right_idx.map_or(0, |v| v + 1) }; - let (left_indices, right_indices) = adjust_indices_by_join_type( + let (left_indices, mut right_indices) = adjust_indices_by_join_type( left_indices, right_indices, index_alignment_range_start..index_alignment_range_end, @@ -902,6 +921,18 @@ impl HashJoinStream { self.right_side_ordered, )?; + // If null-aware RightAnti join, we don't want to emit NULL probe keys + if self.join_type == JoinType::RightAnti && self.null_aware { + let probe_key = &state.values[0]; + let filtered_right_indices = right_indices + .values() + .iter() + .copied() + .filter(|idx| !probe_key.is_null(*idx as usize)) + .collect::>(); + right_indices = UInt32Array::from(filtered_right_indices); + } + // Build output batch and push to coalescer let (build_batch, probe_batch, join_side) = if self.join_type == JoinType::RightMark { From d18f9b178ae105b3b953cad40c9a520061f42ed7 Mon Sep 17 00:00:00 2001 From: saadtajwar Date: Tue, 28 Jul 2026 20:02:07 -0400 Subject: [PATCH 3/4] testing added --- .../physical-plan/src/joins/hash_join/exec.rs | 242 +++++++++++++++++- .../src/joins/hash_join/stream.rs | 1 - datafusion/physical-plan/src/joins/mod.rs | 14 + 3 files changed, 250 insertions(+), 7 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 24ee28c0bd1bf..e6dea1c51327f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -2332,7 +2332,7 @@ async fn collect_left_input( bounds = None; } - let build_has_null = left_values[0].null_count() == 0; + let build_has_null = !left_values.is_empty() && left_values[0].null_count() > 0; let data = JoinLeftData { map, @@ -6765,6 +6765,208 @@ mod tests { Ok(()) } + /// Test null-aware RightAnti when build side (subquery) contains NULL + /// Expected: no rows should be output + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_build_null(batch_size: usize) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + // Build left table (subquery with NULL) + let left = build_table_two_cols( + ("c1", &vec![Some(1), Some(2), Some(3), None]), + ("dummy", &vec![Some(100), Some(200), Some(300), Some(400)]), + ); + + // Build right table (outer rows to potentially output) + let right = build_table_two_cols( + ("c2", &vec![Some(1), Some(2), Some(3), Some(4)]), + ("dummy", &vec![Some(10), Some(20), Some(30), Some(40)]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("c2", &right.schema())?) as _, + )]; + + let join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + )?; + + let stream = join.execute(0, task_ctx)?; + let batches = common::collect(stream).await?; + + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + ++ + ++ + "); + } + Ok(()) + } + + /// Test null-aware RightAnti when probe side (outer) contains NULL keys + /// Expected: rows with NULL keys should not be output + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_probe_null(batch_size: usize) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + // Build left table (subquery, no NULL) + let left = build_table_two_cols( + ("c1", &vec![Some(1), Some(2), Some(3)]), + ("dummy", &vec![Some(100), Some(200), Some(300)]), + ); + + // Build right table with NULL key (this row should not be output) + let right = build_table_two_cols( + ("c2", &vec![Some(1), Some(4), None]), + ("dummy", &vec![Some(10), Some(40), Some(0)]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("c2", &right.schema())?) as _, + )]; + + let join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + )?; + + let stream = join.execute(0, task_ctx)?; + let batches = common::collect(stream).await?; + + // Expected: only c2=4 (not c2=1 which matches, not c2=NULL) + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+-------+ + | c2 | dummy | + +----+-------+ + | 4 | 40 | + +----+-------+ + "); + } + Ok(()) + } + + /// Test null-aware RightAnti with no NULLs (should work like regular RightAnti) + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_no_nulls(batch_size: usize) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + // Build left table (subquery, no NULLs) + let left = build_table_two_cols( + ("c1", &vec![Some(1), Some(2), Some(3)]), + ("dummy", &vec![Some(100), Some(200), Some(300)]), + ); + + // Build right table (outer, no NULLs) + let right = build_table_two_cols( + ("c2", &vec![Some(1), Some(2), Some(4), Some(5)]), + ("dummy", &vec![Some(10), Some(20), Some(40), Some(50)]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("c2", &right.schema())?) as _, + )]; + + let join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + )?; + + let stream = join.execute(0, task_ctx)?; + let batches = common::collect(stream).await?; + + // Expected: c2=4 and c2=5 (they don't match anything in left) + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+-------+ + | c2 | dummy | + +----+-------+ + | 4 | 40 | + | 5 | 50 | + +----+-------+ + "); + } + Ok(()) + } + + /// Test null-aware RightAnti when build side (subquery) is empty + /// Expected: all outer rows should be output, including NULL keys + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_empty_build(batch_size: usize) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + // Build left table (empty subquery) + let left = build_table_two_cols(("c1", &vec![]), ("dummy", &vec![])); + + // Build right table (outer) + let right = build_table_two_cols( + ("c2", &vec![Some(1), None, Some(4)]), + ("dummy", &vec![Some(10), Some(0), Some(40)]), + ); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("c2", &right.schema())?) as _, + )]; + + let join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + )?; + + let stream = join.execute(0, task_ctx)?; + let batches = common::collect(stream).await?; + + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+-------+ + | c2 | dummy | + +----+-------+ + | | 0 | + | 1 | 10 | + | 4 | 40 | + +----+-------+ + "); + } + Ok(()) + } + /// Test that null_aware validation rejects non-LeftAnti join types #[tokio::test] async fn test_null_aware_validation_wrong_join_type() { @@ -6792,12 +6994,40 @@ mod tests { ); assert!(result.is_err()); - assert!( - result - .unwrap_err() - .to_string() - .contains("null_aware can only be true for LeftAnti joins") + assert!(result.unwrap_err().to_string().contains( + "null_aware can only be true for LeftAnti joins and RightAnti joins" + )); + } + + /// Test that null_aware RightAnti rejects Partitioned mode + #[tokio::test] + async fn test_null_aware_validation_right_anti_partitioned() { + let left = + build_table_two_cols(("c1", &vec![Some(1)]), ("dummy", &vec![Some(10)])); + let right = + build_table_two_cols(("c2", &vec![Some(1)]), ("dummy", &vec![Some(100)])); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema()).unwrap()) as _, + Arc::new(Column::new_with_schema("c2", &right.schema()).unwrap()) as _, + )]; + + let result = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightAnti, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + true, ); + + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains( + "null_aware can only be true for LeftAnti joins and RightAnti joins" + )); } /// Test that null_aware validation rejects multi-column joins diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 45d39e8748302..d3a78e97daedd 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -54,7 +54,6 @@ use arrow::record_batch::RecordBatch; use datafusion_common::{ JoinSide, JoinType, NullEquality, Result, internal_datafusion_err, internal_err, }; -use datafusion_expr::JoinType::RightAnti; use datafusion_physical_expr::PhysicalExprRef; use datafusion_common::hash_utils::RandomState; diff --git a/datafusion/physical-plan/src/joins/mod.rs b/datafusion/physical-plan/src/joins/mod.rs index e4f7e2e123e0e..fc16571b797e7 100644 --- a/datafusion/physical-plan/src/joins/mod.rs +++ b/datafusion/physical-plan/src/joins/mod.rs @@ -17,6 +17,9 @@ //! DataFusion Join implementations +use core::fmt; +use std::fmt::{Display, Formatter}; + use arrow::array::BooleanBufferBuilder; pub use cross_join::CrossJoinExec; use datafusion_physical_expr::PhysicalExprRef; @@ -104,6 +107,17 @@ pub enum PartitionMode { Auto, } +impl Display for PartitionMode { + fn fmt(&self, f: &mut Formatter) -> fmt::Result { + let partition_mode = match self { + PartitionMode::Partitioned => "Partitioned", + PartitionMode::CollectLeft => "CollectLeft", + PartitionMode::Auto => "Auto", + }; + write!(f, "{partition_mode}") + } +} + /// Partitioning mode to use for symmetric hash join #[derive(Hash, Clone, Copy, Debug, PartialEq, Eq)] pub enum StreamJoinPartitionMode { From eb3ba98f4a462628c5a0a934cd29034cde4fbf5a Mon Sep 17 00:00:00 2001 From: saadtajwar Date: Tue, 28 Jul 2026 20:13:55 -0400 Subject: [PATCH 4/4] spellchecker smh --- datafusion/physical-plan/src/joins/hash_join/exec.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index e6dea1c51327f..3143be7071845 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -429,7 +429,7 @@ impl HashJoinExecBuilder { if !matches!( (join_type, partition_mode), (JoinType::LeftAnti, _) - | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RigthAnti` is probe-driven + | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RightAnti` is probe-driven ) { return plan_err!( "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, got {join_type} with {partition_mode}"