perf(hash-join): eliminate intermediate array allocations in probe-side collision filter#23209
Open
LiaCastaneda wants to merge 4 commits into
Open
perf(hash-join): eliminate intermediate array allocations in probe-side collision filter#23209LiaCastaneda wants to merge 4 commits into
LiaCastaneda wants to merge 4 commits into
Conversation
LiaCastaneda
changed the title
LiaCastaneda
force-pushed
the
perf/hash-join-no-take-in-probe
branch
from
June 29, 2026 13:03
ccfebda to
06503c2
Compare
LiaCastaneda
force-pushed
the
perf/hash-join-no-take-in-probe
branch
from
June 29, 2026 13:22
06503c2 to
e3b8172
Compare
LiaCastaneda
commented
Jun 29, 2026
| /// Example — keys `["cat", "cat", "dog"]`, next `[0, 1, 2]`: | ||
| /// row 1 → prev 0: "cat"=="cat" ✓ | ||
| /// row 2 → prev 1: "dog"!="cat" → return true (collision found) | ||
| fn detect_key_collisions<T>( |
Contributor
Author
There was a problem hiding this comment.
I checked in the other engine (Trino) and there when a build row is inserted, Trino only inserts rows that share the same key, so by construction, every chain is already "pure" and hash collisions are not possible.
This happens because Trino uses a hash table that resolves key equality at insert time. DataFusion's update_from_iter only receives hashes and row indices (it has no access to key values) so it chains all rows in the same hash bucket together regardless of whether they share a key or not.
Contributor
Author
There was a problem hiding this comment.
Note doing the same in DF would be non trivial because it would require modifying update_from_iter trait signature to accept key columns alongside the hashes, and updating all call sites accordingly.
Contributor
There was a problem hiding this comment.
Yes I think this could be faster at least for single column keys / with type / join type specialization.
The reason I think we / I pursued this design is that take is (much) faster than individual dispatch (and most recent literature also seems to implement it in a separate pass) and probably easier to implement all the join types.
Also combining it with filter is a bit easier with the indices -> filter indices -> collision check, etc.
LiaCastaneda
marked this pull request as ready for review
June 29, 2026 15:24
LiaCastaneda
force-pushed
the
perf/hash-join-no-take-in-probe
branch
from
June 30, 2026 10:12
c3b077f to
d647868
Compare
Contributor
|
run benchmark tpch10 tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (d647868) to bde8e5b (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (d647868) to bde8e5b (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
|
It looks it is mostly faster but also regresses on some |
Contributor
|
run benchmark tpch10 tpcds |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (d647868) to bde8e5b (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (d647868) to bde8e5b (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
joroKr21
pushed a commit
to coralogix/arrow-datafusion
that referenced
this pull request
Jul 1, 2026
## Which issue does this PR close? Adds benchmark query for apache#23237 Useful to test changes in apache#23209 ## Rationale for this change Q16–Q22 existed in `hj.rs` but were missing `.benchmark` files, so they were invisible to `bench.sh` and the CI benchmark bot. Q23 is a new query reproducing a skewed high-fanout pattern where a single partition does nearly all the join work -- almost all probe rows carry the same key. Statistics are disabled to force `Partitioned` mode. ## What changes are included in this PR? - `benchmarks/src/hj.rs`: adds Q23 - `benchmarks/sql_benchmarks/hj/benchmarks/q16–q23.benchmark`: sql harness files for Q16–Q23 so they run via `bench.sh` and the CI benchmark bot ## Are these changes tested? These are benchmark additions, so no tests are needed. ## Are there any user-facing changes? No
Replace `equal_rows_arr` (Arrow take + eq_dyn_null + FilterBuilder, O(matched_pairs) allocations) with an in-place `JoinKeyComparator` loop. On collision-free build sides — detected once at build time by scanning the `next` chain for adjacent pairs with distinct keys — skip the per-pair recheck entirely: probe rows form consecutive runs in the output buffer, so we check the chain head once and accept/reject the whole run. This cuts key comparisons from F (fanout) per probe row down to 1 on uniform-key build sides, producing a 2.4× speedup on high-fanout string-key joins (Q23, SF100: 1.01s → 0.42s join_time). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
LiaCastaneda
force-pushed
the
perf/hash-join-no-take-in-probe
branch
from
July 1, 2026 08:16
d647868 to
f3e1b56
Compare
Contributor
Author
I'm trying to run tpcds SF1 again locally but the results seem very flaky :/ I'm thinking we could try adding a fast-path comparison for cases that compare primitive types and join on a single column, as you suggested -- that may be why some queries regress (if those are regressions and not noise) I rebased the branch before doing the above. Can we run the |
Contributor
Author
|
For TPCH SF10 results does not look like noise though:
|
Contributor
Author
|
Some results of I ran the whole First run Second run Most queries show improvements but I should probably look into Q11 and Q13, will try to profile those |
Contributor
|
run benchmark hj |
Contributor
|
run benchmark tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (f3e1b56) to 9e8dd76 (merge-base) diff using: hj File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/hash-join-no-take-in-probe (f3e1b56) to 9e8dd76 (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagehj — base (merge-base)
hj — branch
File an issue against this benchmark runner |
Contributor
Author
It's strange this one regresses because it uses |
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
3 participants
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?
Rationale for this change
When HashJoinExec builds the probe the hash table, it collects all candidate (build_idx, probe_idx) pairs and then rechecks every pair with
equal_rows_arr(shows in the profile, taking 20% of the query CPU) to filter out hash collisions. This recheck materializes intermediate Arrow arrays sized to the number of matched pairs (take+eq_dyn_null+FilterBuilder), making it O(matched_pairs) in both allocations and comparisons.For high-fanout joins (many build rows per distinct key), matched_pairs = probe_rows × fanout. At fanout 78× over 2.3M probe rows this produces ~176M pairs — and the recheck runs on all of them, even though nearly all pass.
This cost is most visible when:
What changes are included in this PR?
Step 1: Replace
equal_rows_arrwith an in-placeJoinKeyComparatorloop , eliminating the O(matched_pairs) Arrow allocations in the fallback path.Step 2: At build time, scan the next chain once to detect whether the map is collision-free (all adjacent linked pairs share the same key). If so, the probe phase checks once per run of consecutive same-probe-idx pairs and accepts the entire run.
Are these changes tested?
Existing tests pass, I also added some tests for the new function that detects hash collisions
detect_key_collisionsI did some profiling on the added Q23 (this query replicates the level of skew where just 1 partition does the entire join):
We can see the CPU for the HashJoin was cut in half.
Also the benchmarks results show Q23 runs ~x2.5 times faster:
Are there any user-facing changes?
No