andygrove opened a new pull request, #26447:
URL: https://github.com/apache/datafusion/pull/26447
## Which issue does this PR close?
- Part of #17494.
## Rationale for this change
When a build-side key repeats, `HashJoinExec` keeps the build rows of each
key in a chain: the map holds the first
row and `next[row]` the one after it. A probe row follows the chain with one
load for every match, and each load
depends on the one before, so nothing overlaps them. Once the build side no
longer fits in the cache, most of those
loads are misses. That is what makes joins with many build rows per key,
such as the join of `catalog_sales` and
`inventory` in TPC-DS q72, several times slower than a sort-merge join: in
the reproduction posted on #17494,
85% of the hash join's samples are in
`ArrayMap::get_matched_indices_with_limit_offset`, with `traverse_chain`
inlined into it.
## What changes are included in this PR?
The build rows of each repeated key are stored next to each other, so that a
probe row reads its matches in order:
- `ArrayMap` groups the rows by key with a counting sort when some key
repeats: the rows of the key in `slot` are
`rows[starts[slot]..starts[slot + 1]]`. It uses the same memory as the
chain, one slot array and one index per
build row.
- `JoinHashMapU32` and `JoinHashMapU64` group the rows after the last build
batch is inserted, when the hash values
have at least two rows each on average. The map then holds the position of
each hash value's group, which starts
with its number of rows, and the chain is freed. The groups take one more
index per distinct hash value than the
chain. The build reserves that memory first, and keeps the chain if it
cannot.
- Grouping follows 16 chains at once, so that their loads overlap. Following
one chain at a time cost about 75 ns a
build row. It holds up to 4096 rows of each chain until the chain ends,
and measures longer chains to place them
in a second pass, so that it holds at most 16 x 4096 indices besides the
groups.
- The probe copies a probe row's matches with one slice copy
(`emit_grouped_rows`). `ProbeOffset::MidChain::next`
is then the position in the groups to resume from.
Matches come out in the same order as before, so the join's output does not
change.
## Benchmarks
`datafusion-cli` (`release-nonlto`) on an AMD 7950X3D (16 cores), two runs
each.
The reproduction from #17494: 64M build rows over 640K keys, about 100 rows
per key, 96M probe rows and a residual
filter that keeps 6% of the 9.6B candidate pairs, with 16 partitions:
| Keys | main | this PR |
Sort-merge join, for reference |
| ------------------------------ | -------------- | -------------- |
------------------------------ |
| Dense (`ArrayMap`) | 59.1 s, 60.0 s | 17.4 s, 17.6 s | 10.4 s
to 11.8 s |
| Sparse (`JoinHashMap`) | 76.4 s, 75.5 s | 33.5 s, 33.2 s | 9.4 s
to 13.4 s |
| Unique, dense, 64M build rows | 1.79 s, 1.61 s | 1.77 s, 1.63 s |
|
| Unique, sparse, 64M build rows | 1.82 s, 1.62 s | 1.78 s, 1.68 s |
|
What grouping costs when there is little to probe: the same 64M build rows,
and 16K probe rows:
| Keys | main | this PR |
| ---------------------- | --------------------- | --------------------- |
| Dense (`ArrayMap`) | 0.489, 0.373, 0.379 s | 0.574, 0.421, 0.478 s |
| Sparse (`JoinHashMap`) | 0.686, 0.458, 0.485 s | 0.737, 0.571, 0.566 s |
`dfbench hj` on TPC-H SF10, the median of 12 iterations over three runs that
alternate between main and this PR.
The benchmark turns off join reordering, so Q4 to Q15 build on `lineitem`,
about 600 rows per key, and probe
`supplier`. Where every probe row matches (Q5, Q7, Q9, Q11 and Q13) they are
15% to 34% faster, and the high fan-out
queries Q22 to Q25 are 11% to 22% faster. Where 10% of the probe rows match
(Q4 to Q14, even), the probe returns too
few matches to pay for grouping 60M build rows, and they are 1% to 10%
slower. The other queries are within 6%.
<details>
<summary>All 25 queries</summary>
| Query | Scenario
| main ms | this PR ms | Change |
| ----- |
------------------------------------------------------------------- | ------- |
---------- | ------ |
| Q1 | density=1 prob hit=1 25*1.5M
| 3.0 | 3.1 | +2% |
| Q2 | density=0.026 prob hit=1 25*1.5M
| 6.7 | 6.9 | +2% |
| Q3 | density=1 prob hit=1 100K*60M
| 53.4 | 52.2 | -2% |
| Q4 | density=1 prob hit=0.1 100K*60M
| 208.3 | 215.5 | +3% |
| Q5 | density=0.75 prob hit=1 100K*60M
| 407.8 | 267.1 | -34% |
| Q6 | density=0.75 prob hit=0.1 100K*60M
| 254.2 | 256.7 | +1% |
| Q7 | density=0.5 prob hit=1 100K*60M
| 398.8 | 261.5 | -34% |
| Q8 | density=0.5 prob hit=0.1 100K*60M
| 248.9 | 254.5 | +2% |
| Q9 | density=0.2 prob hit=1 100K*60M
| 398.6 | 266.6 | -33% |
| Q10 | density=0.2 prob hit=0.1 100K*60M
| 251.3 | 259.2 | +3% |
| Q11 | density=0.1 prob hit=1 100K*60M
| 402.1 | 277.3 | -31% |
| Q12 | density=0.1 prob hit=0.1 100K*60M
| 252.2 | 262.9 | +4% |
| Q13 | density=0.01 prob hit=1 100K*60M
| 479.7 | 407.4 | -15% |
| Q14 | density=0.01 prob hit=0.1 100K*60M
| 279.5 | 308.4 | +10% |
| Q15 | density=0.2 prob hit=0.1 100K (20% dups)*60M
| 221.0 | 215.8 | -2% |
| Q16 | density=1 prob hit=1 25*1.5M RightSemi
| 3.7 | 3.6 | -2% |
| Q17 | density=1 prob hit=1 100K*60M RightSemi
| 56.4 | 55.4 | -2% |
| Q18 | density=1 prob hit=0.1 100K*60M RightSemi
| 57.7 | 54.4 | -6% |
| Q19 | density=1 prob hit=1 25*1.5M RightAnti
| 3.0 | 3.0 | +2% |
| Q20 | density=1 prob hit=1 100K*60M RightAnti
| 43.1 | 43.9 | +2% |
| Q21 | density=1 prob hit=0.1 100K*60M RightAnti
| 67.5 | 64.9 | -4% |
| Q22 | density=1 prob hit=0.01 100K (fanout 100)*60M RightSemi
| 47.6 | 42.5 | -11% |
| Q23 | density=1 prob hit=1 32K (fanout~78)*2.3M long keys count
| 114.8 | 92.9 | -19% |
| Q24 | density=1 prob hit=1 3K (single hot bucket)*20K long keys count
| 308.7 | 241.2 | -22% |
| Q25 | density=1 prob hit=1 20K (fanout~78 multi key)*240K multi key
count | 176.0 | 152.7 | -13% |
</details>
The `hash_join_probe` criterion benchmark, whose build keys are unique, is
within 2% of main for 11 of its 13
cases, as close as two runs of main are to each other.
`right_semi_clustered_10pct` was 5% slower and
`right_semi_sparse_100k_all_hit` 6% to 8% slower, in a single run.
## What is the testing strategy for this PR?
- New unit tests compare grouped maps with the chain or with a brute-force
result: `JoinHashMapU32` and
`JoinHashMapU64` with rows inserted in batches from the last to the first
and one row left out, as for a NULL key,
probed with limits from 1 to 8192 and with NULL probe keys, and `ArrayMap`
with repeated, NULL and out of range
keys. Others cover chains long enough for the second pass of grouping, the
size of the groups, the memory freed,
and that short chains are not grouped.
- The existing hash join tests, the join fuzz tests (`--features
extended_tests`), the join and memory limit tests
of `core_integration`, and all sqllogictests pass, with grouping on
wherever keys repeat.
## Are there any user-facing changes?
No. `JoinHashMapType`, which is public for internal use only, gains
`grouped_rows_size` and `group_rows`, both with
default implementations.
## Notes for reviewers
- Most of what is left of the gap to the sort-merge join in the reproduction
is gathering build columns at the
matched rows, which are scattered across the build batch. Ordering the
build batch by key would make that
sequential too, but it changes the order of build-side output for outer,
semi and anti joins, so it is not part
of this PR.
- Grouping is decided at build time. Deciding it from how much the probe has
followed chains would avoid the cost
when the probe side is small, but the map is shared across probe streams
in `CollectLeft` mode, and would need
interior mutability and memory accounting during the probe.
--
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]
