From 8c0a4ee37cb541c39f421b8a87c4f14a616ffb4b Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Mon, 2 Nov 2026 11:36:25 -0500 Subject: [PATCH 1/6] perf: speed up non-null dictionary hash scatter Skip clearing the reusable hash buffer when a dictionary with non-null keys and values will overwrite every row. Gather dictionary hashes in independent lane groups so the CPU can overlap the key-indexed loads. --- datafusion/common/src/hash_utils.rs | 123 +++++++++++++++++++++++++++- 1 file changed, 121 insertions(+), 2 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index c9cc48a11dd21..2fec8b8ee5840 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -96,6 +96,18 @@ pub fn combine_hashes(l: u64, r: u64) -> u64 { /// to this size. const MAX_BUFFER_SIZE: usize = 524_288; +#[cfg(not(feature = "force_hash_collisions"))] +fn dictionary_overwrites_hash_buffer(array: &dyn Array) -> bool { + if !matches!(array.data_type(), DataType::Dictionary(_, _)) || array.null_count() != 0 + { + return false; + } + downcast_dictionary_array! { + array => array.values().null_count() == 0, + _ => false + } +} + thread_local! { /// Thread-local buffer for hash computations to avoid repeated allocations. /// The buffer is reused across calls and truncated if it exceeds MAX_BUFFER_SIZE. @@ -158,8 +170,17 @@ where let mut buffer = cell.try_borrow_mut() .map_err(|_| _internal_datafusion_err!("with_hashes cannot be called reentrantly on the same thread"))?; - // Ensure buffer has sufficient length, clearing old values - buffer.clear(); + // A dictionary with non-null keys and values overwrites every output + // slot, so retain the initialized storage instead of zeroing it first. + #[cfg(not(feature = "force_hash_collisions"))] + let reuse_buffer = iter + .peek() + .is_some_and(|array| dictionary_overwrites_hash_buffer(array.as_dyn_array())); + #[cfg(feature = "force_hash_collisions")] + let reuse_buffer = false; + if !reuse_buffer { + buffer.clear(); + } buffer.resize(required_size, 0); // Create hashes in the buffer - this consumes the iterator @@ -544,6 +565,46 @@ fn hash_generic_byte_view_array( } } +#[cfg(not(feature = "force_hash_collisions"))] +#[inline(always)] +fn scatter_non_null_dictionary( + keys: &[K::Native], + dict_hashes: &[u64], + hashes: &mut [u64], +) { + const LANES: usize = 8; + let len = keys.len().min(hashes.len()); + let keys = &keys[..len]; + let hashes = &mut hashes[..len]; + + let (key_chunks, key_tail) = keys.as_chunks::(); + let (hash_chunks, hash_tail) = hashes.as_chunks_mut::(); + for (keys, hashes) in key_chunks.iter().zip(hash_chunks) { + let gathered: [u64; LANES] = std::array::from_fn(|lane| unsafe { + let key = keys[lane]; + // DictionaryArray guarantees that every non-null key is in bounds. + *dict_hashes.get_unchecked(key.as_usize()) + }); + if MULTI_COL { + for (hash, dict_hash) in hashes.iter_mut().zip(gathered) { + *hash = combine_hashes(dict_hash, *hash); + } + } else { + *hashes = gathered; + } + } + + for (hash, key) in hash_tail.iter_mut().zip(key_tail) { + // DictionaryArray guarantees that every non-null key is in bounds. + let dict_hash = unsafe { *dict_hashes.get_unchecked(key.as_usize()) }; + *hash = if MULTI_COL { + combine_hashes(dict_hash, *hash) + } else { + dict_hash + }; + } +} + /// Scatter precomputed dictionary value hashes to key positions. /// /// Uses const generics to eliminate runtime branching in the hot loop: @@ -562,6 +623,14 @@ fn hash_dictionary_scatter< dict_hashes: &[u64], hashes_buffer: &mut [u64], ) { + if !HAS_NULL_KEYS && !HAS_NULL_VALUES { + scatter_non_null_dictionary::( + array.keys().values(), + dict_hashes, + hashes_buffer, + ); + return; + } let dict_values = array.values(); if HAS_NULL_KEYS { for (hash, key) in hashes_buffer.iter_mut().zip(array.keys().iter()) { @@ -2284,6 +2353,56 @@ mod tests { assert_eq!(result, expected_hashes); } + #[test] + #[cfg(not(feature = "force_hash_collisions"))] + fn test_non_null_dictionary_hash_fast_path() { + let random_state = RandomState::with_seed(0); + let seed: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4])); + let non_null: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(vec![0, 1, 0, 1]), + Arc::new(StringArray::from(vec!["a", "b"])), + ) + .unwrap(), + ); + let null_values: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(vec![0, 1, 0, 1]), + Arc::new(StringArray::from(vec![Some("a"), None])), + ) + .unwrap(), + ); + let null_keys: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(vec![Some(0), None, Some(1), None]), + Arc::new(StringArray::from(vec!["a", "b"])), + ) + .unwrap(), + ); + + let mut full = vec![0; non_null.len()]; + create_hashes([&non_null], &random_state, &mut full).unwrap(); + let mut prefix = vec![0; non_null.len() - 1]; + create_hashes([&non_null], &random_state, &mut prefix).unwrap(); + assert_eq!(prefix, full[..prefix.len()]); + + for dictionary in [non_null, null_values, null_keys] { + with_hashes([&seed], &random_state, |hashes| { + assert!(hashes.iter().all(|hash| *hash != 0)); + Ok(()) + }) + .unwrap(); + + let mut expected = vec![0; dictionary.len()]; + create_hashes([&dictionary], &random_state, &mut expected).unwrap(); + with_hashes([&dictionary], &random_state, |hashes| { + assert_eq!(hashes, expected); + Ok(()) + }) + .unwrap(); + } + } + #[test] fn test_with_hashes_multi_column() { let int_array: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3])); From cb54f47fd46c7a859c909c6e42416d1cfa6e6d61 Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Mon, 2 Nov 2026 22:44:36 -0500 Subject: [PATCH 2/6] Remove redundant dictionary scatter test Existing hash utility tests cover dictionary hashing; retain a concise comment explaining the lane-gather fast path. --- datafusion/common/src/hash_utils.rs | 51 +---------------------------- 1 file changed, 1 insertion(+), 50 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index 2fec8b8ee5840..abf2d9935045b 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -565,6 +565,7 @@ fn hash_generic_byte_view_array( } } +/// Scatters non-null dictionary hashes in independent lanes to overlap key-indexed loads. #[cfg(not(feature = "force_hash_collisions"))] #[inline(always)] fn scatter_non_null_dictionary( @@ -2353,56 +2354,6 @@ mod tests { assert_eq!(result, expected_hashes); } - #[test] - #[cfg(not(feature = "force_hash_collisions"))] - fn test_non_null_dictionary_hash_fast_path() { - let random_state = RandomState::with_seed(0); - let seed: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4])); - let non_null: ArrayRef = Arc::new( - DictionaryArray::::try_new( - Int32Array::from(vec![0, 1, 0, 1]), - Arc::new(StringArray::from(vec!["a", "b"])), - ) - .unwrap(), - ); - let null_values: ArrayRef = Arc::new( - DictionaryArray::::try_new( - Int32Array::from(vec![0, 1, 0, 1]), - Arc::new(StringArray::from(vec![Some("a"), None])), - ) - .unwrap(), - ); - let null_keys: ArrayRef = Arc::new( - DictionaryArray::::try_new( - Int32Array::from(vec![Some(0), None, Some(1), None]), - Arc::new(StringArray::from(vec!["a", "b"])), - ) - .unwrap(), - ); - - let mut full = vec![0; non_null.len()]; - create_hashes([&non_null], &random_state, &mut full).unwrap(); - let mut prefix = vec![0; non_null.len() - 1]; - create_hashes([&non_null], &random_state, &mut prefix).unwrap(); - assert_eq!(prefix, full[..prefix.len()]); - - for dictionary in [non_null, null_values, null_keys] { - with_hashes([&seed], &random_state, |hashes| { - assert!(hashes.iter().all(|hash| *hash != 0)); - Ok(()) - }) - .unwrap(); - - let mut expected = vec![0; dictionary.len()]; - create_hashes([&dictionary], &random_state, &mut expected).unwrap(); - with_hashes([&dictionary], &random_state, |hashes| { - assert_eq!(hashes, expected); - Ok(()) - }) - .unwrap(); - } - } - #[test] fn test_with_hashes_multi_column() { let int_array: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3])); From c7d9b3a2de4eb70d51e869cf41d9110c4f8cc955 Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Mon, 2 Nov 2026 23:14:40 -0500 Subject: [PATCH 3/6] fix: tighten non-null dictionary hash scatter Keep the buffer-reuse and scatter dispatch checks in one place, validate the unchecked lookup invariant, and cover chunk, tail, multi-column, and buffer-reuse behavior. --- datafusion/common/src/hash_utils.rs | 136 +++++++++++++++++++++++----- 1 file changed, 111 insertions(+), 25 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index abf2d9935045b..6c8e71f8bd084 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -96,18 +96,29 @@ pub fn combine_hashes(l: u64, r: u64) -> u64 { /// to this size. const MAX_BUFFER_SIZE: usize = 524_288; +#[cfg(not(feature = "force_hash_collisions"))] +fn dictionary_nulls( + array: &DictionaryArray, +) -> (bool, bool) { + ( + array.keys().null_count() != 0, + array.values().null_count() != 0, + ) +} + #[cfg(not(feature = "force_hash_collisions"))] fn dictionary_overwrites_hash_buffer(array: &dyn Array) -> bool { - if !matches!(array.data_type(), DataType::Dictionary(_, _)) || array.null_count() != 0 - { - return false; - } downcast_dictionary_array! { - array => array.values().null_count() == 0, + array => dictionary_nulls(array) == (false, false), _ => false } } +#[cfg(feature = "force_hash_collisions")] +fn dictionary_overwrites_hash_buffer(_array: &dyn Array) -> bool { + false +} + thread_local! { /// Thread-local buffer for hash computations to avoid repeated allocations. /// The buffer is reused across calls and truncated if it exceeds MAX_BUFFER_SIZE. @@ -172,12 +183,9 @@ where // A dictionary with non-null keys and values overwrites every output // slot, so retain the initialized storage instead of zeroing it first. - #[cfg(not(feature = "force_hash_collisions"))] let reuse_buffer = iter .peek() .is_some_and(|array| dictionary_overwrites_hash_buffer(array.as_dyn_array())); - #[cfg(feature = "force_hash_collisions")] - let reuse_buffer = false; if !reuse_buffer { buffer.clear(); } @@ -569,11 +577,14 @@ fn hash_generic_byte_view_array( #[cfg(not(feature = "force_hash_collisions"))] #[inline(always)] fn scatter_non_null_dictionary( - keys: &[K::Native], + array: &DictionaryArray, dict_hashes: &[u64], hashes: &mut [u64], ) { const LANES: usize = 8; + assert_eq!(dict_hashes.len(), array.values().len()); + assert_eq!(array.keys().null_count(), 0); + let keys = array.keys().values(); let len = keys.len().min(hashes.len()); let keys = &keys[..len]; let hashes = &mut hashes[..len]; @@ -583,12 +594,13 @@ fn scatter_non_null_dictionary for (keys, hashes) in key_chunks.iter().zip(hash_chunks) { let gathered: [u64; LANES] = std::array::from_fn(|lane| unsafe { let key = keys[lane]; - // DictionaryArray guarantees that every non-null key is in bounds. + // SAFETY: Dictionary keys are in bounds for values, which has the + // same length as dict_hashes. *dict_hashes.get_unchecked(key.as_usize()) }); if MULTI_COL { for (hash, dict_hash) in hashes.iter_mut().zip(gathered) { - *hash = combine_hashes(dict_hash, *hash); + update_dictionary_hash::(hash, dict_hash); } } else { *hashes = gathered; @@ -596,16 +608,23 @@ fn scatter_non_null_dictionary } for (hash, key) in hash_tail.iter_mut().zip(key_tail) { - // DictionaryArray guarantees that every non-null key is in bounds. + // SAFETY: Dictionary keys are in bounds for values, which has the + // same length as dict_hashes. let dict_hash = unsafe { *dict_hashes.get_unchecked(key.as_usize()) }; - *hash = if MULTI_COL { - combine_hashes(dict_hash, *hash) - } else { - dict_hash - }; + update_dictionary_hash::(hash, dict_hash); } } +#[cfg(not(feature = "force_hash_collisions"))] +#[inline(always)] +fn update_dictionary_hash(hash: &mut u64, dict_hash: u64) { + *hash = if MULTI_COL { + combine_hashes(dict_hash, *hash) + } else { + dict_hash + }; +} + /// Scatter precomputed dictionary value hashes to key positions. /// /// Uses const generics to eliminate runtime branching in the hot loop: @@ -625,11 +644,7 @@ fn hash_dictionary_scatter< hashes_buffer: &mut [u64], ) { if !HAS_NULL_KEYS && !HAS_NULL_VALUES { - scatter_non_null_dictionary::( - array.keys().values(), - dict_hashes, - hashes_buffer, - ); + scatter_non_null_dictionary::(array, dict_hashes, hashes_buffer); return; } let dict_values = array.values(); @@ -651,7 +666,7 @@ fn hash_dictionary_scatter< } else { for (hash, key) in hashes_buffer.iter_mut().zip(array.keys().values()) { let idx = key.as_usize(); - if !HAS_NULL_VALUES || dict_values.is_valid(idx) { + if dict_values.is_valid(idx) { // SAFETY: `idx` is a valid Arrow dict key, in-bounds for `dict_hashes`. let dict_hash = unsafe { *dict_hashes.get_unchecked(idx) }; *hash = if MULTI_COL { @@ -671,8 +686,7 @@ fn dispatch_dictionary_scatter( hashes_buffer: &mut [u64], multi_col: bool, ) { - let has_null_keys = array.keys().null_count() != 0; - let has_null_values = array.values().null_count() != 0; + let (has_null_keys, has_null_values) = dictionary_nulls(array); match (has_null_keys, has_null_values, multi_col) { (false, false, false) => hash_dictionary_scatter::( @@ -2354,6 +2368,78 @@ mod tests { assert_eq!(result, expected_hashes); } + #[test] + #[cfg(not(feature = "force_hash_collisions"))] + fn test_non_null_dictionary_hash_fast_path() { + let random_state = RandomState::with_seed(0); + let keys = vec![2, 0, 1, 2, 1, 0, 2, 0, 1, 2, 0]; + let non_null: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(keys.clone()), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + ) + .unwrap(), + ); + let null_values: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(keys), + Arc::new(StringArray::from(vec![Some("a"), None, Some("c")])), + ) + .unwrap(), + ); + let null_keys: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int32Array::from(vec![ + Some(2), + None, + Some(1), + Some(2), + None, + Some(0), + Some(2), + None, + Some(1), + Some(2), + Some(0), + ]), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + ) + .unwrap(), + ); + + let assert_matches = |dictionary: &ArrayRef| { + let strings = arrow::compute::cast(dictionary, &DataType::Utf8).unwrap(); + let mut expected = vec![0; dictionary.len()]; + create_hashes([strings], &random_state, &mut expected).unwrap(); + with_hashes([dictionary], &random_state, |hashes| { + assert_eq!(hashes, expected); + Ok(()) + }) + .unwrap(); + }; + + let strings = arrow::compute::cast(&non_null, &DataType::Utf8).unwrap(); + let mut expected = vec![0; strings.len()]; + create_hashes([&strings], &random_state, &mut expected).unwrap(); + let mut prefix = vec![0; non_null.len() - 1]; + create_hashes([&non_null], &random_state, &mut prefix).unwrap(); + assert_eq!(prefix, expected[..prefix.len()]); + + let seed: ArrayRef = + Arc::new(Int32Array::from_iter_values(0..non_null.len() as i32)); + let mut expected = vec![0; seed.len()]; + create_hashes([&seed, &strings], &random_state, &mut expected).unwrap(); + with_hashes([&seed, &non_null], &random_state, |hashes| { + assert_eq!(hashes, expected); + Ok(()) + }) + .unwrap(); + + assert_matches(&non_null); + assert_matches(&null_values); + assert_matches(&null_keys); + } + #[test] fn test_with_hashes_multi_column() { let int_array: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3])); From 0a7c4da0cdcea319338b4150a28108ad9c0fffad Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Tue, 3 Nov 2026 09:01:49 -0500 Subject: [PATCH 4/6] fix: tighten dictionary scatter invariants Use debug assertions for internal scatter invariants, avoid repeated dictionary downcasts when deciding whether to reuse the hash buffer, and document why slices and key widths share the same path. --- datafusion/common/src/hash_utils.rs | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index 6c8e71f8bd084..1bc1fb360f57a 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -106,17 +106,8 @@ fn dictionary_nulls( ) } -#[cfg(not(feature = "force_hash_collisions"))] fn dictionary_overwrites_hash_buffer(array: &dyn Array) -> bool { - downcast_dictionary_array! { - array => dictionary_nulls(array) == (false, false), - _ => false - } -} - -#[cfg(feature = "force_hash_collisions")] -fn dictionary_overwrites_hash_buffer(_array: &dyn Array) -> bool { - false + matches!(array.data_type(), DataType::Dictionary(_, _)) && !array.is_nullable() } thread_local! { @@ -582,10 +573,17 @@ fn scatter_non_null_dictionary hashes: &mut [u64], ) { const LANES: usize = 8; - assert_eq!(dict_hashes.len(), array.values().len()); - assert_eq!(array.keys().null_count(), 0); + debug_assert_eq!(dict_hashes.len(), array.values().len()); + debug_assert_eq!(dictionary_nulls(array), (false, false)); let keys = array.keys().values(); let len = keys.len().min(hashes.len()); + debug_assert_eq!( + len, + hashes.len(), + "dictionary scatter must overwrite every hash slot" + ); + // `values()` accounts for key-array slices, and all key widths share the + // same `as_usize` lookup below. let keys = &keys[..len]; let hashes = &mut hashes[..len]; From b41d408fd729ac0c7397bdc0ac5118063ca534ec Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Tue, 3 Nov 2026 09:18:40 -0500 Subject: [PATCH 5/6] bench: restore hash buffer clearing Restore buffer initialization so the hosted benchmark can isolate the benefit of skipping it from the dictionary scatter changes. --- datafusion/common/src/hash_utils.rs | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index 1bc1fb360f57a..eb1293b5e5345 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -106,10 +106,6 @@ fn dictionary_nulls( ) } -fn dictionary_overwrites_hash_buffer(array: &dyn Array) -> bool { - matches!(array.data_type(), DataType::Dictionary(_, _)) && !array.is_nullable() -} - thread_local! { /// Thread-local buffer for hash computations to avoid repeated allocations. /// The buffer is reused across calls and truncated if it exceeds MAX_BUFFER_SIZE. @@ -172,14 +168,8 @@ where let mut buffer = cell.try_borrow_mut() .map_err(|_| _internal_datafusion_err!("with_hashes cannot be called reentrantly on the same thread"))?; - // A dictionary with non-null keys and values overwrites every output - // slot, so retain the initialized storage instead of zeroing it first. - let reuse_buffer = iter - .peek() - .is_some_and(|array| dictionary_overwrites_hash_buffer(array.as_dyn_array())); - if !reuse_buffer { - buffer.clear(); - } + // Ensure buffer has sufficient length, clearing old values + buffer.clear(); buffer.resize(required_size, 0); // Create hashes in the buffer - this consumes the iterator From 90f551c9cd7522ca1d81f4880294be4209536203 Mon Sep 17 00:00:00 2001 From: Richard Baah Date: Tue, 3 Nov 2026 09:32:14 -0500 Subject: [PATCH 6/6] perf: keep hash buffer reuse Restoring buffer clearing made the focused dictionary benchmarks 1.04x and 1.16x slower, so retain the guarded reuse path. --- datafusion/common/src/hash_utils.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index eb1293b5e5345..1bc1fb360f57a 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -106,6 +106,10 @@ fn dictionary_nulls( ) } +fn dictionary_overwrites_hash_buffer(array: &dyn Array) -> bool { + matches!(array.data_type(), DataType::Dictionary(_, _)) && !array.is_nullable() +} + thread_local! { /// Thread-local buffer for hash computations to avoid repeated allocations. /// The buffer is reused across calls and truncated if it exceeds MAX_BUFFER_SIZE. @@ -168,8 +172,14 @@ where let mut buffer = cell.try_borrow_mut() .map_err(|_| _internal_datafusion_err!("with_hashes cannot be called reentrantly on the same thread"))?; - // Ensure buffer has sufficient length, clearing old values - buffer.clear(); + // A dictionary with non-null keys and values overwrites every output + // slot, so retain the initialized storage instead of zeroing it first. + let reuse_buffer = iter + .peek() + .is_some_and(|array| dictionary_overwrites_hash_buffer(array.as_dyn_array())); + if !reuse_buffer { + buffer.clear(); + } buffer.resize(required_size, 0); // Create hashes in the buffer - this consumes the iterator