You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/18 06:35:17 UTC
(datafusion) branch main updated: fix: Omit NULL values from build side of hash joins (#22893)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 3f4bcf15c9 fix: Omit NULL values from build side of hash joins (#22893)
3f4bcf15c9 is described below
commit 3f4bcf15c9a68972deb0f294062eb81c4b77e9c8
Author: Neil Conway <ne...@gmail.com>
AuthorDate: Thu Jun 18 02:20:30 2026 -0400
fix: Omit NULL values from build side of hash joins (#22893)
## Which issue does this PR close?
- Closes #22875
## Rationale for this change
Previously, when the build side of a hash join was backed by an
`ArrayMap`, rows with NULLs in their join keys were omitted, but they
were included for `HashMap`-backed hash joins. Under
`NullEqualsNothing`, we can safely omit rows that have a NULL in any of
their join keys, because they will never contribute to the output of the
join. Omitting NULLs reduces the size of the build-side hash table.
The previous probe behavior also resulted in searching the hash table
for probe rows with NULLs in their join keys. This was wasted work;
indeed, because all NULL build rows will end up in the same hash chain,
this could actually be very expensive for joins over NULL-heavy data
sets. For example, joining two 10k tables on all-NULL join keys took ~6
seconds (!). That drops to a few milliseconds after this PR.
## What changes are included in this PR?
* Omit build rows with one or more NULLs in their join keys from
`HashMap`
* Don't probe the map for probe rows with NULLs in their join keys
* Fix a few places that assumed that an empty build-side hash table
meant the build input was empty
* Add unit tests
## Are these changes tested?
Yes; new tests added.
## Are there any user-facing changes?
No.
---
datafusion/common/src/join_type.rs | 18 +++
.../physical-plan/src/joins/hash_join/exec.rs | 168 +++++++++++++++++++++
.../physical-plan/src/joins/hash_join/stream.rs | 36 +++--
.../physical-plan/src/joins/join_hash_map.rs | 77 +++++++++-
datafusion/physical-plan/src/joins/mod.rs | 9 ++
.../physical-plan/src/joins/stream_join_utils.rs | 3 +
.../physical-plan/src/joins/symmetric_hash_join.rs | 25 ++-
datafusion/physical-plan/src/joins/utils.rs | 114 +++++++++++++-
8 files changed, 435 insertions(+), 15 deletions(-)
diff --git a/datafusion/common/src/join_type.rs b/datafusion/common/src/join_type.rs
index d517844db4..c77a1475ed 100644
--- a/datafusion/common/src/join_type.rs
+++ b/datafusion/common/src/join_type.rs
@@ -156,6 +156,24 @@ impl JoinType {
| JoinType::RightSemi
)
}
+
+ /// Returns true when an empty build-side map necessarily produces an empty
+ /// result for this join type, even if the build side still contains rows.
+ ///
+ /// Every output row of these join types requires a matching build row, so
+ /// when the map has no matchable keys the result is empty regardless of the
+ /// probe side. Note this is a subset of
+ /// [`Self::empty_build_side_produces_empty_result`]: an empty build side
+ /// yields an empty map, but the map can also be empty when every build row
+ /// has a NULL join key under [`NullEquality::NullEqualsNothing`].
+ ///
+ /// [`NullEquality::NullEqualsNothing`]: crate::NullEquality::NullEqualsNothing
+ pub fn empty_map_produces_empty_result(self) -> bool {
+ matches!(
+ self,
+ JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi
+ )
+ }
}
impl Display for JoinType {
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 1994f9d74c..5ef767ebdd 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -2038,6 +2038,7 @@ async fn collect_left_input(
&mut hashes_buffer,
0,
true,
+ null_equality,
)?;
offset += batch.num_rows();
}
@@ -3352,6 +3353,171 @@ 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, metrics) = join_collect_with_partition_mode(
+ Arc::clone(&left),
+ Arc::clone(&right),
+ on.clone(),
+ &join_type,
+ partition_mode,
+ NullEquality::NullEqualsNothing,
+ Arc::new(TaskContext::default()),
+ )
+ .await?;
+
+ // For join types whose output requires a build-side match, an
+ // empty map guarantees an empty result, so `state_after_build_ready`
+ // completes the stream without ever fetching a probe batch (probe
+ // `input_rows` stays 0). All other join types must still scan the
+ // probe side. `input_rows` is summed across every partition.
+ let probe_rows = metrics
+ .sum_by_name("input_rows")
+ .map(|v| v.as_usize())
+ .unwrap_or(0);
+ if join_type.empty_map_produces_empty_result() {
+ assert_eq!(
+ probe_rows, 0,
+ "{join_type} should skip the probe side for an all-NULL build"
+ );
+ } else {
+ assert!(probe_rows > 0, "{join_type} must scan the probe side");
+ }
+
+ 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(
@@ -4482,6 +4648,7 @@ mod tests {
&[right_keys_values],
NullEquality::NullEqualsNothing,
&hashes_buffer,
+ None,
8192,
(0, None),
&mut probe_indices_buffer,
@@ -4543,6 +4710,7 @@ mod tests {
&[right_keys_values],
NullEquality::NullEqualsNothing,
&hashes_buffer,
+ None,
8192,
(0, None),
&mut probe_indices_buffer,
diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs
index d403fa43cd..ed605301ad 100644
--- a/datafusion/physical-plan/src/joins/hash_join/stream.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs
@@ -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::{
@@ -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::{
@@ -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
@@ -394,6 +399,7 @@ 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>,
@@ -401,6 +407,7 @@ pub(super) fn lookup_join_hashmap(
) -> 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,
@@ -516,8 +523,15 @@ impl HashJoinStream {
join_type: JoinType,
left_data: &JoinLeftData,
) -> HashJoinStreamState {
- if left_data.map().is_empty()
- && join_type.empty_build_side_produces_empty_result()
+ let build_empty = left_data.batch().num_rows() == 0;
+ // The map can be empty even when the build side has rows: under
+ // `NullEqualsNothing`, build rows with a NULL join key are omitted. For
+ // join types whose every output row requires a build match, that still
+ // guarantees an empty result, so we can skip scanning the probe side.
+ let map_empty = left_data.map().is_empty();
+
+ if (build_empty && join_type.empty_build_side_produces_empty_result())
+ || (map_empty && join_type.empty_map_produces_empty_result())
{
HashJoinStreamState::Completed
} else {
@@ -679,7 +693,9 @@ impl HashJoinStream {
// Precalculate hash values for fetched batch
let keys_values = evaluate_expressions_to_arrays(&self.on_right, &batch)?;
- if let Map::HashMap(_) = self.build_side.try_as_ready()?.left_data.map() {
+ let valid_keys = 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);
create_hashes(
@@ -687,7 +703,10 @@ impl HashJoinStream {
&self.random_state,
&mut self.hashes_buffer,
)?;
- }
+ matchable_join_keys(&keys_values, self.null_equality)
+ } else {
+ None
+ };
self.join_metrics.input_batches.add(1);
self.join_metrics.input_rows.add(batch.num_rows());
@@ -696,6 +715,7 @@ impl HashJoinStream {
HashJoinStreamState::ProcessProbeBatch(ProcessProbeBatchState {
batch,
values: keys_values,
+ valid_keys,
offset: (0, None),
joined_probe_idx: None,
});
@@ -759,14 +779,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(),
@@ -790,6 +805,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,
diff --git a/datafusion/physical-plan/src/joins/join_hash_map.rs b/datafusion/physical-plan/src/joins/join_hash_map.rs
index 8f0fb66b64..454cc916ae 100644
--- a/datafusion/physical-plan/src/joins/join_hash_map.rs
+++ b/datafusion/physical-plan/src/joins/join_hash_map.rs
@@ -23,7 +23,7 @@ use std::fmt::{self, Debug};
use std::ops::Sub;
use arrow::array::BooleanArray;
-use arrow::buffer::BooleanBuffer;
+use arrow::buffer::{BooleanBuffer, NullBuffer};
use arrow::datatypes::ArrowNativeType;
use hashbrown::HashTable;
use hashbrown::hash_table::Entry::{Occupied, Vacant};
@@ -117,9 +117,14 @@ pub trait JoinHashMapType: Send + Sync {
deleted_offset: Option<usize>,
) -> (Vec<u32>, Vec<u64>);
+ /// Probe rows marked NULL in `valid_keys` are skipped without a lookup:
+ /// their key contains a NULL, which cannot match any build row under
+ /// `NullEquality::NullEqualsNothing`. Pass `None` when every probe key is
+ /// matchable.
fn get_matched_indices_with_limit_offset(
&self,
hash_values: &[u64],
+ valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
input_indices: &mut Vec<u32>,
@@ -185,6 +190,7 @@ impl JoinHashMapType for JoinHashMapU32 {
fn get_matched_indices_with_limit_offset(
&self,
hash_values: &[u64],
+ valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
input_indices: &mut Vec<u32>,
@@ -194,6 +200,7 @@ impl JoinHashMapType for JoinHashMapU32 {
&self.map,
&self.next,
hash_values,
+ valid_keys,
limit,
offset,
input_indices,
@@ -263,6 +270,7 @@ impl JoinHashMapType for JoinHashMapU64 {
fn get_matched_indices_with_limit_offset(
&self,
hash_values: &[u64],
+ valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
input_indices: &mut Vec<u32>,
@@ -272,6 +280,7 @@ impl JoinHashMapType for JoinHashMapU64 {
&self.map,
&self.next,
hash_values,
+ valid_keys,
limit,
offset,
input_indices,
@@ -376,10 +385,12 @@ where
(input_indices, match_indices)
}
+#[expect(clippy::too_many_arguments)]
pub fn get_matched_indices_with_limit_offset<T>(
map: &HashTable<(u64, T)>,
next_chain: &[T],
hash_values: &[u64],
+ valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
input_indices: &mut Vec<u32>,
@@ -401,6 +412,10 @@ where
let start = offset.0;
let end = (start + limit).min(hash_values.len());
for (i, &hash) in hash_values[start..end].iter().enumerate() {
+ // NULL keys cannot match any build row
+ if valid_keys.is_some_and(|valid| valid.is_null(start + i)) {
+ continue;
+ }
if let Some((_, idx)) = map.find(hash, |(h, _)| hash == *h) {
input_indices.push(start as u32 + i as u32);
match_indices.push((*idx - one).into());
@@ -445,6 +460,10 @@ where
let hash_values_len = hash_values.len();
for (i, &hash) in hash_values[to_skip..].iter().enumerate() {
let row_idx = to_skip + i;
+ // NULL keys cannot match any build row
+ if valid_keys.is_some_and(|valid| valid.is_null(row_idx)) {
+ continue;
+ }
if let Some((_, idx)) = map.find(hash, |(h, _)| hash == *h) {
let idx: T = *idx;
let is_last = row_idx == hash_values_len - 1;
@@ -494,4 +513,60 @@ mod tests {
}
}
}
+
+ #[test]
+ fn test_get_matched_indices_skips_invalid_keys() {
+ let mut hash_map = JoinHashMapU32::with_capacity(3);
+ hash_map.update_from_iter(Box::new([10u64, 20u64, 30u64].iter().enumerate()), 0);
+
+ let probe_hashes = vec![10, 20, 30];
+ // The probe row for hash 20 has a NULL key and must not match.
+ let valid_keys = NullBuffer::from(vec![true, false, true]);
+
+ let mut input_indices = vec![];
+ let mut match_indices = vec![];
+ let next_offset = hash_map.get_matched_indices_with_limit_offset(
+ &probe_hashes,
+ Some(&valid_keys),
+ 8192,
+ (0, None),
+ &mut input_indices,
+ &mut match_indices,
+ );
+
+ assert_eq!(next_offset, None);
+ assert_eq!(input_indices, vec![0, 2]);
+ assert_eq!(match_indices, vec![0, 2]);
+ }
+
+ #[test]
+ fn test_get_matched_indices_skips_invalid_keys_with_duplicates() {
+ // Duplicate build keys chain multiple rows under one hash value.
+ let mut hash_map = JoinHashMapU32::with_capacity(4);
+ hash_map.update_from_iter(
+ Box::new([10u64, 20u64, 10u64, 20u64].iter().enumerate()),
+ 0,
+ );
+
+ let probe_hashes = vec![10, 20];
+ // The probe row for hash 10 has a NULL key: none of the build rows in
+ // its chain may match, while the valid probe row for hash 20 must
+ // still match its entire chain.
+ let valid_keys = NullBuffer::from(vec![false, true]);
+
+ let mut input_indices = vec![];
+ let mut match_indices = vec![];
+ let next_offset = hash_map.get_matched_indices_with_limit_offset(
+ &probe_hashes,
+ Some(&valid_keys),
+ 8192,
+ (0, None),
+ &mut input_indices,
+ &mut match_indices,
+ );
+
+ assert_eq!(next_offset, None);
+ assert_eq!(input_indices, vec![1, 1]);
+ assert_eq!(match_indices, vec![3, 1]);
+ }
}
diff --git a/datafusion/physical-plan/src/joins/mod.rs b/datafusion/physical-plan/src/joins/mod.rs
index 2cdfa1e6ac..bbb25dda65 100644
--- a/datafusion/physical-plan/src/joins/mod.rs
+++ b/datafusion/physical-plan/src/joins/mod.rs
@@ -50,6 +50,15 @@ pub mod join_hash_map;
use array_map::ArrayMap;
use utils::JoinHashMapType;
+/// The build-side map of a hash join, indexing build rows by join key.
+///
+/// Under [`NullEquality::NullEqualsNothing`], build rows with a NULL in any
+/// join key column can never match a probe row and are omitted from the map.
+/// [`Map::is_empty`] and [`Map::num_of_distinct_key`] therefore reflect the
+/// *matchable* build rows: the map can be empty even when the build side
+/// contains rows.
+///
+/// [`NullEquality::NullEqualsNothing`]: datafusion_common::NullEquality::NullEqualsNothing
pub enum Map {
HashMap(Box<dyn JoinHashMapType>),
ArrayMap(ArrayMap),
diff --git a/datafusion/physical-plan/src/joins/stream_join_utils.rs b/datafusion/physical-plan/src/joins/stream_join_utils.rs
index 571c199abb..05a56d2411 100644
--- a/datafusion/physical-plan/src/joins/stream_join_utils.rs
+++ b/datafusion/physical-plan/src/joins/stream_join_utils.rs
@@ -37,6 +37,7 @@ use arrow::array::{
ArrowPrimitiveType, BooleanArray, BooleanBufferBuilder, NativeAdapter,
PrimitiveArray, RecordBatch,
};
+use arrow::buffer::NullBuffer;
use arrow::compute::concat_batches;
use arrow::datatypes::{ArrowNativeType, Schema, SchemaRef};
use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -80,6 +81,7 @@ impl JoinHashMapType for PruningJoinHashMap {
fn get_matched_indices_with_limit_offset(
&self,
hash_values: &[u64],
+ valid_keys: Option<&NullBuffer>,
limit: usize,
offset: MapOffset,
input_indices: &mut Vec<u32>,
@@ -91,6 +93,7 @@ impl JoinHashMapType for PruningJoinHashMap {
&self.map,
&next,
hash_values,
+ valid_keys,
limit,
offset,
input_indices,
diff --git a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
index ef92964fad..a56ad1712a 100644
--- a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
+++ b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
@@ -44,7 +44,7 @@ use crate::joins::utils::{
BatchSplitter, BatchTransformer, ColumnIndex, JoinFilter, JoinHashMapType, JoinOn,
JoinOnRef, NoopBatchTransformer, StatefulStreamResult, apply_join_filter_to_indices,
build_batch_from_indices, build_join_schema, check_join_is_valid, equal_rows_arr,
- symmetric_join_output_partitioning, update_hash,
+ matchable_join_keys, symmetric_join_output_partitioning, update_hash,
};
use crate::projection::{
ProjectionExec, join_allows_pushdown, join_table_borders, new_join_children,
@@ -1113,8 +1113,20 @@ fn lookup_join_hashmap(
// (5,1)
//
// With this approach, the lexicographic order on both the probe side and the build side is preserved.
+ //
+ // Probe rows whose key contains a NULL cannot match any build row and are
+ // skipped without a map lookup.
+ let valid_keys = matchable_join_keys(&keys_values, null_equality);
let (mut matched_probe, mut matched_build) = build_hashmap.get_matched_indices(
- Box::new(hash_values.iter().enumerate().rev()),
+ Box::new(
+ hash_values
+ .iter()
+ .enumerate()
+ .filter(|(i, _)| {
+ valid_keys.as_ref().is_none_or(|valid| valid.is_valid(*i))
+ })
+ .rev(),
+ ),
deleted_offset,
);
@@ -1191,6 +1203,7 @@ impl OneSideHashJoiner {
///
/// * `batch` - The incoming [RecordBatch] to be merged with the internal input buffer
/// * `random_state` - The random state used to hash values
+ /// * `null_equality` - Null semantics to use
///
/// # Returns
///
@@ -1199,6 +1212,7 @@ impl OneSideHashJoiner {
&mut self,
batch: &RecordBatch,
random_state: &RandomState,
+ null_equality: NullEquality,
) -> Result<()> {
// Merge the incoming batch with the existing input buffer:
self.input_buffer = concat_batches(&batch.schema(), [&self.input_buffer, batch])?;
@@ -1215,6 +1229,7 @@ impl OneSideHashJoiner {
&mut self.hashes_buffer,
self.deleted_offset,
false,
+ null_equality,
)?;
Ok(())
}
@@ -1688,7 +1703,11 @@ impl<T: BatchTransformer> SymmetricHashJoinStream<T> {
probe_side_metrics.input_batches.add(1);
probe_side_metrics.input_rows.add(probe_batch.num_rows());
// Update the internal state of the hash joiner for the build side:
- probe_hash_joiner.update_internal_state(probe_batch, &self.random_state)?;
+ probe_hash_joiner.update_internal_state(
+ probe_batch,
+ &self.random_state,
+ self.null_equality,
+ )?;
// Join the two sides:
let equal_result = join_with_probe_batch(
build_hash_joiner,
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index 5687be04ad..7ecace6b0e 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -1373,7 +1373,9 @@ pub(crate) fn build_batch_from_indices(
Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
}
-/// Returns a new [RecordBatch] resulting of a join where the build/left side is empty.
+/// Returns a new [RecordBatch] for a probe batch when no probe row can find a
+/// match: the build-side map is empty, either because the build side has no
+/// rows or because none of its rows has a matchable (non-NULL) join key.
/// The resulting batch has [Schema] `schema`.
pub(crate) fn build_batch_empty_build_side(
schema: &Schema,
@@ -2104,6 +2106,9 @@ pub fn swap_join_projection(
/// `fifo_hashmap` sets the order of iteration over `batch` rows while updating hashmap,
/// which allows to keep either first (if set to true) or last (if set to false) row index
/// as a chain head for rows with equal hash values.
+///
+/// Under [`NullEquality::NullEqualsNothing`], rows with a NULL in any key
+/// column can never match a probe row, so they are not inserted into the map.
#[expect(clippy::too_many_arguments)]
pub fn update_hash(
on: &[PhysicalExprRef],
@@ -2114,6 +2119,7 @@ pub fn update_hash(
hashes_buffer: &mut [u64],
deleted_offset: usize,
fifo_hashmap: bool,
+ null_equality: NullEquality,
) -> Result<()> {
// evaluate the keys
let keys_values = evaluate_expressions_to_arrays(on, batch)?;
@@ -2124,10 +2130,14 @@ pub fn update_hash(
// For usual JoinHashmap, the implementation is void.
hash_map.extend_zero(batch.num_rows());
+ // Unmatchable NULL-key rows are filtered out below.
+ let valid_keys = matchable_join_keys(&keys_values, null_equality);
+
// Updating JoinHashMap from hash values iterator
let hash_values_iter = hash_values
.iter()
.enumerate()
+ .filter(|(i, _)| valid_keys.as_ref().is_none_or(|nulls| nulls.is_valid(*i)))
.map(|(i, val)| (i + offset, val));
if fifo_hashmap {
@@ -2139,6 +2149,31 @@ pub fn update_hash(
Ok(())
}
+/// Returns the combined validity of the join key columns `join_key_arrays`: a row
+/// is valid only if every key column is non-NULL at that row.
+///
+/// Returns `None` when no rows need to be filtered: either every row has
+/// fully non-NULL keys, or `null_equality` is
+/// [`NullEquality::NullEqualsNull`], where NULL keys are matchable.
+pub(crate) fn matchable_join_keys(
+ join_key_arrays: &[ArrayRef],
+ null_equality: NullEquality,
+) -> Option<NullBuffer> {
+ match null_equality {
+ NullEquality::NullEqualsNothing => {
+ let logical_nulls: Vec<_> = join_key_arrays
+ .iter()
+ .map(|values| values.logical_nulls())
+ .collect();
+ NullBuffer::union_many(logical_nulls.iter().map(Option::as_ref))
+ // An all-valid array can still have a validity buffer; return
+ // `None` in that case, since there is nothing to filter.
+ .filter(|nulls| nulls.null_count() > 0)
+ }
+ NullEquality::NullEqualsNull => None,
+ }
+}
+
pub(super) fn equal_rows_arr(
indices_left: &UInt64Array,
indices_right: &UInt32Array,
@@ -2463,6 +2498,83 @@ mod tests {
assert_u32_values(&result, &[2, 4, 6, 7]);
}
+ #[test]
+ fn update_hash_skips_null_keys_for_null_equals_nothing() -> Result<()> {
+ use crate::joins::join_hash_map::JoinHashMapU32;
+
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![
+ Some(1),
+ None,
+ Some(2),
+ None,
+ Some(1),
+ ]))],
+ )?;
+ let on: Vec<PhysicalExprRef> = vec![Arc::new(Column::new("a", 0))];
+ let random_state = RandomState::with_seed(42);
+ let mut hashes_buffer = vec![0; batch.num_rows()];
+ create_hashes([batch.column(0)], &random_state, &mut hashes_buffer)?;
+
+ let matched_build_indices =
+ |map: &JoinHashMapU32, hashes_buffer: &[u64]| -> Vec<u64> {
+ let mut input_indices = vec![];
+ let mut match_indices = vec![];
+ map.get_matched_indices_with_limit_offset(
+ hashes_buffer,
+ None,
+ 8192,
+ (0, None),
+ &mut input_indices,
+ &mut match_indices,
+ );
+ match_indices.sort_unstable();
+ match_indices.dedup();
+ match_indices
+ };
+
+ let mut map = JoinHashMapU32::with_capacity(batch.num_rows());
+ update_hash(
+ &on,
+ &batch,
+ &mut map,
+ 0,
+ &random_state,
+ &mut hashes_buffer,
+ 0,
+ true,
+ NullEquality::NullEqualsNothing,
+ )?;
+ // NULL keys can never match under NullEqualsNothing, so they must not
+ // be inserted into the map. Assert row indices rather than map length:
+ // with forced hash collisions, multiple logical keys can share one
+ // hash table entry.
+ assert_eq!(matched_build_indices(&map, &hashes_buffer), vec![0, 2, 4]);
+
+ let mut map = JoinHashMapU32::with_capacity(batch.num_rows());
+ update_hash(
+ &on,
+ &batch,
+ &mut map,
+ 0,
+ &random_state,
+ &mut hashes_buffer,
+ 0,
+ true,
+ NullEquality::NullEqualsNull,
+ )?;
+ // Under NullEqualsNull, NULL keys can match, so the build-side NULL
+ // rows must be present in the map.
+ assert_eq!(
+ matched_build_indices(&map, &hashes_buffer),
+ vec![0, 1, 2, 3, 4]
+ );
+
+ Ok(())
+ }
+
#[test]
fn get_anti_indices_handles_dense_matches() {
let input = UInt32Array::from(vec![2, 3, 4, 5]);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org