Skip to content

fix: Omit NULL values from build side of hash joins #22893

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

Merged
kosiew merged 5 commits into from
Jun 18, 2026
Merged

fix: Omit NULL values from build side of hash joins #22893

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e547fe5
.
neilconway Jun 10, 2026
b3a1e3c
Fix CI failure
neilconway Jun 11, 2026
8df38ed
Cleanup per code review
neilconway Jun 17, 2026
2891b1c
Improve early-exit for all-NULL build side, per review
neilconway Jun 17, 2026
0b26bd7
Make test coverage tighter
neilconway Jun 17, 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
150 changes: 150 additions & 0 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 @@ -2015,6 +2015,7 @@ async fn collect_left_input(
&mut hashes_buffer,
0,
true,
null_equality,
)?;
offset += batch.num_rows();
}
Expand Down Expand Up @@ -3329,6 +3330,153 @@ mod tests {
Ok(())
}

/// Under NullEqualsNothing, NULL join keys are not inserted into the hash
/// map, so a build side whose keys are all NULL produces an empty map even
/// though it contains rows. Join types that emit unmatched build rows must
/// still produce them from the visited bitmap.
#[rstest]
#[tokio::test]
async fn join_all_null_build_keys(
#[values(PartitionMode::CollectLeft, PartitionMode::Partitioned)]
partition_mode: PartitionMode,
) -> Result<()> {
let left = build_table_two_cols(
("a1", &vec![Some(1), Some(2)]),
("b1", &vec![None, None]), // all build-side join keys are NULL
);
let right = build_table_two_cols(
("a2", &vec![Some(10), Some(20), Some(30)]),
("b1", &vec![Some(4), None, Some(6)]),
);
let on = vec![(
Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
)];

for join_type in [
JoinType::Inner,
JoinType::Left,
JoinType::Right,
JoinType::Full,
JoinType::LeftSemi,
JoinType::LeftAnti,
JoinType::RightSemi,
JoinType::RightAnti,
JoinType::LeftMark,
JoinType::RightMark,
] {
let (_, batches, _) = join_collect_with_partition_mode(
Arc::clone(&left),
Arc::clone(&right),
on.clone(),
&join_type,
partition_mode,
NullEquality::NullEqualsNothing,
Arc::new(TaskContext::default()),
)
.await?;

match join_type {
JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi => {
let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(num_rows, 0, "unexpected rows for {join_type}");
}
JoinType::Left => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+----+----+
| a1 | b1 | a2 | b1 |
+----+----+----+----+
| 1 | | | |
| 2 | | | |
+----+----+----+----+
");
}
}
JoinType::Right => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+----+----+
| a1 | b1 | a2 | b1 |
+----+----+----+----+
| | | 10 | 4 |
| | | 20 | |
| | | 30 | 6 |
+----+----+----+----+
");
}
}
JoinType::Full => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+----+----+
| a1 | b1 | a2 | b1 |
+----+----+----+----+
| | | 10 | 4 |
| | | 20 | |
| | | 30 | 6 |
| 1 | | | |
| 2 | | | |
+----+----+----+----+
");
}
}
JoinType::LeftAnti => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+
| a1 | b1 |
+----+----+
| 1 | |
| 2 | |
+----+----+
");
}
}
JoinType::RightAnti => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+
| a2 | b1 |
+----+----+
| 10 | 4 |
| 20 | |
| 30 | 6 |
+----+----+
");
}
}
JoinType::LeftMark => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+-------+
| a1 | b1 | mark |
+----+----+-------+
| 1 | | false |
| 2 | | false |
+----+----+-------+
");
}
}
JoinType::RightMark => {
allow_duplicates! {
assert_snapshot!(batches_to_sort_string(&batches), @r"
+----+----+-------+
| a2 | b1 | mark |
+----+----+-------+
| 10 | 4 | false |
| 20 | | false |
| 30 | 6 | false |
+----+----+-------+
");
}
}
}
}

Ok(())
}

#[apply(hash_join_exec_configs)]
#[tokio::test]
async fn partitioned_join_left_one(
Expand Down Expand Up @@ -4459,6 +4607,7 @@ mod tests {
&[right_keys_values],
NullEquality::NullEqualsNothing,
&hashes_buffer,
None,
8192,
(0, None),
&mut probe_indices_buffer,
Expand Down Expand Up @@ -4520,6 +4669,7 @@ mod tests {
&[right_keys_values],
NullEquality::NullEqualsNothing,
&hashes_buffer,
None,
8192,
(0, None),
&mut probe_indices_buffer,
Expand Down
20 changes: 13 additions & 7 deletions datafusion/physical-plan/src/joins/hash_join/stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ use crate::joins::hash_join::shared_bounds::{
PartitionBounds, PartitionBuildData, SharedBuildAccumulator,
};
use crate::joins::utils::{
OnceFut, equal_rows_arr, get_final_indices_from_shared_bitmap,
OnceFut, equal_rows_arr, get_final_indices_from_shared_bitmap, matchable_join_keys,
};
use crate::stream::EmptyRecordBatchStream;
use crate::{
Expand All @@ -48,6 +48,7 @@ use crate::{
};

use arrow::array::{Array, ArrayRef, UInt32Array, UInt64Array};
use arrow::buffer::NullBuffer;
use arrow::datatypes::{Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::{
Expand Down Expand Up @@ -156,6 +157,10 @@ pub(super) struct ProcessProbeBatchState {
batch: RecordBatch,
/// Probe-side on expressions values
values: Vec<ArrayRef>,
/// Combined validity of the probe-side key columns, set when NULL keys
/// exist and cannot match (`NullEquality::NullEqualsNothing`); NULL rows
/// are skipped during JoinHashMap lookups
valid_keys: Option<NullBuffer>,
/// Starting offset for JoinHashMap lookups
offset: MapOffset,
/// Max joined probe-side index from current batch
Expand Down Expand Up @@ -394,13 +399,15 @@ pub(super) fn lookup_join_hashmap(
probe_side_values: &[ArrayRef],
null_equality: NullEquality,
hashes_buffer: &[u64],
valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
probe_indices_buffer: &mut Vec<u32>,
build_indices_buffer: &mut Vec<u64>,
) -> Result<(UInt64Array, UInt32Array, Option<MapOffset>)> {
let next_offset = build_hashmap.get_matched_indices_with_limit_offset(
hashes_buffer,
valid_keys,
limit,
offset,
probe_indices_buffer,
Expand Down Expand Up @@ -516,7 +523,7 @@ impl HashJoinStream {
join_type: JoinType,
left_data: &JoinLeftData,
) -> HashJoinStreamState {
if left_data.map().is_empty()
if left_data.batch().num_rows() == 0
Comment thread
neilconway marked this conversation as resolved.
Outdated
&& join_type.empty_build_side_produces_empty_result()
{
HashJoinStreamState::Completed
Expand Down Expand Up @@ -679,6 +686,7 @@ impl HashJoinStream {
// Precalculate hash values for fetched batch
let keys_values = evaluate_expressions_to_arrays(&self.on_right, &batch)?;

let mut valid_keys = None;
Comment thread
neilconway marked this conversation as resolved.
Outdated
if let Map::HashMap(_) = self.build_side.try_as_ready()?.left_data.map() {
self.hashes_buffer.clear();
self.hashes_buffer.resize(batch.num_rows(), 0);
Expand All @@ -687,6 +695,7 @@ impl HashJoinStream {
&self.random_state,
&mut self.hashes_buffer,
)?;
valid_keys = matchable_join_keys(&keys_values, self.null_equality);
}

self.join_metrics.input_batches.add(1);
Expand All @@ -696,6 +705,7 @@ impl HashJoinStream {
HashJoinStreamState::ProcessProbeBatch(ProcessProbeBatchState {
batch,
values: keys_values,
valid_keys,
offset: (0, None),
joined_probe_idx: None,
});
Expand Down Expand Up @@ -759,14 +769,9 @@ impl HashJoinStream {
}
}

// If the build side is empty, this stream only reaches ProcessProbeBatch for
// join types whose output still depends on probe rows.
let is_empty = build_side.left_data.map().is_empty();

if is_empty {
// Invariant: state_after_build_ready should have already completed
// join types whose result is fixed to empty when the build side is empty.
debug_assert!(!self.join_type.empty_build_side_produces_empty_result());
let result = build_batch_empty_build_side(
&self.schema,
build_side.left_data.batch(),
Expand All @@ -790,6 +795,7 @@ impl HashJoinStream {
&state.values,
self.null_equality,
&self.hashes_buffer,
state.valid_keys.as_ref(),
self.batch_size,
state.offset,
&mut self.probe_indices_buffer,
Expand Down
Loading
Loading