Repository navigation
Reuse dictionary value hashes in RepartitionExec - #26816
Closed
Rich-T-kid wants to merge 9 commits into
Closed
Rich-T-kid wants to merge 9 commits into
Rich-T-kid wants to merge 9 commits into
Conversation
Contributor
Author
|
run benchmark tpch tpcds clickbench_partitioned env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
#26816 (comment) @asolimando I think I found it 😄 |
Contributor
Author
|
run benchmark tpcds env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"Results will be posted here when complete File an issue against this benchmark runner |
Rich-T-kid
commented
Oct 27, 2026
Rich-T-kid
left a comment
Rich-T-kid
left a comment
Contributor
Author
There was a problem hiding this comment.
this PR itself shows good promise but I think a PR that breaks apart the hash vs spreading of the keys will make the other PRs related to #26780 far smaller & simpler & non dependent on each other
| #[cfg(not(feature = "force_hash_collisions"))] | ||
| use arrow::{downcast_dictionary_array, downcast_primitive_array}; | ||
| use foldhash::fast::FixedState; | ||
| #[cfg(not(feature = "force_hash_collisions"))] |
Contributor
Author
There was a problem hiding this comment.
we can trim this. we just need to break apart the actually hashing of values from the spreading of the keys.
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 14d1378 (14d1378) to 5d9b7fd diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "5d9b7fd45ba0e27225effa91fd4e7a0ae4eeeff9"
changed:
ref: "14d1378f2bdc9b4f64a5c43dbf189d031a706fa0"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Rich-T-kid
force-pushed
the
rich-T-kid/repartition-dictionary-hash-cache
branch
from
October 27, 2026 18:38
14d1378 to
81d2bb6
Compare
Contributor
Author
|
run benchmark tpch tpcds clickbench_partitioned env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d" |
github-actions
Bot
added
common
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark tpch
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 81d2bb6 (81d2bb6) to acbad14 diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "acbad1400bb75b5837a2c389ea622d5ff4b6f77c"
changed:
ref: "81d2bb625063861415e9382ceff01df526e9e55d"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Rich-T-kid
force-pushed
the
rich-T-kid/repartition-dictionary-hash-cache
branch
from
October 27, 2026 19:40
81d2bb6 to
d4b5712
Compare
Member
It looks to me that you are working the same angle as @jayzhan211 in #26703 (Q18/Q33/Q34 ~ 1.8–1.9×, total −28 % vs −29 % here), maybe you two could join forces on this one? I didn't have time to go through all the details of both PRs, but if you check and still have doubts, one quick way to check is to rebase this on #26703 and re-run Also, both runs compare against a baseline with |
Rich-T-kid
force-pushed
the
rich-T-kid/repartition-dictionary-hash-cache
branch
from
October 27, 2026 19:56
d4b5712 to
4b4472c
Compare
Rich-T-kid
changed the title
Contributor
Author
|
run benchmark tpch tpcds clickbench_partitioned env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
baseline:
ref: "4ede5833f68dad5bc8169492dedc5749fcc4b8d5"
changed:
ref: "4b4472c7bcc38d5b478cc7fbf64c6780efc979e4" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 4b4472c (4b4472c) to 4ede583 diff Run configurationrun benchmark tpcds
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "4ede5833f68dad5bc8169492dedc5749fcc4b8d5"
changed:
ref: "4b4472c7bcc38d5b478cc7fbf64c6780efc979e4"Results will be posted here when complete File an issue against this benchmark runner |
jayzhan211
requested changes
Nov 7, 2026
jayzhan211
left a comment
jayzhan211
left a comment
Contributor
There was a problem hiding this comment.
@Rich-T-kid thanks for this! The hashing itself looks right: 4800 randomized batches matched create_hashes exactly across key types, nulls, slices and nested values, and hits save up to 96% of the hash time at D=65536. I had approved earlier but have to change that, because two things need to land here: a test that detects the reuse, and reserving the cache in the memory pool. I know #26826 is waiting on this, and I've run both changes locally, so they should be quick to apply.
Optional cleanup in the description: Depends on #26817 can go now that the helper is part of this PR, and a ClickBench run at the final head would replace the note that the results predate it.
| assert_hashes_match(&mut hasher, &second_batch, &random_state)?; | ||
| assert_eq!( | ||
| assert_cached_values(&hasher, 1, &dictionary_values), | ||
| value_hash_allocation |
Contributor
There was a problem hiding this comment.
These pointer checks also pass on a miss because the Vec is recycled, so nothing here shows the rehash was skipped. I made the cache never hit and made BatchPartitioner ignore the hasher, and all the tests still passed in both cases. Since reuse is the point of this PR and #26814 asks for it to be tested directly, could you add this in this PR? It poisons the cached hashes, so only a real hit can return the sentinel, and it fails when the cache never hits. It also covers equal values behind a new Arc and a dropped Arc:
#[cfg(not(feature = "force_hash_collisions"))]
#[test]
fn caching_array_hasher_skips_rehash_only_for_same_values_arc() -> Result<()> {
let random_state = RandomState::with_seed(0);
let values: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
let batch = vec![string_dictionary(vec![0, 1], &values)];
let mut hasher = caching_hasher(&batch, &random_state);
let mut hashes = vec![0; 2];
hasher.create_hashes(&batch, &mut hashes)?;
let fresh = hashes.clone();
// Poison the cached hashes: only a cache hit can return the sentinel.
const SENTINEL: u64 = 0xDEAD_BEEF;
hasher.dictionaries[0].as_mut().unwrap().1.fill(SENTINEL);
let same_arc = vec![string_dictionary(vec![1, 0], &values)];
hasher.create_hashes(&same_arc, &mut hashes)?;
assert_eq!(hashes, vec![SENTINEL; 2]);
// Equal contents behind a different `Arc` is a miss.
let equal_values: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
let new_arc = vec![string_dictionary(vec![0, 1], &equal_values)];
hasher.create_hashes(&new_arc, &mut hashes)?;
assert_eq!(hashes, fresh);
// A dropped values array must not be matched by a later allocation.
hasher.dictionaries[0].as_mut().unwrap().1.fill(SENTINEL);
drop(new_arc);
drop(equal_values);
for _ in 0..1000 {
let reused: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
let batch = vec![string_dictionary(vec![0, 1], &reused)];
hasher.create_hashes(&batch, &mut hashes)?;
assert_eq!(hashes, fresh);
hasher.dictionaries[0].as_mut().unwrap().1.fill(SENTINEL);
}
Ok(())
}The pointer assertions can go once this is in. The test in my comment on size() covers the BatchPartitioner wiring.
Comment on lines
+120
to
+122
| hashes.resize(values.len(), 0); | ||
| // Fresh hashing leaves null value slots unchanged. | ||
| hashes.fill(0); |
Contributor
There was a problem hiding this comment.
Optional cleanup: clearing first means each slot is only written once, and the randomized comparison against create_hashes still passes.
Suggested change
| hashes.resize(values.len(), 0); | |
| // Fresh hashing leaves null value slots unchanged. | |
| hashes.fill(0); | |
| // Fresh hashing leaves null value slots unchanged. | |
| hashes.clear(); | |
| hashes.resize(values.len(), 0); |
| pub struct CachingArrayHasher { | ||
| /// Stored with the cache so value hashes cannot cross hash seeds. | ||
| random_state: RandomState, | ||
| /// One entry per column. Weak identity avoids retaining dictionary values. |
Contributor
There was a problem hiding this comment.
Optional cleanup: the address comparison is only safe because the Weak also keeps the allocation reserved, which is worth stating so nobody later swaps it for a raw pointer.
Suggested change
| /// One entry per column. Weak identity avoids retaining dictionary values. | |
| /// One entry per column. Weak identity avoids retaining dictionary values, | |
| /// and keeps their allocation reserved so a new `Arc` can't reuse the address. |
Comment on lines
24
to
25
| #[cfg(test)] | ||
| use crate::hash_utils::create_hashes; |
Contributor
There was a problem hiding this comment.
Optional cleanup: only the test at the bottom of this file uses this import, so it can move into mod tests and drop the cfg. Clippy passes with this.
-#[cfg(test)]
-use crate::hash_utils::create_hashes;
@@
mod tests {
+ use crate::hash_utils::create_hashes;
Contributor
|
Since this PR isn't ready yet, I'd prefer to move forward with #26703 first, though there may still be some things left to do there. |
Contributor
Author
|
addressing the comments now!
I actually think we should go with both. I think for repartionExec we should cache the value computation as that plays into the strengths of dictionary encoding and then for other portions of the code base like |
alamb
pushed a commit
to geoffreyclaude/datafusion
that referenced
this pull request
Nov 7, 2026
## Which issue does this PR close? - Closes apache#26696. ## Rationale for this change Queries that group on a high-cardinality dictionary column slow down as the group count grows. On ClickBench `hits_partitioned` with `enable_rle_to_dictionary = true`, this PR takes the suite from 78.8 s to 42.8 s, with Q33 and Q34 about 3.5x faster and peak memory slightly lower ([bot run](apache#26703 (comment))). A hash aggregate builds its output once and emits it in `batch_size` slices, and slicing a `DictionaryArray` keeps every value. Every slice therefore carries every group's value, so hashing, functions such as `lower`, the repartition coalescer and the final aggregate do work per slice that grows with the group count. ## What changes are included in this PR? Start reading at `next_batch` in `aggregates/aggregate_hash_table/common.rs`. - **Aggregate output, the main gain.** A slice with more values than rows now holds only the values its rows reference. No threshold is needed: each group is emitted once, so the cost is fixed per group. - **A bound for other batches.** Filters and scans also keep whole dictionaries, so `hash_dictionary` and the new `map_dictionary_values` handle at most `MAX_DICTIONARY_VALUES_PER_ROW` (64) values per row and compact past that. Too high only gives up an improvement, while too low makes cheap work slower: `lower` and `upper` break even at about 40 values per row with keys in random order. The filter and scan queries I measured showed no visible difference, so this is a guard, not a measured speedup. - **`compact_dictionary`** does both, taking each referenced value once. `take(values, keys)` copies per row, and multi-column group keys repeat dictionary keys, so it can overflow `i32` offsets, need more rows than the key type holds, and copy a long value thousands of times. - **`map_dictionary_values`** replaces `dictionary.with_values(f(values))` in the string functions, Spark `hex`, `array_agg`, `regexp_replace` and `substr`, and `clippy.toml` disallows calling `with_values` directly. If `f` fails, it retries on only the referenced values, because `f` can see values no row references. This replaces `substr`'s own fallback. Each aggregate slice now has its own values `Arc`, so the hash cache proposed in apache#26816 would miss on these slices. The two haven't been measured together. ## Benchmarks | ClickBench partitioned | main | this PR | |---|---|---| | Total | 78.8 s | 42.8 s | | Q33 | 14.9 s | 4.4 s | | Q34 | 14.8 s | 4.1 s | | Q18 | 7.6 s | 2.3 s | | Peak memory | 13.1 GiB | 12.7 GiB | Q35, Q32 and Q40 read 1.05x to 1.10x slower by minimum. None groups on a dictionary column, so this PR only adds a type check per output batch for them, and the bot's noise is about 5-7% per query. <details><summary>Local runs on a busy 12-core laptop: `lower`, memory limits, filter queries</summary> Results were identical on every query. Treat small differences as noise. `lower` over a dictionary GROUP BY, which ClickBench doesn't cover: ```sql SELECT lower(s), c FROM ( SELECT "SearchPhrase" s, COUNT(*) c FROM hits WHERE "SearchPhrase" <> '' GROUP BY s ) ORDER BY c DESC, 1 LIMIT 10; ``` | partitions | 12 | 24 | 32 | 48 | 64 | |---|---|---|---|---|---| | ratio to main | 0.08x | 0.14x | 0.16x | 0.24x | 0.22x | Q33 under `--memory-limit`, one run each: | limit | main | this PR | |---|---|---| | 12g | fails: TopK resources exhausted | 6.3 s | | 8g | 23.6 s | 16.4 s | | 6g | 31.2 s | 19.4 s | | 4g | 20.2 s | 14.3 s | Filter and scan queries using `reverse`, `initcap`, `LIKE` and hashing after a filter, with the bound at 40: 0.94x to 1.05x of main. </details> ## What is the testing strategy for this PR? - `dictionary.slt` compares each function on aggregate output and on filter output with the same function on the decoded value. The comparison is projected, because a filter on the group key is pushed below the aggregate. - Unit tests cover `next_batch`, `map_dictionary_values` and hashing: repeated keys, NULL groups, more rows than the key type holds, and a value too long to copy per row. - The `dictionary_sliced` case in `datafusion/common/benches/with_hashes.rs` covers 8 and 128 values per row. ## Are there any user-facing changes? No behavior changes. `datafusion_common::utils` gains `compact_dictionary`, `map_dictionary_values` and `MAX_DICTIONARY_VALUES_PER_ROW`.
Contributor
|
@Rich-T-kid since you asked about merging this one: #26703 has merged, and it affects this PR. It now gives each aggregate output slice with more values than rows only the values its rows reference (ClickBench partitioned 78.8 s to 42.8 s). Those slices then have their own values The cache should still help where batches share a values Could you rebase this on main and rerun |
jayzhan211
requested changes
Nov 7, 2026
Contributor
There was a problem hiding this comment.
@Rich-T-kid thanks for the update, the reuse test and the reservation both do their job. I'm still requesting changes, mainly for the miss path in hash_utils.rs (4x slower than main on one query, inline).
Following up on my note about #26703, I've now measured the two together by merging this head with main locally, and the overlap I expected is there: Q33 0.91x, Q34 0.95x and Q18 0.87x, against 0.94x for a Utf8 control, on a busy laptop. Summed repartition_time on main is already under 1 s for Q16, Q18, Q33 and Q34. The cache does still help a scan hashed on a dictionary key: 0.54 s to 0.45 s on a join over 12 files of 200K values each. You asked about going with both, and that works if the cache stays below #26703's bound and leaves larger dictionaries to it, which is what the inline change does.
Once you've rebased as I asked, could you apply these changes? I'll review that head first, and the clickbench_partitioned rerun can wait until then. It will give the description its numbers and settle whether the helper stays here or moves into #26826. Depends on #26817 can go then too.
| downcast_dictionary_array! { | ||
| array => { | ||
| let values = array.values(); | ||
| if !dictionary.as_ref().is_some_and(|(cached, _)| { |
Contributor
There was a problem hiding this comment.
A cache miss hashes every value, which skips the per-row bound #26703 added to hash_dictionary. So with more than 64 values per row and a changing values Arc, this is slower than main: 1.60 s to 6.67 s (4.17x) on a LIMIT subquery that interleaves 3 files with 2M values each. Could you leave those misses to hash_dictionary, after rebasing? With this change that query takes 1.64 s, and the 0.83x win on a shared Arc below the bound stays. Shared values above the bound stop being cached, which also caps the retained buffer at 4 MiB per column for 8192-row batches.
-use crate::utils::compact_oversized_dictionary;
+use crate::utils::{MAX_DICTIONARY_VALUES_PER_ROW, compact_oversized_dictionary};
@@ CachingArrayHasher::hash_array
- if !dictionary.as_ref().is_some_and(|(cached, _)| {
+ let cached = dictionary.as_ref().is_some_and(|(cached, _)| {
std::ptr::addr_eq(cached.as_ptr(), Arc::as_ptr(values))
- }) {
+ });
+ // Above the bound `hash_dictionary` only hashes the values
+ // these rows reference, so a miss must not hash all of them.
+ if !cached && values.len() > MAX_DICTIONARY_VALUES_PER_ROW * array.len()
+ {
+ hash_single_array(
+ array,
+ &self.random_state,
+ hashes_buffer,
+ rehash,
+ )?;
+ return Ok(());
+ }
+ if !cached {Boundary test and the SQL repro
#[cfg(not(feature = "force_hash_collisions"))]
#[test]
fn caching_array_hasher_does_not_fill_above_values_per_row_bound() -> Result<()> {
let random_state = RandomState::with_seed(0);
let dictionary_of = |len: usize| -> ArrayRef {
let values: ArrayRef = Arc::new(StringArray::from_iter_values(
(0..len).map(|i| format!("{i}")),
));
string_dictionary(vec![0], &values)
};
// One row may hash up to the bound of values: this fills the cache.
let at_bound = vec![dictionary_of(MAX_DICTIONARY_VALUES_PER_ROW)];
let mut hasher = caching_hasher(&at_bound, &random_state);
assert_hashes_match(&mut hasher, &at_bound, &random_state)?;
assert!(hasher.dictionary_hashes(0).is_some());
// One value more is a miss that must not hash, or retain, all values.
let above_bound = vec![dictionary_of(MAX_DICTIONARY_VALUES_PER_ROW + 1)];
assert_hashes_match(&mut hasher, &above_bound, &random_state)?;
let cached_len = hasher.dictionaries[0].as_ref().unwrap().1.len();
assert_eq!(cached_len, MAX_DICTIONARY_VALUES_PER_ROW);
Ok(())
}Write this COPY three times, to b/f0.parquet, b/f1.parquet and b/f2.parquet:
COPY (SELECT 'value_' || CAST(floor(random()*2000000) AS BIGINT) AS k, value AS id
FROM generate_series(1,8000000))
TO 'b/f0.parquet' STORED AS PARQUET OPTIONS(
'format.dictionary_page_size_limit' '200000000',
'format.max_row_group_size' '100000000',
'format.dictionary_enabled' 'true',
'format.compression' 'uncompressed');
SET datafusion.execution.parquet.enable_rle_to_dictionary = true;
SET datafusion.execution.target_partitions = 12;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = false;
CREATE EXTERNAL TABLE tb STORED AS PARQUET LOCATION 'b/';
SELECT COUNT(*) FROM (SELECT k FROM tb LIMIT 24000000) a
JOIN (SELECT DISTINCT k FROM tb WHERE id = 5) b ON a.k = b.k;Min of 4 interleaved runs, same result (43) on all sides. Hash time alone per 8192-row batch, 8-byte values, main vs a miss on every batch: 65 values per row 78 vs 399 us, 512 per row 95 vs 3068 us.
| ); | ||
| } | ||
|
|
||
| #[test] |
Contributor
There was a problem hiding this comment.
This test panics under force_hash_collisions, because the cache is never filled there, so the merge queue's hash-collisions job would fail. That one's on me, since it came from my suggestion.
Suggested change
| #[test] | |
| #[cfg(not(feature = "force_hash_collisions"))] | |
| #[test] |
| } | ||
| timer.done(); | ||
| } | ||
| partitioner.reserve_hash_cache(&hash_cache_reservation); |
Contributor
There was a problem hiding this comment.
Nothing tests this call: I removed it and all 80 repartition tests still pass, so the cache could silently fall out of the memory pool again. Could you add this test in this PR? It fails with the call removed and passes otherwise.
End-to-end test that records the hash cache reservation
#[cfg(not(feature = "force_hash_collisions"))]
#[tokio::test]
async fn hash_repartition_reserves_dictionary_hash_cache() -> Result<()> {
use datafusion_execution::memory_pool::{
MemoryPool, MemoryReservation, UnboundedMemoryPool,
};
/// Records the largest reservation of the hash cache consumers.
#[derive(Debug, Default)]
struct CachePeakPool {
inner: UnboundedMemoryPool,
cache_peak: AtomicUsize,
}
impl CachePeakPool {
fn record(&self, reservation: &MemoryReservation, additional: usize) {
if reservation.consumer().name().contains("hash cache") {
self.cache_peak.fetch_max(
reservation.size() + additional,
AtomicOrdering::Relaxed,
);
}
}
}
impl Display for CachePeakPool {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_str("CachePeakPool")
}
}
impl MemoryPool for CachePeakPool {
fn name(&self) -> &str {
"cache_peak"
}
fn grow(&self, reservation: &MemoryReservation, additional: usize) {
self.record(reservation, additional);
self.inner.grow(reservation, additional);
}
fn shrink(&self, reservation: &MemoryReservation, shrink: usize) {
self.inner.shrink(reservation, shrink);
}
fn try_grow(
&self,
reservation: &MemoryReservation,
additional: usize,
) -> Result<()> {
self.record(reservation, additional);
self.inner.try_grow(reservation, additional)
}
fn reserved(&self) -> usize {
self.inner.reserved()
}
}
let dictionary_type =
DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
let schema = Arc::new(Schema::new(vec![Field::new("key", dictionary_type, true)]));
let values: ArrayRef = Arc::new(StringArray::from(vec!["a", "bb", "ccc"]));
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(DictionaryArray::<Int32Type>::try_new(
Int32Array::from(vec![0, 1, 2]),
values,
)?)],
)?;
let pool = Arc::new(CachePeakPool::default());
let runtime = RuntimeEnvBuilder::new()
.with_memory_pool(Arc::clone(&pool) as Arc<dyn MemoryPool>)
.build_arc()?;
let task_ctx = Arc::new(TaskContext::default().with_runtime(runtime));
let exec = TestMemoryExec::try_new_exec(&[vec![batch]], Arc::clone(&schema), None)?;
let exec = RepartitionExec::try_new(
exec,
Partitioning::Hash(vec![col("key", &schema)?], 1),
)?;
let output = crate::common::collect(exec.execute(0, task_ctx)?).await?;
assert_eq!(output.iter().map(RecordBatch::num_rows).sum::<usize>(), 3);
assert!(pool.cache_peak.load(AtomicOrdering::Relaxed) >= 3 * size_of::<u64>());
Ok(())
}| let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(8)); | ||
| let reservation = MemoryConsumer::new("cache").register(&pool); | ||
| partitioner.reserve_hash_cache(&reservation); | ||
| assert!(cached(&partitioner).1.is_none()); |
Contributor
There was a problem hiding this comment.
Optional cleanup: this also checks that the reservation shrinks after the cache is dropped, which no test covers today. It fails if the reservation.resize call goes away.
Suggested change
| assert!(cached(&partitioner).1.is_none()); | |
| assert!(cached(&partitioner).1.is_none()); | |
| assert_eq!(pool.reserved(), cached(&partitioner).0); |
Contributor
Author
|
run benchmark clickbench_partitioned env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/repartition-dictionary-hash-cache (401b3ee) to d38fde4 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/repartition-dictionary-hash-cache (401b3ee) to d38fde4 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmark clickbench_partitioned env: |
2 similar comments
Contributor
Author
|
run benchmark clickbench_partitioned env: |
Contributor
Author
|
run benchmark clickbench_partitioned env: |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/repartition-dictionary-hash-cache (bee0600) to 9d15cf0 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpch tpcds clickbench_partitioned |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark tpcds
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark tpcds
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing bee0600 (bee0600) to 9d15cf0 diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
ref: "9d15cf079fe9646a967196fd6688f35be1e36c6b"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "false"
changed:
ref: "bee06006b516e96df5ca17a4e7a1859fbf15b5a0"
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
Author
|
This is good progress compared with the original paired RLE off/on run. ClickBench went from 71.16s vs 19.00s (3.75x slower, with Q33/Q34 9.20x/8.93x slower) to 40.14s vs 19.41s here (2.07x overall, with Q33/Q34 2.55x/2.43x slower). TPC-H also improved from ~11% slower to 6.6%, and TPC-DS from ~4% to 3.3%. There is still a gap, and this spans other changes since #24227, but we are moving in the right direction. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #26814.
Depends on #26817.
Rationale for this change
Dictionary batches often share one values array while only their keys change.
RepartitionExeccurrently hashes all dictionary values again for every batch.What changes are included in this PR?
BatchPartitionerowns one schema-configuredCachingArrayHasherfor all hash expressions. It is initialized from the first evaluated key batch and reused across later batches. Dictionary columns reuse value hashes while the valuesArcis unchanged and replace the cache when it changes. Non-dictionary columns use the existing hashing path, and composite-key hashing keeps the same combine order.Results
The original hosted GKE run compared
9f6f79f7withd648959f, with RLE-to-dictionary enabled on both sides.ClickBench improves Q33 by 1.93x, Q18 by 1.90x, Q34 by 1.84x, Q16 by 1.62x, and Q17 by 1.59x. TPC-DS Q72 improves by 1.06x. No primary-summary query regresses.
These results predate the helper and main rebases. The operator logic is unchanged, but this is not final-head benchmark evidence.
What is the testing strategy for this PR?
One end-to-end test covers shared dictionary values across batches, null keys, and partition routing. It passes on the rebased head.
Are there any user-facing changes?
No.