Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25434 +/- ##
==========================================
+ Coverage 82.33% 82.42% +0.09%
==========================================
Files 1137 1139 +2
Lines 431887 435702 +3815
Branches 431887 435702 +3815
==========================================
+ Hits 355580 359125 +3545
- Misses 54811 54863 +52
- Partials 21496 21714 +218 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @sunchao , 2 suggestions
| // Keep the old allocation charged during rehash. The fixed-size | ||
| // allowance covers control-group padding; at least eight elements | ||
| // also covers hashbrown's minimum bucket sizes. | ||
| let minimum = (table.len() + additional).max(chunk_rows); |
There was a problem hiding this comment.
Second growth always jumps to num_rows, and it triggers once len + 8192 > 14336, so any build with more than ~6,144 distinct hashes gets the same table as main. Measured with 1M Utf8 rows: 6,000 distinct → 278,536 B table; 6,200 distinct → 35,651,592 B; 10K and 100K distinct → 35,651,592 B.
Cheap fix: shrink after the build. With this, 6,200 → 139 KB, 10K → 278 KB, 100K → 2.2 MB, ≥500K unchanged, and all 6 new tests still pass. Peak is still row-count-sized when the pool admits it
break;
}
+ // The one-shot row-count preallocation overshoots when only a fraction
+ // of the rows carry distinct hashes; return the unused buckets.
+ if table.capacity() / 4 > table.len() {
+ let old_bytes = table.allocation_size();
+ table.shrink_to(table.len(), |&(hash, _)| hash);
+ reservation.shrink(old_bytes - table.allocation_size());
+ }
drop(hashes);There was a problem hiding this comment.
Addressed in 57791ea. After releasing the hash scratch buffer, tables with capacity greater than four times their entry count are compacted. The replacement allocation is admitted while the old table remains charged, and the overlap is included in peak memory. If admission fails, compaction is skipped and the completed join still succeeds. As you noted, this reduces retained memory; the earlier row-count preallocation peak can still occur.
| use crate::test::TestMemoryExec; | ||
|
|
||
| #[tokio::test] | ||
| async fn compact_hash_build_with_duplicates_and_nulls() -> Result<()> { |
There was a problem hiding this comment.
No case with HASH_BUILD_CHUNK_ROWS < distinct << rows, so nothing pins table size in the mid-cardinality range. Add one that bounds the final allocation.
let rows = 1_000_000;
let distinct = 10_000;
let batches: Vec<_> = (0..rows)
.step_by(HASH_BUILD_CHUNK_ROWS)
.map(|start| {
let end = (start + HASH_BUILD_CHUNK_ROWS).min(rows);
RecordBatch::try_from_iter([(
"key",
Arc::new(Int32Array::from_iter_values(
(start..end).map(|i| (i % distinct) as i32),
)) as ArrayRef,
)])
})
.collect::<Result<_, _>>()?;
let on = vec![Arc::new(Column::new("key", 0)) as PhysicalExprRef];
let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 30));
let reservation = MemoryConsumer::new("mid cardinality").register(&pool);
let (table, _next) = build_compact_hash_map::<u32>(
&batches,
&on,
rows,
HASH_JOIN_SEED.random_state(),
NullEquality::NullEqualsNothing,
&reservation,
&mut 0,
)?;
assert!(
table.allocation_size()
<= estimate_memory_size::<(u64, u32)>(4 * distinct, size_of::<JoinHashMapU32>())?
);There was a problem hiding this comment.
Added the 1,000,000-row / 10,000-key final-allocation regression in 57791ea. It fails against the original PR head and passes with compaction, for both u32 and u64 row indices. It also checks every row chain and exact retained accounting. I extended the growth test with a constrained 100,000-key case that skips compaction while preserving the same heads/chains, and compare reported peak memory with PeakRecordingPool. All seven focused tests pass both normally and with forced hash collisions.
|
Speculate-to-
Disabling this block restores base time (5.85 ms vs 6.03 ms), so the cost is the compaction rehash. Acceptable as a trade for ~33 MB less retained during probe, but please cover it in the bench and note it in the description. for distribution in [
"duplicates",
+ "mid_20k",
+ "mid_100k",
"mostly_unique", "duplicates" => row % 64,
+ "mid_20k" => row % 20_000,
+ "mid_100k" => row % 100_000,
"mostly_unique" => row % 900_000,@sunchao I think we should try to improve mid_20k and mid 100k case 🤔 |
|
Addressed in commit 048610951, including the requested The slowdown came from allocating a table for all one million rows, then moving its entries into a smaller table. At the first growth, the builder now samples up to 2,048 positions across the buffered string/binary column and uses that only to choose capacity. Every row is still inserted normally. Uncertain estimates retain full preallocation; underestimated estimates can grow again. Smaller sampled allocations reserve their final compaction workspace upfront. The insertion loop also uses a generic internal iterator while preserving the public boxed API. Against exact upstream base Seven additional stress cases also improved in both paired runs: the initial-table size boundary, skewed inputs with unique rows at either end, and grouped 100k keys. No timing regression reproduced in the final PR benchmark runs or these stress cases. The new tests also cover two failures found during development: a tight-memory LeftAnti query that needs a visited-row bitmap, and an underestimated sample whose denied full-growth attempt previously retained an oversized table. Both regressions pass and verify reservation cleanup; separate focused tests validate row chains and FIFO order. All 19 focused tests pass normally and with forced hash collisions. Extended workspace tests, all 521 SQL logic files, core/CLI tests, strict Clippy, and the full repository lint suite pass locally. The unique-key All GitHub checks for |
Which issue does this PR close?
Closes #25392.
Rationale for this change
Generic hash joins reserve lookup-table capacity for every build row, even when many rows share a key. A join over 10,000 customer records with state
GAmust retain all 10,000 matching rows, but needs only one lookup entry for their hash. Reserving unused buckets can make otherwise small joins fail withResourcesExhausted.This PR sizes the lookup table around the build keys while preserving every row in the separate index chain. The regression test joins 65,536 rows containing repeated
GA,NULL, andCAkeys in a 2 MiB memory pool; the old implementation exhausts the pool, while the compact build preserves all duplicate matches and both NULL-matching modes.The reported 20,000- and 100,000-key benchmarks exposed another important case. With one million rows, growing directly to the full row count and shrinking afterward allocates an oversized table, then scans it and moves its populated entries into a smaller table. That extra compaction work explains why reducing retained memory could still slow construction. These distributions are now included in the benchmark, and eligible builds estimate their required capacity before making that large allocation.
What changes are included in this PR?
The generic hash table starts with capacity for one bounded chunk. At its first growth, a sample of up to 2,048 row positions across the buffered input can estimate capacity for a single non-null string or binary column. Sampling reads values directly without copying them or evaluating expressions again. When the sample cannot support a useful estimate, or the key representation is unsupported, the builder tries full-row preallocation. Every row still passes through normal hashing and insertion; the estimate affects capacity only.
A smaller sampled allocation is admitted together with the workspace needed to compact it. If that combined reservation cannot fit, the builder tries full-row preallocation, then bounded growth. An underestimated sample can fall back to full preallocation once. If only a replacement table fits, the builder releases its partial table and rebuilds from the buffered rows. Final compaction also handles bounded growth after a denied full fallback, preserving memory needed by subsequent join work.
The compact builder passes its iterator directly to a generic internal insertion helper, removing its per-chunk iterator box and allowing compiler specialization. The existing public boxed helper keeps its signature and delegates to that implementation.
The memory pool accounts for row indices, hash scratch, table overlap, and reserved compaction workspace.
build_mem_usedreports their tracked reservation peak, which can include workspace reserved before it is allocated. It is not process RSS and does not account for every allocation during concatenation of the buffered payload.For the one-million-unique-key case,
build_mem_usedchanges from 55,592,952 to 59,937,088 bytes. Most of the increase comes from the pre-existing 4,000,000-byte row-index chain and 65,536-byte hash scratch becoming charged. Growth also adds 278,544 bytes of actual temporary overlap for the initial small table, plus a small fixed bookkeeping allowance in the counter. Both revisions retain the same 39,651,656-byte hash index, including its row-index chain, after construction.What is the testing strategy for this PR?
Focused tests cover duplicate and NULL semantics, row-chain order, both index widths, empty input, sliced and empty batches, all supported string/binary representations, and dictionary, nested, multi-column, and computed keys. Memory tests compare the reported peak with an independently recording pool and verify cleanup after denied allocations and rebuilds.
Two regressions exercise downstream memory availability: constrained LeftAnti joins that need a visited-row bitmap, and an underestimated sample whose denied full-table fallback previously retained twice the necessary table allocation. Uniform and grouped intermediate-cardinality inputs verify that the construction peak stays below a full-row table allocation.
Final-revision validation:
force_hash_collisionsanddatafusion-common/force_hash_collisionsenabled.cargo fmt --all: passed.cargo clippy --all-targets --all-features -- -D warnings: passed../dev/rust_lint.sh: passed, including documentation and dependency checks.async-compression0.4.44,compression-codecs0.4.40,toml1.1.5,uuid1.26.0). The original lockfile is restored and is not changed by this PR; all current-head GitHub checks passed with the published dependency versions (38 successful checks, 3 optional skips).The seven Criterion cases build from one million UTF-8 rows in batches of 8,192 and probe with one absent key. Input generation is outside timing; execution consumes and drops the full join.
The last two cases contain the same multiset: 500,000 unique keys and 500,000 rows repeating 64 other keys, in opposite input order. These measurements establish behavior for the tested workloads; sampling remains a capacity heuristic rather than a universal performance guarantee.
Compared exact base
62f039f0d9583054b81d47b2f08b6728a14ac452with048610951fbcf5cd7aa566bfd288f5485fd10109on AMD EPYC Milan/Linux x86_64 using Rust 1.98.1, separate target directories, and the identical submitted seven-case harness. Both use the default release profile (lto=true, one codegen unit), CPU 0, one Tokio worker, 30 samples, 2 s warmup, and 5 s target measurement. Process order was base → PR → PR → base. The table preserves both paired passes because both binaries showed distinct fast/slow timing modes; it does not pool them into a single speedup.Additional runs with the submitted harness confirmed the mostly-unique, unique, and both input-order cases. A duplicate-first control had one long sample, which is retained in the results; a preselected follow-up with 80 samples and 10 s target measurement improved by 7.1% and 5.4% in its two reversed pairs. Earlier input-order controls overlapped an unrelated build that saturated the host; those observations are preserved but excluded from the performance conclusion.
A separate non-LTO comparison of exact base → reviewed PR head
57791ea2→ updated PR reproduced the old head's middle-cardinality slowdown. At 20k distinct keys, the three means were 66.07 → 80.97 → 30.30 ms; at 100k, 71.54 → 93.11 → 29.06 ms. All seven updated-PR cases improved against the base in this screen. This diagnostic harness selects the requested case before allocating inputs, so its absolute timings are kept separate from the submitted-harness LTO table.Seven further non-LTO stress cases also improved in both reversed pairs: 6,144/6,145 distinct keys around the initial-table boundary; half unique rows mixed with 8,000 repeated keys in both orders; a 1.5% unique tail mixed with one repeated key in both orders; and grouped 100k keys. The final candidate showed no reproducible timing regression in the seven PR cases or this stress matrix. Reservation peaks are reported separately from timing and retained storage; this is not a claim that every memory counter decreases.
To reproduce each revision after installing the same toolchain/dependencies:
CARGO_TARGET_DIR=/path/to/separate-target cargo bench --locked \ -p datafusion-physical-plan --features test_utils \ --bench hash_join_semi_anti --no-run TOKIO_WORKER_THREADS=1 taskset -c 0 /path/to/benchmark-binary \ --bench '^hash_join_build/' --sample-size 30 \ --warm-up-time 2 --measurement-time 5 --noplotCopy the benchmark file from the updated PR onto the base checkout so the requested middle-cardinality cases exist in both, and reverse the run order for the second pair.
Are there any user-facing changes?
Duplicate-heavy generic hash joins can execute with smaller memory pools while preserving their results. Intermediate-cardinality string and binary builds can avoid full-row lookup-table allocation. No configuration or public API changes are required. The specialized perfect-hash
ArrayMapand auxiliary NULL-aware map construction paths are unchanged.