Skip to content

feat: null aware RightAnti hash join execution #23957

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saadtajwar wants to merge 4 commits into
base: main
Choose a base branch
from
+321 −36
Open

feat: null aware RightAnti hash join execution #23957

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
54be73d
feat: supporting null-aware RightAnti hash join for NOT IN
saadtajwar Jul 28, 2026
be053ca
modifying hash join streaming for changes
saadtajwar Jul 28, 2026
d18f9b1
testing added
saadtajwar Jul 29, 2026
eb3ba98
spellchecker smh
saadtajwar Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
255 changes: 248 additions & 7 deletions datafusion/physical-plan/src/joins/hash_join/exec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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

@saadtajwar saadtajwar Jul 29, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need the partition_mode check for CollectLeft here? Or should we just count on it being enforced/never chosen as anything but CollectLeft beforehand? I suppose it's possible to explicitly create a plan with a different partition mode if you bypass the SQL -> plan frontend and create directly?

) {
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();
Expand Down Expand Up @@ -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,
Expand All @@ -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)
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading