This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch branch-1.0
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/branch-1.0 by this push:
new e1102f488c fix: keep the dictionary hash fast path off nested and
reseeded buffers (#5757) (#5817)
e1102f488c is described below
commit e1102f488c302456f0b7bd5f6dbe1e902d8597df
Author: Andy Grove <[email protected]>
AuthorDate: Wed Sep 9 17:23:19 2026 -0600
fix: keep the dictionary hash fast path off nested and reseeded buffers
(#5757) (#5817)
The dictionary fast path hashes each distinct dictionary value once and
reuses that
result for every key. It was selected by the column's position, `i == 0`,
and it
restarted from a hardcoded seed of 42.
Both parts are wrong. `create_hashes_internal!` also runs on recursion, so a
dictionary nested in a list, struct or map arrives as the only column of
its call and
looks like a first column even though the buffer already holds the hash
accumulated
for earlier elements of that row. That hash was discarded, and a
dictionary-encoded
list element hashed differently from the identical decoded value.
Separately, the
hardcoded 42 is wrong whenever the caller supplies its own seed, as
`hash(col, seed)`
and `xxhash64(col, seed)` allow, so even a genuine first column disagreed
with its
decoded form for a non-default seed.
The reuse is valid exactly when every row carries the same incoming hash,
so that is
what is now checked, and the per-value hashes start from the seed the
buffer actually
holds rather than an assumed 42. A top-level dictionary keeps the
optimisation.
The uniformity check is a scan of the hash buffer, which is measurable:
running it for
every column cost 17% on an int column and 11% on a string column in a local
criterion benchmark. It is therefore done inside the dictionary arm, so only
dictionary columns pay it and other types are untouched.
Both hash implementations share this structure and both are fixed, with
regression
tests that fail without the change: a dictionary as a list element hashes
3853467749
rather than the 1401423033 of the decoded data.
The single-row cases above pin the hardcoded seed but not the uniformity
check, since
one row is uniform by definition. Each algorithm therefore also gets a
multi-row case
whose incoming seeds all differ, which forces the unpacking fallback, and
which
includes a null key and a key pointing at a null dictionary value. Dropping
the
uniformity check leaves the other twenty hash tests green and fails exactly
those two.
(cherry picked from commit 92ad99e97482c861062f52372b172320197c2001)
Co-authored-by: Liang-Chi Hsieh <[email protected]>
Co-authored-by: Claude Code <[email protected]>
---
native/spark-expr/src/hash_funcs/murmur3.rs | 111 +++++++++++++++++++++++++--
native/spark-expr/src/hash_funcs/utils.rs | 20 ++++-
native/spark-expr/src/hash_funcs/xxhash64.rs | 65 +++++++++++++++-
3 files changed, 183 insertions(+), 13 deletions(-)
diff --git a/native/spark-expr/src/hash_funcs/murmur3.rs
b/native/spark-expr/src/hash_funcs/murmur3.rs
index dc0f804ab2..233097ffc1 100644
--- a/native/spark-expr/src/hash_funcs/murmur3.rs
+++ b/native/spark-expr/src/hash_funcs/murmur3.rs
@@ -139,20 +139,23 @@ pub fn spark_compatible_murmur3_hash<T:
AsRef<[u8]>>(data: T, seed: u32) -> u32
fn create_hashes_dictionary<K: ArrowDictionaryKeyType>(
array: &ArrayRef,
hashes_buffer: &mut [u32],
- first_col: bool,
+ seeds_are_pristine: bool,
) -> datafusion::common::Result<()> {
let dict_array =
array.as_any().downcast_ref::<DictionaryArray<K>>().unwrap();
- if !first_col {
+ if !seeds_are_pristine {
// unpack the dictionary array as each row may have a different hash
input
let unpacked = take(dict_array.values().as_ref(), dict_array.keys(),
None)?;
create_murmur3_hashes(&[unpacked], hashes_buffer)?;
} else {
- // For the first column, hash each dictionary value once, and then use
- // that computed hash for each key value to avoid a potentially
- // expensive redundant hashing for large dictionary elements (e.g.
strings)
+ // Every row still carries the untouched seed, so each distinct
dictionary value hashes to
+ // the same result no matter which row it appears in. Hash each value
once and reuse it per
+ // key, which avoids redundant hashing of large dictionary elements
(e.g. strings).
let dict_values = Arc::clone(dict_array.values());
- // same initial seed as Spark
- let mut dict_hashes = vec![42; dict_values.len()];
+ // Seed from the buffer rather than assuming Spark's 42: `hash(col,
seed)` lets the caller
+ // choose, and the reuse is only sound if the per-value hashes start
from the same seed the
+ // rows carry. The caller guarantees the buffer is uniform, so any
row's value will do.
+ let seed = hashes_buffer.first().copied().unwrap_or(42);
+ let mut dict_hashes = vec![seed; dict_values.len()];
create_murmur3_hashes(&[dict_values], &mut dict_hashes)?;
for (hash, key) in
hashes_buffer.iter_mut().zip(dict_array.keys().iter()) {
if let Some(key) = key {
@@ -205,6 +208,100 @@ mod tests {
test_hashes_with_nulls!(create_murmur3_hashes, T, values, expected,
u32);
}
+ /// A dictionary array reached through a nested type arrives as the only
column of its recursive
+ /// call, so deciding the dictionary fast path from column position alone
treated it as a first
+ /// column and restarted from the seed, discarding the hash accumulated
for earlier elements of
+ /// the same row. The result differed from the identical decoded data.
+ #[test]
+ fn test_dictionary_element_in_list_matches_decoded() {
+ use arrow::array::{DictionaryArray, Int32Array, ListArray};
+ use arrow::buffer::OffsetBuffer;
+ use arrow::datatypes::{Field, Int8Type};
+
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20]));
+ let keys = arrow::array::Int8Array::from(vec![0i8, 1]);
+ let dict: ArrayRef =
Arc::new(DictionaryArray::<Int8Type>::try_new(keys, values).unwrap());
+ let decoded: ArrayRef = Arc::new(Int32Array::from(vec![10, 20]));
+
+ // One row holding both elements, so the second element's hash chains
onto the first.
+ let as_list = |elems: ArrayRef| -> ArrayRef {
+ Arc::new(ListArray::new(
+ Arc::new(Field::new("item", elems.data_type().clone(), true)),
+ OffsetBuffer::new(vec![0i32, 2].into()),
+ elems,
+ None,
+ ))
+ };
+
+ let mut from_dict = vec![42u32; 1];
+ create_murmur3_hashes(&[as_list(dict)], &mut from_dict).unwrap();
+ let mut from_decoded = vec![42u32; 1];
+ create_murmur3_hashes(&[as_list(decoded)], &mut from_decoded).unwrap();
+
+ assert_eq!(
+ from_dict, from_decoded,
+ "a dictionary-encoded list element must hash like the decoded
value"
+ );
+ }
+
+ /// The fast path must survive for a genuine first column, including one
with a caller-supplied
+ /// seed that is not Spark's 42, since the test for it is that every row
is seeded alike.
+ #[test]
+ fn test_top_level_dictionary_matches_decoded() {
+ use arrow::array::{DictionaryArray, Int32Array};
+ use arrow::datatypes::Int8Type;
+
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20]));
+ let keys = arrow::array::Int8Array::from(vec![0i8, 1, 0]);
+ let dict: ArrayRef =
Arc::new(DictionaryArray::<Int8Type>::try_new(keys, values).unwrap());
+ let decoded: ArrayRef = Arc::new(Int32Array::from(vec![10, 20, 10]));
+
+ for seed in [42u32, 7u32] {
+ let mut a = vec![seed; 3];
+ create_murmur3_hashes(&[Arc::clone(&dict)], &mut a).unwrap();
+ let mut b = vec![seed; 3];
+ create_murmur3_hashes(&[Arc::clone(&decoded)], &mut b).unwrap();
+ assert_eq!(
+ a, b,
+ "top-level dictionary must match decoded for seed {seed}"
+ );
+ }
+ }
+
+ /// The uniformity check is what makes the fast path safe, and a
single-row case cannot pin it:
+ /// one row is trivially uniform. This hashes several rows whose incoming
seeds all differ, so a
+ /// dictionary first column has to take the unpacking fallback. It also
covers a null key and a
+ /// key pointing at a null dictionary value, since both skip the hash
update.
+ #[test]
+ fn test_dictionary_with_nonuniform_seeds_matches_decoded() {
+ use arrow::array::{DictionaryArray, Int32Array};
+ use arrow::datatypes::Int8Type;
+
+ // values[2] is null, and one key is itself null
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(10),
Some(20), None]));
+ let keys = arrow::array::Int8Array::from(vec![Some(0), Some(1),
Some(2), None, Some(0)]);
+ let dict: ArrayRef =
Arc::new(DictionaryArray::<Int8Type>::try_new(keys, values).unwrap());
+ // The same logical data with the dictionary resolved.
+ let decoded: ArrayRef = Arc::new(Int32Array::from(vec![
+ Some(10),
+ Some(20),
+ None,
+ None,
+ Some(10),
+ ]));
+
+ let seeds: Vec<u32> = vec![7, 38, 69, 100, 131];
+ let mut from_dict = seeds.clone();
+ create_murmur3_hashes(&[dict], &mut from_dict).unwrap();
+ let mut from_decoded = seeds;
+ create_murmur3_hashes(&[decoded], &mut from_decoded).unwrap();
+
+ assert_eq!(
+ from_dict, from_decoded,
+ "with per-row seeds a dictionary must hash like the decoded array"
+ );
+ }
+
#[test]
fn test_i8() {
test_murmur3_hash::<i8, Int8Array>(
diff --git a/native/spark-expr/src/hash_funcs/utils.rs
b/native/spark-expr/src/hash_funcs/utils.rs
index 979f787b59..18e5a41bc7 100644
--- a/native/spark-expr/src/hash_funcs/utils.rs
+++ b/native/spark-expr/src/hash_funcs/utils.rs
@@ -576,6 +576,15 @@ macro_rules! create_hashes_internal {
use arrow::array::{types::*, *};
for (i, col) in $arrays.iter().enumerate() {
+ // The dictionary fast path hashes each distinct dictionary value
once and reuses that
+ // result for every key, which is only valid while every row
carries the same incoming
+ // hash. Position in the column list is not a sufficient test:
this macro also runs on
+ // recursion, where a nested dictionary arrives as the only column
of its call even
+ // though the buffer already holds the hash accumulated for that
row -- a
+ // dictionary-encoded list element, for instance. So confirm the
buffer is uniform,
+ // which keeps the optimisation for a genuine first column (every
row seeded alike,
+ // whatever the seed) and unpacks otherwise. Only dictionaries
need this, and the scan
+ // is measurable on the hot path, so it is deferred into the
dictionary arm below.
let first_col = i == 0;
match col.data_type() {
DataType::Boolean => {
@@ -729,7 +738,13 @@ macro_rules! create_hashes_internal {
DataType::Decimal128(_, _) => {
$crate::hash_array_decimal!(Decimal128Array, col,
$hashes_buffer, $hash_method);
}
- DataType::Dictionary(index_type, _) => match **index_type {
+ DataType::Dictionary(index_type, _) => {
+ let first_col = first_col
+ && match $hashes_buffer.first() {
+ None => true,
+ Some(first) => $hashes_buffer.iter().all(|h| h ==
first),
+ };
+ match **index_type {
DataType::Int8 => {
$create_dictionary_hash_method::<Int8Type>(col,
$hashes_buffer, first_col)?;
}
@@ -788,7 +803,8 @@ macro_rules! create_hashes_internal {
col.data_type(),
)))
}
- },
+ }
+ }
DataType::List(field) => {
let list_array =
col.as_any().downcast_ref::<ListArray>().unwrap();
let values = list_array.values();
diff --git a/native/spark-expr/src/hash_funcs/xxhash64.rs
b/native/spark-expr/src/hash_funcs/xxhash64.rs
index c9d0f93ef8..7009fc99c2 100644
--- a/native/spark-expr/src/hash_funcs/xxhash64.rs
+++ b/native/spark-expr/src/hash_funcs/xxhash64.rs
@@ -85,10 +85,10 @@ fn spark_compatible_xxhash64<T: AsRef<[u8]>>(data: T, seed:
u64) -> u64 {
fn create_xxhash64_hashes_dictionary<K: ArrowDictionaryKeyType>(
array: &ArrayRef,
hashes_buffer: &mut [u64],
- first_col: bool,
+ seeds_are_pristine: bool,
) -> Result<()> {
let dict_array =
array.as_any().downcast_ref::<DictionaryArray<K>>().unwrap();
- if !first_col {
+ if !seeds_are_pristine {
let unpacked = take(dict_array.values().as_ref(), dict_array.keys(),
None)?;
create_xxhash64_hashes(&[unpacked], hashes_buffer)?;
} else {
@@ -96,8 +96,11 @@ fn create_xxhash64_hashes_dictionary<K:
ArrowDictionaryKeyType>(
// hash for each key value to avoid a potentially expensive
// redundant hashing for large dictionary elements (e.g. strings)
let dict_values = Arc::clone(dict_array.values());
- // same initial seed as Spark
- let mut dict_hashes = vec![42u64; dict_values.len()];
+ // Seed from the buffer rather than assuming Spark's 42:
`xxhash64(col, seed)` lets the
+ // caller choose, and the reuse is only sound if the per-value hashes
start from the same
+ // seed the rows carry. The caller guarantees the buffer is uniform.
+ let seed = hashes_buffer.first().copied().unwrap_or(42u64);
+ let mut dict_hashes = vec![seed; dict_values.len()];
create_xxhash64_hashes(&[dict_values], &mut dict_hashes)?;
for (hash, key) in
hashes_buffer.iter_mut().zip(dict_array.keys().iter()) {
@@ -151,6 +154,60 @@ mod tests {
test_hashes_with_nulls!(create_xxhash64_hashes, T, values, expected,
u64);
}
+ /// The dictionary fast path is shared in shape with murmur3, so it has
the same requirement:
+ /// a dictionary reached through a nested type must not restart from the
seed.
+ #[test]
+ fn test_dictionary_element_in_list_matches_decoded() {
+ use arrow::array::{DictionaryArray, ListArray};
+ use arrow::buffer::OffsetBuffer;
+ use arrow::datatypes::{Field, Int8Type};
+
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20]));
+ let keys = arrow::array::Int8Array::from(vec![0i8, 1]);
+ let dict: ArrayRef =
Arc::new(DictionaryArray::<Int8Type>::try_new(keys, values).unwrap());
+ let decoded: ArrayRef = Arc::new(Int32Array::from(vec![10, 20]));
+
+ let as_list = |elems: ArrayRef| -> ArrayRef {
+ Arc::new(ListArray::new(
+ Arc::new(Field::new("item", elems.data_type().clone(), true)),
+ OffsetBuffer::new(vec![0i32, 2].into()),
+ elems,
+ None,
+ ))
+ };
+
+ let mut from_dict = vec![42u64; 1];
+ create_xxhash64_hashes(&[as_list(dict)], &mut from_dict).unwrap();
+ let mut from_decoded = vec![42u64; 1];
+ create_xxhash64_hashes(&[as_list(decoded)], &mut
from_decoded).unwrap();
+ assert_eq!(from_dict, from_decoded);
+ }
+
+ /// Companion to the murmur3 test: per-row seeds force the unpacking
fallback here too.
+ #[test]
+ fn test_dictionary_with_nonuniform_seeds_matches_decoded() {
+ use arrow::array::DictionaryArray;
+ use arrow::datatypes::Int8Type;
+
+ let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(10),
Some(20), None]));
+ let keys = arrow::array::Int8Array::from(vec![Some(0), Some(1),
Some(2), None, Some(0)]);
+ let dict: ArrayRef =
Arc::new(DictionaryArray::<Int8Type>::try_new(keys, values).unwrap());
+ let decoded: ArrayRef = Arc::new(Int32Array::from(vec![
+ Some(10),
+ Some(20),
+ None,
+ None,
+ Some(10),
+ ]));
+
+ let seeds: Vec<u64> = vec![7, 38, 69, 100, 131];
+ let mut from_dict = seeds.clone();
+ create_xxhash64_hashes(&[dict], &mut from_dict).unwrap();
+ let mut from_decoded = seeds;
+ create_xxhash64_hashes(&[decoded], &mut from_decoded).unwrap();
+ assert_eq!(from_dict, from_decoded);
+ }
+
#[test]
fn test_i8() {
test_xxhash64_hash::<i8, Int8Array>(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
