diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs index a51278b4556a..08eb5aa22bb9 100644 --- a/datafusion/common/src/hash_utils.rs +++ b/datafusion/common/src/hash_utils.rs @@ -321,15 +321,9 @@ fn hash_array_primitive( if array.null_count() == 0 { if rehash { - for (hash, &value) in hashes_buffer.iter_mut().zip(array.values().iter()) { - let mut hasher = random_state.seeded_state(*hash).build_hasher(); - value.hash_write(&mut hasher); - *hash = hasher.finish(); - } + hash_prim_rehash_dense(array.values(), hashes_buffer, random_state); } else { - for (hash, &value) in hashes_buffer.iter_mut().zip(array.values().iter()) { - *hash = value.hash_one(random_state); - } + hash_prim_fresh_dense(array.values(), hashes_buffer, random_state); } } else if rehash { for i in array.nulls().unwrap().valid_indices() { @@ -346,6 +340,82 @@ fn hash_array_primitive( } } +/// Overwrite each slot with `value.hash_one(state)`, 16 rows per iteration. +/// Gathering + hashing a whole lane batch before storing keeps the independent +/// hashes in flight so the pipeline isn't serialized on store-to-load deps. +#[cfg(not(feature = "force_hash_collisions"))] +#[inline(always)] +fn hash_prim_fresh_dense(values: &[N], hashes: &mut [u64], state: &S) +where + N: HashValue + Copy, + S: HashState, +{ + debug_assert_eq!(values.len(), hashes.len()); + const LANES: usize = 16; + let len = values.len(); + let aligned_end = len & !(LANES - 1); + let mut row = 0; + while row < aligned_end { + unsafe { + let new_hashes: [u64; LANES] = std::array::from_fn(|lane| { + values.get_unchecked(row + lane).hash_one(state) + }); + for lane in 0..LANES { + *hashes.get_unchecked_mut(row + lane) = new_hashes[lane]; + } + } + row += LANES; + } + while row < len { + unsafe { + *hashes.get_unchecked_mut(row) = values.get_unchecked(row).hash_one(state); + } + row += 1; + } +} + +/// Fold `prev_hash` + `value` into a new hash, 8 rows per iteration. 8 is +/// narrower than `hash_prim_fresh_dense` because each lane holds both `prev` and +/// `value` live; widening past 8 spilled registers in the multi-column case. +#[cfg(not(feature = "force_hash_collisions"))] +#[inline(always)] +fn hash_prim_rehash_dense(values: &[N], hashes: &mut [u64], state: &S) +where + N: HashValue + Copy, + S: HashState, +{ + debug_assert_eq!(values.len(), hashes.len()); + const LANES: usize = 8; + let len = values.len(); + let aligned_end = len & !(LANES - 1); + let mut row = 0; + while row < aligned_end { + unsafe { + let new_hashes: [u64; LANES] = std::array::from_fn(|lane| { + let value = *values.get_unchecked(row + lane); + let prev_hash = *hashes.get_unchecked(row + lane); + let mut hasher = state.seeded_state(prev_hash).build_hasher(); + value.hash_write(&mut hasher); + hasher.finish() + }); + for lane in 0..LANES { + *hashes.get_unchecked_mut(row + lane) = new_hashes[lane]; + } + } + row += LANES; + } + while row < len { + unsafe { + let value = *values.get_unchecked(row); + let prev_hash = *hashes.get_unchecked(row); + let mut hasher = state.seeded_state(prev_hash).build_hasher(); + value.hash_write(&mut hasher); + *hashes.get_unchecked_mut(row) = hasher.finish(); + } + row += 1; + } +} + /// Hashes one array into the `hashes_buffer` /// If `rehash==true` this combines the previous hash value in the buffer /// with the new hash using `combine_hashes`