This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23957-006dffce955346bb031daa0cf23075221b7d3baf in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 0a4e9305226767b336b78b631fa455dd1feec5d6 Author: Saad Tajwar <[email protected]> AuthorDate: Tue Aug 18 14:59:20 2026 +0000 feat: null aware RightAnti hash join execution + planning support (#23957) ## Which issue does this PR close? - Closes #23931 ## Rationale for this change (copied from issue): DataFusion plans NOT IN (subquery) as a null-aware anti join, but HashJoinExec only supports null_aware = true for LeftAnti with a single join key (validated in datafusion-physical-plan/src/joins/hash_join/exec.rs). Since HashJoinExec always builds on the left input, the build side of a null-aware anti join is the outer table, not the subquery. This has two costs: Memory scales with the wrong side. For SELECT ... FROM big_fact WHERE key NOT IN (SELECT k FROM small_dim), the hash table is built over the entire fact table. Memory is O(outer) when it could be O(subquery). The operator cannot be distributed. The null-aware logic coordinates three pieces of global state across probe partitions through in-process shared memory: probe_side_has_null: AtomicBool, probe_side_non_empty: AtomicBool, and the visited-build-row bitmap, with the last probe partition to finish emitting the unmatched build rows (hash_join/stream.rs). This is correct and cheap in one process, but engines that split probe partitions across processes get independent copies of all three and produce duplicated or incorrect rows. Ballista hit exactly this (https://github.com/apache/datafusion-ballista/issues/2187) and currently has to force the join into a single task (https://github.com/apache/datafusion-ballista/pull/2188), losing all parallelism. ## What changes are included in this PR? This PR adds physical execution support for null-aware `RightAnti` hash joins using the `CollectLeft` partition mode. The build side now records whether it contains a NULL join key, and the join outputs no rows when the build side contains NULL, filters probe rows with NULL keys, and outputs all probe rows when the build side is empty. For planner support, `JoinSelection` was updated to swap a null-aware `LeftAnti` join to a null-aware `RightAnti` join when statistics show that the right side is smaller with the swapped join always using the `CollectLeft` partition mode ## Are these changes tested? Yes ## Are there any user-facing changes? `NOT IN` queries may now use a null-aware `RightAnti` physical plan when the subquery is smaller. Query results remain unchanged, but these queries can use less memory and keep the outer table partitioned. --- .../tests/physical_optimizer/join_selection.rs | 186 +++++++ .../physical-optimizer/src/join_selection.rs | 49 +- .../physical-plan/src/joins/hash_join/exec.rs | 543 ++++++++++++++++++++- .../physical-plan/src/joins/hash_join/stream.rs | 101 ++-- datafusion/physical-plan/src/joins/mod.rs | 14 + .../test_files/dynamic_filter_pushdown_config.slt | 23 +- .../test_files/null_aware_anti_join.slt | 12 +- datafusion/sqllogictest/test_files/subquery.slt | 1 + 8 files changed, 852 insertions(+), 77 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index 265279a8ca..63654ae048 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -444,6 +444,192 @@ 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(()) +} + +/// A filtered null-aware `LeftAnti` (produced by decorrelating a correlated +/// `NOT IN`) must not be swapped to `RightAnti`: the swap is only valid for +/// unfiltered null-aware joins, since a filtered `RightAnti` would apply its +/// build-side NULL short-circuit before the filter. +#[tokio::test] +async fn test_null_aware_left_anti_with_filter_does_not_swap() -> 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())?), + )], + nl_join_filter(), + &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("filtered null-aware anti join should not swap"); + + assert_eq!(*unswapped_join.join_type(), JoinType::LeftAnti); + assert_eq!(*unswapped_join.partition_mode(), PartitionMode::CollectLeft); + assert!(unswapped_join.null_aware); + assert!(unswapped_join.filter().is_some()); + 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]; diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs index 42736f8205..4ce6329f32 100644 --- a/datafusion/physical-optimizer/src/join_selection.rs +++ b/datafusion/physical-optimizer/src/join_selection.rs @@ -186,6 +186,14 @@ 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` with no filters +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 + && hash_join.filter().is_none())) +} + /// 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 +232,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 +253,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 +278,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 +331,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 } diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 08d209003a..34e901b9a3 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -221,6 +221,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 { @@ -426,9 +429,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(); @@ -438,6 +446,11 @@ impl HashJoinExecBuilder { on.len() ); } + if *join_type == JoinType::RightAnti && exec.filter.is_some() { + return plan_err!( + "null_aware RightAnti join does not support a join filter" + ); + } } if preserve_properties { @@ -1504,6 +1517,7 @@ impl ExecutionPlan for HashJoinExec { enable_dynamic_filter_pushdown, Arc::clone(context.session_config().options()), self.null_equality, + self.null_aware && self.join_type == JoinType::RightAnti, array_map_created_count, )) })?, @@ -1524,6 +1538,7 @@ impl ExecutionPlan for HashJoinExec { enable_dynamic_filter_pushdown, Arc::clone(context.session_config().options()), self.null_equality, + false, array_map_created_count, )) } @@ -2238,6 +2253,7 @@ async fn collect_left_input( should_compute_dynamic_filters: bool, config: Arc<ConfigOptions>, null_equality: NullEquality, + compute_build_side_has_null: bool, array_map_created_count: Count, ) -> Result<JoinLeftData> { let schema = left_stream.schema(); @@ -2414,6 +2430,10 @@ async fn collect_left_input( bounds = None; } + let build_has_null = compute_build_side_has_null + && !left_values.is_empty() + && left_values[0].logical_null_count() > 0; + let data = JoinLeftData { map, batch, @@ -2425,6 +2445,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) @@ -2482,7 +2503,8 @@ mod tests { }; use arrow::array::{ - Date32Array, Int32Array, Int64Array, StructArray, UInt32Array, UInt64Array, + Array, ArrayRef, Date32Array, DictionaryArray, Int32Array, Int64Array, + StructArray, UInt32Array, UInt64Array, }; use arrow::buffer::NullBuffer; use arrow::datatypes::{DataType, Field}; @@ -2499,6 +2521,7 @@ mod tests { use datafusion_physical_expr::{ EquivalenceProperties, PhysicalSortExpr, RangePartitioning, SplitPoint, }; + use futures::StreamExt; use hashbrown::HashTable; use insta::{allow_duplicates, assert_snapshot}; use rstest::*; @@ -2645,6 +2668,36 @@ mod tests { TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap() } + /// Build a two-column table whose join key is a dictionary. + /// + /// `dict_values` are the dictionary entries; `keys` are all physically + /// valid indices into that dictionary. When `dict_values` contains NULL, + /// the resulting array can have `null_count() == 0` but + /// `logical_null_count() > 0`. + fn build_table_dict_key( + key_name: &str, + dict_values: Vec<Option<i32>>, + keys: Vec<i32>, + dummy_name: &str, + dummy: Vec<Option<i32>>, + ) -> Arc<dyn ExecutionPlan> { + let dict_type = + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Int32)); + let schema = Arc::new(Schema::new(vec![ + Field::new(key_name, dict_type, true), + Field::new(dummy_name, DataType::Int32, true), + ])); + let values = Int32Array::from(dict_values); + let key_array = Int32Array::from(keys); + let dict: ArrayRef = Arc::new(DictionaryArray::new(key_array, Arc::new(values))); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![dict, Arc::new(Int32Array::from(dummy))], + ) + .unwrap(); + TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap() + } + fn join( left: Arc<dyn ExecutionPlan>, right: Arc<dyn ExecutionPlan>, @@ -6871,6 +6924,337 @@ 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(()) + } + + /// Null-aware RightAnti must treat dictionary keys that point at a null + /// dictionary value as build-side NULLs, even when the key bitmap has no + /// physical nulls (`null_count() == 0` but `logical_null_count() > 0`). + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_build_logical_null( + batch_size: usize, + ) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + // Dictionary values: [1, NULL, 2]; keys: [0, 1, 2] => logical [1, NULL, 2] + let left = build_table_dict_key( + "c1", + vec![Some(1), None, Some(2)], + vec![0, 1, 2], + "dummy", + vec![Some(100), Some(200), Some(300)], + ); + let left_key = left + .execute(0, Arc::clone(&task_ctx))? + .next() + .await + .unwrap()?; + assert_eq!(left_key.column(0).null_count(), 0); + assert_eq!(left_key.column(0).logical_null_count(), 1); + + let right = build_table_dict_key( + "c2", + vec![Some(1), Some(2), Some(3), Some(4)], + vec![0, 1, 2, 3], + "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(()) + } + + /// Null-aware RightAnti must drop outer rows whose dictionary key is only + /// logically NULL (key points at a null dictionary value). + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_right_anti_probe_logical_null( + batch_size: usize, + ) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + let left = build_table_dict_key( + "c1", + vec![Some(1), Some(2), Some(3)], + vec![0, 1, 2], + "dummy", + vec![Some(100), Some(200), Some(300)], + ); + + // Dictionary values: [1, NULL, 4]; keys: [0, 1, 2] => logical [1, NULL, 4] + let right = build_table_dict_key( + "c2", + vec![Some(1), None, Some(4)], + vec![0, 1, 2], + "dummy", + vec![Some(10), Some(0), Some(40)], + ); + let right_key = right + .execute(0, Arc::clone(&task_ctx))? + .next() + .await + .unwrap()?; + assert_eq!(right_key.column(0).null_count(), 0); + assert_eq!(right_key.column(0).logical_null_count(), 1); + + 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 logically NULL c2) + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+-------+ + | c2 | dummy | + +----+-------+ + | 4 | 40 | + +----+-------+ + "); + } + Ok(()) + } + /// Test that null_aware validation rejects non-LeftAnti join types #[tokio::test] async fn test_null_aware_validation_wrong_join_type() { @@ -6898,12 +7282,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 @@ -6946,6 +7358,121 @@ mod tests { ); } + /// A null-aware `RightAnti` short-circuits on the build-side NULL before any + /// filter runs, so the combination must be rejected at construction (and thus + /// via protobuf decoding, which routes through the same builder). A filtered + /// null-aware `LeftAnti`, by contrast, is a valid decorrelated correlated + /// `NOT IN` plan and must still be accepted. + #[tokio::test] + async fn test_null_aware_filter_rejected_only_for_right_anti() { + let make_inputs = || { + 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 _, + )]; + (left, right, on) + }; + + // RightAnti + filter is rejected. + let (left, right, on) = make_inputs(); + let result = HashJoinExec::try_new( + left, + right, + on, + Some(prepare_join_filter()), + &JoinType::RightAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + ); + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .to_string() + .contains("null_aware RightAnti join does not support a join filter") + ); + + // LeftAnti + filter is accepted (correlated NOT IN decorrelates to this). + let (left, right, on) = make_inputs(); + let result = HashJoinExec::try_new( + left, + right, + on, + Some(prepare_join_filter()), + &JoinType::LeftAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + ); + assert!(result.is_ok()); + } + + #[apply(hash_join_exec_configs)] + #[tokio::test] + async fn test_null_aware_left_anti_probe_logical_null( + batch_size: usize, + ) -> Result<()> { + let task_ctx = prepare_task_ctx(batch_size, false); + + let left = build_table_dict_key( + "c1", + vec![Some(1), Some(2), Some(3)], + vec![0, 1, 2], + "dummy", + vec![Some(100), Some(200), Some(300)], + ); + + let right = build_table_dict_key( + "c2", + vec![Some(1), None, Some(4)], + vec![0, 1, 2], + "dummy", + vec![Some(10), Some(0), Some(40)], + ); + let right_key = right + .execute(0, Arc::clone(&task_ctx))? + .next() + .await + .unwrap()?; + assert_eq!(right_key.column(0).null_count(), 0); + assert_eq!(right_key.column(0).logical_null_count(), 1); + + 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::LeftAnti, + 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] fn test_lr_is_preserved() { assert_eq!(lr_is_preserved(JoinType::Inner), (true, true)); diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 686939537e..3bb2d6d989 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -750,40 +750,63 @@ 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]; + let probe_has_null = if self.filter.is_some() { + probe_key_column.null_count() > 0 + } else { + probe_key_column.logical_null_count() > 0 + }; + if probe_has_null { + // 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); + } + } + _ => {} } } @@ -902,7 +925,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, @@ -910,6 +933,26 @@ 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]; + // if the valid_keys mask is available, use that, else use the probe key + let mask = state + .valid_keys + .clone() + .or_else(|| probe_key.logical_nulls()); + // we only need this copy if there are NULLs + if let Some(mask) = mask.filter(|m| m.null_count() > 0) { + let filtered_right_indices = right_indices + .values() + .iter() + .copied() + .filter(|idx| !mask.is_null(*idx as usize)) + .collect::<Vec<_>>(); + 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 e4f7e2e123..fc16571b79 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 { diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index eec6e5ae17..41d259e88c 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -366,8 +366,8 @@ physical_plan statement count 0 RESET datafusion.execution.parquet.pushdown_filters; -# LEFT ANTI JOIN: both self generated and parent filters can push to the -# preserved (left/build) side. +# NULL-AWARE ANTI JOIN: JoinSelection swaps the physical join to RightAnti so +# the subquery is the build side and the preserved outer input is the probe side. query TT EXPLAIN SELECT l.* FROM left_parquet l @@ -381,9 +381,9 @@ logical_plan 05)----SubqueryAlias: r 06)------TableScan: right_parquet projection=[id] physical_plan -01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet -03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet +01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(id@0, id@0)], null_aware +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet # LEFT MARK JOIN: the OR prevents decorrelation to LeftSemi, so the optimizer # uses LeftMark. Self-generated dynamic filter pushes to the probe side. @@ -457,9 +457,10 @@ ORDER BY l.id LIMIT 2; 1 left1 3 left3 -# ANTI JOIN with TopK parent: the TopK dynamic filter on `id` is pushed only -# to the preserved output side. Filtering the non-output side can create -# anti-join output. +# ANTI JOIN with TopK parent: JoinSelection swaps the physical join to +# RightAnti (build = subquery, probe = preserved outer input). The TopK dynamic +# filter on `id` is pushed only to the preserved output (probe) side; filtering +# the non-output build side can create anti-join output. query TT EXPLAIN SELECT l.* FROM left_parquet l @@ -476,9 +477,9 @@ logical_plan 07)--------TableScan: right_parquet projection=[id] physical_plan 01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false] -02)--HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware -03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible -04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet +02)--HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(id@0, id@0)], null_aware +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet +04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible # Correctness check query IT diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index bdb56cf220..4beab6998d 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -48,7 +48,8 @@ SELECT * FROM outer_table WHERE id NOT IN (SELECT id FROM inner_table_no_null); 1 a 3 c -# Verify the plan uses LeftAnti join +# The logical plan remains LeftAnti, while JoinSelection swaps the physical +# join to RightAnti so the smaller subquery is the build side. query TT EXPLAIN SELECT * FROM outer_table WHERE id NOT IN (SELECT id FROM inner_table_no_null); ---- @@ -58,7 +59,7 @@ logical_plan 03)--SubqueryAlias: __correlated_sq_1 04)----TableScan: inner_table_no_null projection=[id] physical_plan -01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware +01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(id@0, id@0)], null_aware 02)--DataSourceExec: partitions=1, partition_sizes=[1] 03)--DataSourceExec: partitions=1, partition_sizes=[1] @@ -198,11 +199,10 @@ WHERE id NOT IN (SELECT id FROM inner_table_with_null WHERE value = 'x'); # so this behaves like regular anti join (not null-aware) ############# -## Test 11: Verify NULL-aware flag is set for LeftAnti joins +## Test 11: Verify the null-aware LeftAnti logical join is physically swapped ############# -# Check that the physical plan shows null-aware anti join -# Note: The exact format may vary, but we should see LeftAnti join type +# Check that JoinSelection builds on the smaller subquery using RightAnti. query TT EXPLAIN SELECT * FROM outer_table WHERE id NOT IN (SELECT id FROM inner_table_with_null); ---- @@ -212,7 +212,7 @@ logical_plan 03)--SubqueryAlias: __correlated_sq_1 04)----TableScan: inner_table_with_null projection=[id] physical_plan -01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware +01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(id@0, id@0)], null_aware 02)--DataSourceExec: partitions=1, partition_sizes=[1] 03)--DataSourceExec: partitions=1, partition_sizes=[1] diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 594964778b..c38fb4c7ee 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -2626,6 +2626,7 @@ SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null) ORDER # The null-aware anti join must be planned as a CollectLeft HashJoinExec even with # prefer_hash_join = false: SortMergeJoinExec is not null-aware and must not be used. +# This test has no usable size statistics, so JoinSelection keeps LeftAnti. query TT EXPLAIN SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null); ---- --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
