QinXi-ai opened a new pull request, #25592:
URL: https://github.com/apache/datafusion/pull/25592
## Which issue does this PR close?
Closes #25392.
This is a **draft alternative implementation** to #25434, which already
addresses the same issue. It is not an incremental fix for that PR. I am
posting this separately to make the different approach and its tradeoffs
reviewable; it is not ready for merge.
## Rationale for this change
Generic hash joins allocate lookup buckets for every build row even when
many rows share a join key. The duplicate-row chain must keep every row, but
the lookup table only needs one entry per distinct hash. Overallocating that
table can reject a join whose payload, row chain, and useful lookup entries fit
in the memory pool.
This implementation grows the generic lookup table with the observed hashes
and uses a bounded sample to reduce repeated growth for high-cardinality
inputs. It preserves duplicate matches and performs reservation checks before
allocating either the initial table or a replacement table.
## What changes are included in this PR?
- Start reservation-backed generic maps with a small lookup table while
sizing the row chain for all build rows.
- Charge the initial allocation before allocating. On growth, reserve both
old and new tables, including hashbrown's trailing control group, then settle
to the actual retained allocation.
- When a full table receives an existing hash, update its chain without
requesting growth capacity.
- Sample up to 1,024 hashes using fixed stack space. The first qualifying
batch can pre-size for that batch; a later batch with unseen, distinct samples
can request full-row capacity. A rejected speculative reservation falls back to
incremental growth. Stratified jitter avoids the fixed-stride aliasing
demonstrated by periodic inputs.
- Keep reservation ownership with the map, propagate allocation errors
through the insertion callers, and report retained map storage in
`build_mem_used`. This metric is not a measurement of the temporary reservation
peak or process RSS.
- Add bounded-pool and sampling regressions, a real string-key HashJoin
regression in both CollectLeft and Partitioned modes, and a native Criterion
build benchmark with repeated, unique, periodic, and skewed input orders.
The specialized perfect-hash `ArrayMap` path is unchanged. Existing generic
auxiliary scope maps also use the updated constructor. There is no new spilling
behavior, and this does not solve payload concatenation accounting.
The allocation estimate is tied to hashbrown 0.17's layout; tests compare it
with actual allocation sizes. Sampling is a capacity heuristic, not a guarantee
for arbitrary distributions. It can still reserve more than a particular
distribution requires.
## What is the testing strategy for this PR?
Unit tests exercise both index widths, duplicate chains at a full table with
an exact memory limit, reservation-before-allocation, old/new table overlap,
cleanup, rejected sampling hints, repeated batches, and periodic hashes. The
execution test uses string keys to force the generic path and checks every
resulting row, including duplicate probe matches.
The new benchmark can be run with:
```sh
cargo bench --locked -p datafusion-physical-plan --features test_utils \
--bench hash_join_semi_anti -- 'generic_hash_join_build/'
```
For a base comparison, copy the same benchmark file to the base checkout and
use separate target directories. It builds from 200,000 UTF-8 rows, varies
batch size between 8,192 and 200,000, and probes with one absent key to
emphasize construction rather than result materialization.
Earlier validation on base `b300cea226af7935c9f915221925b9e07f5f316e`,
before updating this branch to current main:
- 2,256 physical-plan library tests passed.
- Four independent allocation-audit regressions passed. Full-table duplicate
updates fit in 2,644 / 3,096 bytes for U32 / U64; the observed growth peak and
reservation were both 10,560 bytes; a zero-budget rejection made no allocation
of 400,000 bytes or more.
- A separate release microbenchmark measured map allocation, insertion, and
destruction, not full SQL execution. For 200,000 rows with all distinct hashes,
this implementation was **14–15% slower than that upstream base**, despite
pre-sizing. The tested 8-, 112-, and 8,192-distinct-hash cases were faster than
that base. These are preliminary pre-rebase observations, not performance
claims for the submitted revision or the native benchmark; the remaining
high-cardinality regression is one reason to keep this PR in draft.
The submission branch is based on
`a948ff63807a10dd61c3d20fd78f84ebb3e71461`, which includes the Arrow 60 upgrade
and the existing row-chain accounting fix. Validation uses the installed Rust
1.98.0 Windows MSVC toolchain; the repository's Rust 1.98.1 pin and Cargo.lock
are unchanged.
Submission-branch checks:
- `cargo test --offline --locked -p datafusion-physical-plan --lib`: **2,326
passed, 0 failed**.
- Formatting and `git diff --check`: passed.
- `cargo clippy --offline --locked -p datafusion-physical-plan --all-targets
--features test_utils -- -D warnings -A clippy::unnecessary_semicolon`: passed,
including the benchmark target. The allowance is for the unchanged
`datafusion/common/src/rounding.rs:257`; the unsuppressed check fails there on
this compiler. MSVC also emits a linker-stdout warning that is not governed by
`-D warnings`.
- Full `cargo clippy --all-targets --all-features -- -D warnings`:
attempted, blocked while building `snmalloc-sys` by MSVC C4819/C2220 (code page
936 interpreting a source file).
- `dev/rust_lint.sh`: attempted, stopped at its missing `python3`
prerequisite in the local Bash environment.
- Extended workspace tests and native Criterion timing comparisons have not
been completed for this revision. Full CI and performance validation remain
required before leaving draft.
## Are there any user-facing changes?
Duplicate-heavy generic hash joins can use smaller memory pools while
preserving all matching rows. No SQL syntax or configuration changes are
introduced.
**Rust API changes:** `JoinHashMapType::update_from_iter` and the shared
insertion helper now return `Result<()>`; the trait gains a required `size`
method and a default `reserve_from_batch` method, and the insertion helper
accepts an optional reservation. Downstream implementations or callers of these
public, primarily internal interfaces need adjustment. Please apply the `api
change` label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
