perf: reduce generic hash join memory for low-cardinality builds

[sunchao]

Which issue does this PR close?

Closes #25392.

Rationale for this change

A generic hash join can fail with ResourcesExhausted because it reserves lookup-table capacity for every build row, even when those rows share very few join keys. This PR reduces that unnecessary reservation so duplicate-heavy joins can execute with smaller memory pools.

For example, suppose the build input contains 10,000 customer records with state code GA. In an equality join, a probe for GA must return all 10,000 matching records, but the lookup index needs only one entry to locate their group. DataFusion already stores one lookup entry per distinct hash value and links the corresponding row indices together, checking actual keys when probing to handle collisions. Yet it preallocates the lookup table as though every row had a different hash. For a column such as state, status, or category, this can leave a large allocation of unused buckets.

The regression test makes the consequence concrete. It builds 65,536 rows with keys repeating GA, NULL, and CA, each carrying a distinct payload. With a 2 MiB memory pool, the old implementation fails because its lookup-table reservation alone exceeds that limit. With this change, the same join completes with identical results to a run with ample memory, including all duplicate matches and both NULL-matching modes.

A larger local benchmark also shows the reduction in reported build memory:

Generic hash-join build input	Before	After
1,000,000 rows over 64 string keys	45.45 MiB	15.59 MiB

That is 65.7% lower build_mem_used for this case. This measures tracked build memory, not process RSS or execution time.

What changes are included in this PR?

The generic hash join now starts its lookup table with a small, bounded capacity and expands it as it processes the build input. The row-index chain still has room for every build row, so the join preserves every duplicate match. When many rows share a few keys, the lookup table can stay small even as the row data grows.

When the initial table needs to grow, the join tries full row-count preallocation once to avoid repeated resizing for mostly unique keys. If the memory pool cannot admit that allocation, it continues with smaller requests based on the hashes already stored and the next chunk of rows. This balances memory savings for repeated keys with construction efficiency for unique keys; some intermediate key distributions can still reach full preallocation.

Resizing must also work within the memory limit. Normally, the existing table remains live while a larger replacement is allocated. If the replacement would fit but both tables cannot coexist, the join releases its partial index and rebuilds it from the already buffered rows. This trades additional build work for a lower temporary memory requirement, without discarding rows or adding spilling.

The memory pool now accounts for the row-index chain, temporary hash buffers, and overlapping tables during growth, and the build-memory metric includes temporary peaks. This accounting is necessary to respect the pool's limit during construction. It can also increase reported memory for mostly unique inputs because the old implementation left some allocations uncharged.

The change applies to the generic lookup table in HashJoinExec; the specialized perfect-hash ArrayMap and auxiliary NULL-aware maps keep their existing construction paths. It does not add final table compaction or address memory accounting for concatenating the buffered build payload.

What is the testing strategy for this PR?

The six new tests in joins::hash_join::compact_hash_map::tests cover the 2 MiB duplicate/NULL regression, row-chain ordering across uneven batches, dictionary/nested/computed keys, constrained growth and rebuilding, failed-allocation cleanup, both row-index widths, and empty/all-NULL inputs. The constrained join regression fails with ResourcesExhausted on unchanged base 62f039f0d and passes with this change.

Separate local review probes covered inner, left, and full joins in partitioned and shared-build modes, both NULL-matching settings, and a join filter. All 24 head runs matched the corresponding base results. Constrained runs exercised allocation fallback and released all tracked memory. These probes are not part of the committed suite.

Recorded local validation on Rust 1.98.1:

• 2,262 physical-plan tests and 2,479 core/CLI tests and doctests passed; 57 doctests were ignored.
• All 521 SQL logic files completed successfully. The extended workspace suite passed 11,919 tests, with 8 ignored.
• All six new tests passed with forced hash collisions.
• Formatting, strict all-target/all-feature Clippy, and the full ./dev/rust_lint.sh suite passed, including documentation compilation with warnings denied.

Benchmark results

The five hash_join_build Criterion cases compare the exact base 62f039f0d with PR head a063ee768. Each builds a generic hash join from 1,000,000 UTF-8 rows, in batches of 8,192, and probes with one absent key. The benchmark verifies that the entire build is consumed and no output rows are produced. Input generation is outside the timed region; timing includes creating, executing, and dropping the join.

Build-key distribution (1M rows)	Base time	PR time	Elapsed change	Tracked build memory, base → PR
64 repeated keys	12.80 ms	8.66 ms	-32.3%	45.45 → 15.59 MiB
900,000 distinct keys	71.31 ms	70.06 ms	-1.8%	52.96 → 57.10 MiB
1,000,000 distinct keys	69.81 ms	69.43 ms	-0.5%	53.02 → 57.16 MiB
Unique rows first, repeated keys second	55.36 ms	55.09 ms	-0.5%	49.20 → 53.34 MiB
Repeated keys first, unique rows second	57.29 ms	55.01 ms	-4.0%	49.20 → 53.34 MiB

Times are the average of two Criterion run means per revision; negative elapsed changes mean less time. The last two cases contain the same multiset: 500,000 unique keys plus 500,000 rows repeating 64 other keys, presented in opposite input order.

The repeated-key case used 32.3% less time and 65.7% less tracked build memory in these local runs. Its per-run means were 12.94/12.66 ms before and 8.66/8.66 ms after. The other timing changes were much smaller and should be treated as local observations, not general query-speed guarantees. Their higher reported memory includes newly charged row-index and scratch memory, as well as temporary overlap while the table grows. build_mem_used is a tracked build-memory peak, not process RSS.

Benchmark environment and reproduction

Measurements used an AMD EPYC-Milan x86_64 Linux VM, Rust 1.98.1, and the optimized release-nonlto profile (optimization level 3, LTO disabled). Each revision was built in its own target directory with identical benchmark source and the same compatible dependency lock described below; the resulting binaries had different SHA-256 hashes. Timings ran sequentially after compilation, pinned to CPU 0 with one Tokio worker, in base → PR → PR → base order.

Each pass used 30 samples, a 2-second warm-up, and a 5-second measurement target; Criterion extended sampling when needed. The table reports the arithmetic average of the two per-run mean estimates. Memory figures were identical across both passes for each revision.

With the benchmark present in each revision's worktree, the equivalent Cargo invocation is:

CARGO_TARGET_DIR=target/bench TOKIO_WORKER_THREADS=1 taskset -c 0 \
  cargo +stable bench --profile release-nonlto --locked \
  -p datafusion-physical-plan --features test_utils \
  --bench hash_join_semi_anti -- hash_join_build \
  --sample-size 30 --warm-up-time 2 --measurement-time 5 --noplot

Local registry availability required validation from main immediately before the sqlparser 0.63 update, with compatible patch versions of four dependencies (async-compression 0.4.44, compression-codecs 0.4.40, toml 1.1.5, uuid 1.26.0). These lockfile adjustments are not included in the PR. CI provides validation of the published lockfile and the merge with current main; its current results are tracked in the PR checks.

Are there any user-facing changes?

Duplicate-heavy generic hash joins can complete with smaller memory pools while preserving their results. No configuration or public API changes are required.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 84.50704% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.33%. Comparing base (62f039f) to head (a063ee7).
⚠️ Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
...sical-plan/src/joins/hash_join/compact_hash_map.rs	88.03%	8 Missing and 6 partials ⚠️
...tafusion/physical-plan/src/joins/hash_join/exec.rs	68.00%	7 Missing and 1 partial ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #25434    +/-   ##
========================================
  Coverage   82.33%   82.33%            
========================================
  Files        1137     1138     +1     
  Lines      431887   432492   +605     
  Branches   431887   432492   +605     
========================================
+ Hits       355580   356112   +532     
- Misses      54811    54862    +51     
- Partials    21496    21518    +22     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.