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/10/03 07:25:18 UTC
(datafusion) branch gh-readonly-queue/main/pr-25793-b550f59884c47cb03675295d3ef9f35159afc7d5 created (now 2566bc1052)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25793-b550f59884c47cb03675295d3ef9f35159afc7d5
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 2566bc1052 perf: hash only the selected values of union arrays (#25793)
This branch includes the following new commits:
new 2566bc1052 perf: hash only the selected values of union arrays (#25793)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: perf: hash only the selected values of union arrays (#25793)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25793-b550f59884c47cb03675295d3ef9f35159afc7d5
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2566bc1052b32cce327b11d196a4dade52dd57d8
Author: Abhishek <16...@users.noreply.github.com>
AuthorDate: Sat Oct 3 07:24:54 2026 +0000
perf: hash only the selected values of union arrays (#25793)
## Which issue does this PR close?
- Closes #20211.
## Rationale for this change
Hashing a union column (in a `GROUP BY`, a hash join or hash
repartitioning) did a lot of avoidable work:
- Every row went through a `HashMap`, either to look up its child's
hashes or to group rows by type.
- Sparse unions with 1 or 2 types hashed every child in full. With 3 or
more types, every selected value was copied with `take`.
- `UnionArray::slice` doesn't slice dense children, so a sliced dense
union hashed its children in full. This is the same problem #19500 and
#20179 fixed for lists and maps.
## What changes are included in this PR?
Sparse unions:
- Rows are grouped by type id once, with an O(n) counting sort.
- If every row picks the same child, that child is hashed directly.
- Utf8, LargeUtf8, Binary and LargeBinary children that own at least
1/32 of the rows are hashed without copying their value buffers, by
nulling out the other rows through `nullif`. This is the null buffer
idea @Jefffrey raised in
https://github.com/apache/datafusion/pull/20179#discussion_r2777750677.
- All other children go through `take`. Masking was slower for
fixed-width values, except about 10% faster for primitives with nulls,
which wasn't worth a special case. It was also slower for children that
own only a few rows, because building the mask costs O(len) per child.
Masking still won at 1/32 of the rows and lost at 1/64 for short, medium
and long strings alike, hence the cutoff.
- The special case for 1 or 2 types is gone.
Dense unions:
- A lookup table indexed by type id replaces the per-row `HashMap`.
- When the children's total length exceeds the union's, only the span
their offsets reference gets hashed.
Hash values are unchanged.
Not done: a `create_hashes` variant that hashes by index list instead of
`take` (the other idea from the issue). It would avoid the `take` copy,
but I haven't built or measured it, so I left it out of this PR.
Benchmarks: `with_hashes` gains `sparse_union (2 types)`, `sparse_union
(utf8)` and `dense_union_sliced`. `sparse_union_array` now builds its
array through a small `sparse_union_of` helper, and the data it produces
is identical to before.
## What is the testing strategy for this PR?
- New `create_hashes_for_union_arrays_match_selected_values` test. It
checks that each row of a union used as the second column contributes
the hash of its selected child value on its own, through both
`create_hashes` and `create_hashes_with_hasher`. It covers:
- sparse and sliced sparse unions, including a slice whose rows all
select one child
- a string child routed through `take`
- dense and sliced dense unions
- non-contiguous type ids
- The existing union tests pass unchanged.
- During development I compared the new code against the previous
implementation on 12,000 random unions. They used 28 child types
covering every category the hasher handles: primitives, decimals,
temporal types, strings, binary, views, dictionaries, nested types,
Null, run-end encoded and nested unions. Each union was sliced or
unsliced, and was hashed alone and as the first and second column, with
the fast and quality hash states and a custom `BuildHasher`. Every hash
matched. That test isn't committed because it embeds the old code.
Benchmarks (`cargo bench -p datafusion-common --bench with_hashes --
union`): medians averaged over two interleaved runs, one pinned core,
i7-11700F.
| benchmark | main | PR | speedup |
|---|---|---|---|
| sparse_union: single, no nulls | 93.3 µs | 27.7 µs | 3.37x |
| sparse_union: multiple, no nulls | 268.2 µs | 79.8 µs | 3.36x |
| sparse_union (2 types): single, no nulls | 81.5 µs | 28.8 µs | 2.83x |
| sparse_union (2 types): multiple, no nulls | 248.3 µs | 87.2 µs |
2.85x |
| sparse_union (utf8): single, no nulls | 176.6 µs | 61.3 µs | 2.88x |
| sparse_union (utf8): multiple, no nulls | 536.4 µs | 189.1 µs | 2.84x
|
| sparse_union_sliced: 1/10 of 81920 rows | 95.9 µs | 27.6 µs | 3.47x |
| sparse_union_sliced: 1/5 of 40960 rows | 93.2 µs | 26.8 µs | 3.47x |
| sparse_union_sliced: 1/2 of 16384 rows | 92.6 µs | 28.0 µs | 3.30x |
| dense_union: single, no nulls | 81.0 µs | 14.1 µs | 5.73x |
| dense_union: multiple, no nulls | 244.9 µs | 40.1 µs | 6.11x |
| dense_union_sliced: 1/10 of 81920 rows | 136.9 µs | 22.6 µs | 6.07x |
| dense_union_sliced: 1/5 of 40960 rows | 101.0 µs | 22.0 µs | 4.58x |
| dense_union_sliced: 1/2 of 16384 rows | 88.0 µs | 22.9 µs | 3.84x |
I also measured 25 more cases: 1 to 127 types; int64 with and without
nulls; short, medium and long strings; string view, bool, fixed size
binary, dictionary, struct, list, and mixed children. The PR was faster
than main in every one, from 1.3x to 7.5x.
## Are there any user-facing changes?
No API or result changes. Queries that hash union columns can run
faster.
---------
Co-authored-by: Kumar Ujjawal <uj...@gmail.com>
---
datafusion/common/benches/with_hashes.rs | 64 +++++--
datafusion/common/src/hash_utils.rs | 281 ++++++++++++++++++++++---------
2 files changed, 251 insertions(+), 94 deletions(-)
diff --git a/datafusion/common/benches/with_hashes.rs b/datafusion/common/benches/with_hashes.rs
index 0e9c53c896..ea75b32bdd 100644
--- a/datafusion/common/benches/with_hashes.rs
+++ b/datafusion/common/benches/with_hashes.rs
@@ -94,6 +94,24 @@ fn criterion_benchmark(c: &mut Criterion) {
array: sparse_union_array(BATCH_SIZE),
supports_nulls: false,
},
+ BenchData {
+ name: "sparse_union (2 types)",
+ array: sparse_union_of(
+ (0..2)
+ .map(|_| primitive_array::<Int64Type>(BATCH_SIZE))
+ .collect(),
+ ),
+ supports_nulls: false,
+ },
+ BenchData {
+ name: "sparse_union (utf8)",
+ array: sparse_union_of(
+ (0..5)
+ .map(|_| pool.string_array::<i32>(BATCH_SIZE))
+ .collect(),
+ ),
+ supports_nulls: false,
+ },
BenchData {
name: "dense_union",
array: dense_union_array(BATCH_SIZE),
@@ -354,6 +372,24 @@ fn sliced_array_benchmark(c: &mut Criterion) {
},
);
}
+
+ // Sliced Dense UnionArray
+ {
+ let full_array = dense_union_array(total_rows);
+ let sliced: ArrayRef = Arc::new(
+ full_array
+ .as_any()
+ .downcast_ref::<UnionArray>()
+ .unwrap()
+ .slice(slice_offset, slice_len),
+ );
+ c.bench_function(
+ &format!("dense_union_sliced: 1/{ratio} of {total_rows} rows"),
+ |b| {
+ do_hash_test_with_len(b, std::slice::from_ref(&sliced), slice_len);
+ },
+ );
+ }
}
}
@@ -430,23 +466,25 @@ fn map_array(num_rows: usize) -> ArrayRef {
}
fn sparse_union_array(num_rows: usize) -> ArrayRef {
+ sparse_union_of(
+ (0..5)
+ .map(|_| primitive_array::<Int64Type>(num_rows))
+ .collect(),
+ )
+}
+
+/// Sparse union whose rows each select a random child
+fn sparse_union_of(children: Vec<ArrayRef>) -> ArrayRef {
let mut rng = make_rng();
- let num_types = 5;
+ let num_types = children.len() as i32;
- let type_ids: Vec<i8> = (0..num_rows)
+ let type_ids: Vec<i8> = (0..children[0].len())
.map(|_| rng.random_range(0..num_types) as i8)
.collect();
- let (fields, children): (Vec<_>, Vec<_>) = (0..num_types)
- .map(|i| {
- (
- (
- i as i8,
- Arc::new(Field::new(format!("f{i}"), DataType::Int64, true)),
- ),
- primitive_array::<Int64Type>(num_rows),
- )
- })
- .unzip();
+ let fields = children.iter().enumerate().map(|(i, child)| {
+ let field = Field::new(format!("f{i}"), child.data_type().clone(), true);
+ (i as i8, Arc::new(field))
+ });
Arc::new(
UnionArray::try_new(
diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs
index 8da165e8fb..a51278b455 100644
--- a/datafusion/common/src/hash_utils.rs
+++ b/datafusion/common/src/hash_utils.rs
@@ -20,15 +20,15 @@
use arrow::array::types::{IntervalDayTime, IntervalMonthDayNano};
use arrow::array::*;
#[cfg(not(feature = "force_hash_collisions"))]
-use arrow::compute::take;
+use arrow::buffer::ScalarBuffer;
+#[cfg(not(feature = "force_hash_collisions"))]
+use arrow::compute::{nullif, take};
use arrow::datatypes::*;
#[cfg(not(feature = "force_hash_collisions"))]
use arrow::{downcast_dictionary_array, downcast_primitive_array};
use foldhash::fast::FixedState;
#[cfg(not(feature = "force_hash_collisions"))]
use itertools::Itertools;
-#[cfg(not(feature = "force_hash_collisions"))]
-use std::collections::HashMap;
use std::hash::{BuildHasher, Hash, Hasher};
/// [`RandomState`] is optimized for speed and suitable for hash tables and
@@ -826,67 +826,61 @@ fn hash_union_array(
};
if array.is_dense() {
- // Dense union: children only contain values of their type, so they're already compact.
- // Use the default hashing approach which is efficient for dense unions.
- hash_union_array_default(array, union_fields, child_hashing, hashes_buffer)
+ hash_dense_union_array(array, union_fields, child_hashing, hashes_buffer)
} else {
- // Sparse union: each child has the same length as the union array.
- // Optimization: only hash the elements that are actually referenced by type_ids,
- // instead of hashing all K*N elements (where K = num types, N = array length).
hash_sparse_union_array(array, union_fields, child_hashing, hashes_buffer)
}
}
-/// Default hashing for union arrays - hashes all elements of each child array fully.
-///
-/// This approach works for both dense and sparse union arrays:
-/// - Dense unions: children are compact (each child only contains values of that type)
-/// - Sparse unions: children have the same length as the union array
-///
-/// For sparse unions with 3+ types, the optimized take/scatter approach in
-/// `hash_sparse_union_array` is more efficient, but for 1-2 types or dense unions,
-/// this simpler approach is preferred.
+/// Slicing a dense union does not slice its children, so when their total length
+/// exceeds the union's, only the span its offsets reference is hashed.
#[cfg(not(feature = "force_hash_collisions"))]
-fn hash_union_array_default(
+fn hash_dense_union_array(
array: &UnionArray,
union_fields: &UnionFields,
child_hashing: &impl ChildHashing,
hashes_buffer: &mut [u64],
) -> Result<()> {
- let mut child_hashes: HashMap<i8, Vec<u64>> =
- HashMap::with_capacity(union_fields.len());
-
- // Hash each child array fully
- for (type_id, _field) in union_fields.iter() {
- let child = array.child(type_id);
- let mut child_hash_buffer = vec![0; child.len()];
- child_hashing.create_hashes([child], &mut child_hash_buffer)?;
-
- child_hashes.insert(type_id, child_hash_buffer);
+ let type_ids = array.type_ids();
+ let offsets = array.offsets().expect("dense union has offsets");
+
+ // `[start, end)` of each child to hash, indexed by type id
+ let mut spans = [(0, 0); 128];
+ for (type_id, _) in union_fields.iter() {
+ spans[type_id as usize].1 = array.child(type_id).len();
+ }
+ if spans.iter().map(|(_, end)| end).sum::<usize>() > array.len() {
+ spans = [(usize::MAX, 0); 128];
+ for (&type_id, &offset) in type_ids.iter().zip(offsets.iter()) {
+ let (start, end) = &mut spans[type_id as usize];
+ *start = (*start).min(offset as usize);
+ *end = (*end).max(offset as usize + 1);
+ }
}
- // Combine hashes for each row using the appropriate child offset
- // For dense unions: value_offset points to the actual position in the child
- // For sparse unions: value_offset equals the row index
- #[expect(clippy::needless_range_loop)]
- for i in 0..array.len() {
- let type_id = array.type_id(i);
- let child_offset = array.value_offset(i);
+ let mut child_hashes = vec![vec![]; spans.len()];
+ for (type_id, _) in union_fields.iter() {
+ let (start, end) = spans[type_id as usize];
+ if start < end {
+ let child = array.child(type_id).slice(start, end - start);
+ let hashes = &mut child_hashes[type_id as usize];
+ hashes.resize(end - start, 0);
+ child_hashing.create_hashes([child], hashes)?;
+ }
+ }
- let child_hash = child_hashes.get(&type_id).expect("invalid type_id");
- hashes_buffer[i] = combine_hashes(hashes_buffer[i], child_hash[child_offset]);
+ for ((hash, &type_id), &offset) in hashes_buffer.iter_mut().zip(type_ids).zip(offsets)
+ {
+ let start = spans[type_id as usize].0;
+ let child_hash = child_hashes[type_id as usize][offset as usize - start];
+ *hash = combine_hashes(*hash, child_hash);
}
Ok(())
}
-/// Hash a sparse union array.
-/// Sparse unions have child arrays with the same length as the union array.
-/// For 3+ types, we optimize by only hashing the N elements that are actually used
-/// (via take/scatter), instead of hashing all K*N elements.
-///
-/// For 1-2 types, the overhead of take/scatter outweighs the benefit, so we use
-/// the default approach of hashing all children (same as dense unions).
+/// Each row of a sparse union uses only one child, so each child is hashed
+/// only at the rows that select it.
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_sparse_union_array(
array: &UnionArray,
@@ -894,55 +888,84 @@ fn hash_sparse_union_array(
child_hashing: &impl ChildHashing,
hashes_buffer: &mut [u64],
) -> Result<()> {
- use std::collections::HashMap;
-
- // For 1-2 types, the take/scatter overhead isn't worth it.
- // Fall back to the default approach (same as dense union).
- if union_fields.len() <= 2 {
- return hash_union_array_default(
- array,
- union_fields,
- child_hashing,
- hashes_buffer,
- );
- }
-
let type_ids = array.type_ids();
+ let len = array.len();
- // Group indices by type_id
- let mut indices_by_type: HashMap<i8, Vec<u32>> = HashMap::new();
- for (i, &type_id) in type_ids.iter().enumerate() {
- indices_by_type.entry(type_id).or_default().push(i as u32);
- }
-
- // For each type, extract only the needed elements, hash them, and scatter back
- for (type_id, _field) in union_fields.iter() {
- if let Some(indices) = indices_by_type.get(&type_id) {
- if indices.is_empty() {
+ // Rows the child hashing skips keep a hash of 0
+ let mut child_hashes = vec![0; len];
+ if let Some(&type_id) = type_ids.first()
+ && type_ids.iter().all(|&t| t == type_id)
+ {
+ child_hashing.create_hashes([array.child(type_id)], &mut child_hashes)?;
+ } else {
+ let (starts, rows) = group_rows_by_type_id(type_ids);
+ for (type_id, _) in union_fields.iter() {
+ let (start, end) = (starts[type_id as usize], starts[type_id as usize + 1]);
+ if start == end {
continue;
}
let child = array.child(type_id);
- let indices_array = UInt32Array::from(indices.clone());
-
- // Extract only the elements we need using take()
- let filtered = take(child.as_ref(), &indices_array, None)?;
-
- // Hash the filtered array
- let mut filtered_hashes = vec![0u64; filtered.len()];
- child_hashing.create_hashes([&filtered], &mut filtered_hashes)?;
-
- // Scatter hashes back to correct positions
- for (hash, &idx) in filtered_hashes.iter().zip(indices.iter()) {
- hashes_buffer[idx as usize] =
- combine_hashes(hashes_buffer[idx as usize], *hash);
+ let variable_width = matches!(
+ child.data_type(),
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Binary
+ | DataType::LargeBinary
+ );
+ // Masking the other rows as null (skipped by hashing) avoids copying
+ // variable-width values but costs O(len) per child, so `take` wins for
+ // fixed-width values and for children selecting few rows
+ if variable_width && (end - start) * 32 >= len {
+ let mut other_rows = BooleanBufferBuilder::new(len);
+ other_rows.append_n(len, true);
+ for &row in &rows[start..end] {
+ other_rows.set_bit(row as usize, false);
+ }
+ let masked =
+ nullif(child, &BooleanArray::new(other_rows.finish(), None))?;
+ child_hashing.create_hashes([masked], &mut child_hashes)?;
+ } else {
+ let indices = UInt32Array::new(rows.slice(start, end - start), None);
+ let mut taken_hashes = vec![0; indices.len()];
+ child_hashing
+ .create_hashes([take(child, &indices, None)?], &mut taken_hashes)?;
+ for (&row, hash) in indices.values().iter().zip(taken_hashes) {
+ child_hashes[row as usize] = hash;
+ }
}
}
}
+ for (hash, child_hash) in hashes_buffer.iter_mut().zip(child_hashes) {
+ *hash = combine_hashes(*hash, child_hash);
+ }
+
Ok(())
}
+/// Counting sort of row indices by type id: the rows of `type_id` are
+/// `rows[starts[type_id]..starts[type_id + 1]]`
+#[cfg(not(feature = "force_hash_collisions"))]
+fn group_rows_by_type_id(type_ids: &[i8]) -> ([usize; 129], ScalarBuffer<u32>) {
+ let mut starts = [0; 129];
+ for &type_id in type_ids {
+ starts[type_id as usize + 1] += 1;
+ }
+ let mut total = 0;
+ for start in &mut starts {
+ total += *start;
+ *start = total;
+ }
+ let mut next = starts;
+ let mut rows = vec![0; type_ids.len()];
+ for (row, &type_id) in type_ids.iter().enumerate() {
+ rows[next[type_id as usize]] = row as u32;
+ next[type_id as usize] += 1;
+ }
+ (starts, rows.into())
+}
+
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_fixed_list_array(
array: &FixedSizeListArray,
@@ -2406,6 +2429,102 @@ mod tests {
assert_eq!(hashes[0], hashes[4]);
}
+ #[test]
+ #[cfg(not(feature = "force_hash_collisions"))]
+ fn create_hashes_for_union_arrays_match_selected_values() {
+ // In sparse unions both string children are masked, the others use `take`
+ let children = |lens: [usize; 5]| -> Vec<ArrayRef> {
+ let ints = |len| {
+ Int32Array::from_iter((0..len as i32).map(|v| (v % 3 != 1).then_some(v)))
+ };
+ let views = (0..lens[0]).map(|v| format!("longer than inline {}", v % 2));
+ let strings = (0..lens[1]).map(|v| (v % 3 != 1).then(|| format!("s{v}")));
+ let large = (0..lens[4]).map(|v| (v % 2 == 0).then(|| format!("l{v}")));
+ vec![
+ Arc::new(StringViewArray::from_iter_values(views)),
+ Arc::new(StringArray::from_iter(strings)),
+ Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("i", DataType::Int32, true)),
+ Arc::new(ints(lens[2])) as ArrayRef,
+ )])),
+ Arc::new(NullArray::new(lens[3])),
+ Arc::new(LargeStringArray::from_iter(large)),
+ ]
+ };
+ let fields: UnionFields = [0, 3, 5, 7, 9]
+ .into_iter()
+ .zip(children([0; 5]))
+ .map(|(id, c)| (id, Arc::new(Field::new("", c.data_type().clone(), true))))
+ .collect();
+ let type_ids = ScalarBuffer::from(vec![0, 3, 9, 0, 0, 0, 5, 7, 3, 9]);
+ let offsets = ScalarBuffer::from(vec![0, 0, 0, 1, 2, 3, 0, 0, 1, 1]);
+ let sparse = UnionArray::try_new(
+ fields.clone(),
+ type_ids.clone(),
+ None,
+ children([10; 5]),
+ )
+ .unwrap();
+ // Utf8 selecting too few rows to mask is hashed through `take` instead
+ let few_strings = (0..40).map(|i| if i == 20 { 3 } else { [0, 5, 7][i % 3] });
+ let sparse_few_strings = UnionArray::try_new(
+ fields.clone(),
+ few_strings.collect(),
+ None,
+ children([40; 5]),
+ )
+ .unwrap();
+ let dense = UnionArray::try_new(
+ fields,
+ type_ids,
+ Some(offsets),
+ children([4, 2, 1, 1, 2]),
+ )
+ .unwrap();
+
+ for union in [
+ sparse.clone(),
+ sparse.slice(1, 6),
+ sparse.slice(3, 3), // every row selects the same child
+ sparse_few_strings,
+ dense.clone(),
+ dense.slice(2, 4),
+ ] {
+ assert_union_hashes_match_selected_values(&union, |arrays, hashes| {
+ let random_state = RandomState::with_seed(0);
+ create_hashes(arrays.iter().copied(), &random_state, hashes).unwrap();
+ });
+ assert_union_hashes_match_selected_values(&union, |arrays, hashes| {
+ let hasher = BuildHasherDefault::<TestHasher>::default();
+ create_hashes_with_hasher(arrays.iter().copied(), &hasher, hashes)
+ .unwrap();
+ });
+ }
+ }
+
+ /// Asserts each union row contributes the hash of its selected value on its own
+ #[cfg(not(feature = "force_hash_collisions"))]
+ fn assert_union_hashes_match_selected_values(
+ union: &UnionArray,
+ create_hashes: impl Fn(&[&dyn Array], &mut [u64]),
+ ) {
+ let prev = Int64Array::from_iter_values(0..union.len() as i64);
+ let mut prev_hashes = vec![0; union.len()];
+ create_hashes(&[&prev], &mut prev_hashes);
+
+ let mut hashes = vec![0; union.len()];
+ create_hashes(&[&prev, union], &mut hashes);
+ for (i, hash) in hashes.into_iter().enumerate() {
+ let value = union
+ .child(union.type_id(i))
+ .slice(union.value_offset(i), 1);
+ let mut value_hash = [0];
+ create_hashes(&[&value], &mut value_hash);
+ let expected = combine_hashes(prev_hashes[i], value_hash[0]);
+ assert_eq!(hash, expected, "row {i} of {union:?}");
+ }
+ }
+
#[test]
#[cfg(not(feature = "force_hash_collisions"))]
fn create_hashes_for_sliced_run_array() -> Result<()> {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org