diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index ccdb050d168e0..3143be7071845 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -217,6 +217,9 @@ 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 has null for the single join key + pub(super) build_side_has_null: bool, } impl JoinLeftData { @@ -422,9 +425,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 `RightAnti` 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 joins and RightAnti joins with `CollectLeft` `PartitionMode`, got {join_type} with {partition_mode}" ); } let on = exec.on(); @@ -2324,6 +2332,8 @@ async fn collect_left_input( bounds = None; } + let build_has_null = !left_values.is_empty() && left_values[0].null_count() > 0; + let data = JoinLeftData { map, batch, @@ -2335,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) @@ -6754,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() { @@ -6781,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 2aa6e69dff807..d3a78e97daedd 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -742,40 +742,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 +912,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 +920,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 { 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 {