diff --git a/native/spark-expr/src/hash_funcs/murmur3.rs b/native/spark-expr/src/hash_funcs/murmur3.rs index dc0f804ab26..233097ffc15 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>(data: T, seed: u32) -> u32 fn create_hashes_dictionary( array: &ArrayRef, hashes_buffer: &mut [u32], - first_col: bool, + seeds_are_pristine: bool, ) -> datafusion::common::Result<()> { let dict_array = array.as_any().downcast_ref::>().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::::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::::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::::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 = 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::( diff --git a/native/spark-expr/src/hash_funcs/utils.rs b/native/spark-expr/src/hash_funcs/utils.rs index 979f787b596..18e5a41bc75 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::(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::().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 c9d0f93ef8b..7009fc99c2c 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>(data: T, seed: u64) -> u64 { fn create_xxhash64_hashes_dictionary( array: &ArrayRef, hashes_buffer: &mut [u64], - first_col: bool, + seeds_are_pristine: bool, ) -> Result<()> { let dict_array = array.as_any().downcast_ref::>().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( // 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::::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::::try_new(keys, values).unwrap()); + let decoded: ArrayRef = Arc::new(Int32Array::from(vec![ + Some(10), + Some(20), + None, + None, + Some(10), + ])); + + let seeds: Vec = 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::(