2010YOUY01 opened a new issue, #27398:
URL: https://github.com/apache/datafusion/issues/27398
There are two major opportunities to improve hash join performance:
1. Existence join short-circuiting: For the query below, `t1.v1` may have
heavy duplicates. However, we only need one build-side row satisfying both join
predicates, so probing can stop after the first qualifying match.
```sql
SELECT *
FROM t1
RIGHT SEMI JOIN t2
ON t1.v1 = t2.v1 AND t1.v2 > t2.v2
```
2. Duplicate-heavy build side: Suppose the build side has 10 distinct keys,
each with 1M duplicate rows. The existing hash table uses linked-list collision
chains, so fetching all 1M matching rows incurs expensive pointer chasing.
## Proposal
### Physical hash table redesign
The high-level idea is to store build-side row IDs for equal join keys
contiguously. The physical design has three components:
- SwissTable: Reuse `hashbrown::HashTable`, mapping each distinct key to an
entry in `MatchHandleArray`.
- MatchHandleArray: Each entry is either `One(row_id)` or `Multi(group_id)`.
The multi-match case stores only a reference, without expanding the matched
rows.
- MatchedGroups: Stores row IDs for each multi-match group contiguously,
using an offsets/payload layout.
<img width="747" height="558" alt="image"
src="https://github.com/user-attachments/assets/38f094a1-e029-433e-92d2-80da72dc407b";
/>
This indirection also makes existence join short-circuiting easier: we can
iterate over active probe-row `MatchHandle`s in round-robin fashion, evaluating
candidates in batches and stopping each probe row after its first qualifying
match.
### Refactor for clearer structure
The current hash join implementation has two major problems:
1. Regular joins and existence joins share the same execution path, despite
having different optimization opportunities. Separating them should simplify
both implementations.
- See https://github.com/apache/datafusion/issues/26361
2. The execution state machine and low-level hash table logic are mixed
together, making the implementation difficult to maintain.
Ideally, regular and existence joins should use separate streams, with
clearer layering. From bottom to top:
1. JoinIndexMap: Maps a probe row to matching build-side row IDs (equi-join
equality).
- e.g. `key1 -> [row0], key2 -> [row1, row3, row4]`
- Multi-row matches are only a handle, caller will materialize them
incrementally.
2. BatchedJoinProcessor: Performs the pipeline `equi-join matches ->
residual join filter -> output batch construction`, processing matches in
chunks bounded by `batch_size`.
3. Entry point: The stream state machine drives `BatchedJoinProcessor`.
`JoinIndexMap` provides the abstraction boundary for the proposed hash table
redesign.
## Implementation
The implementation is split into three PRs:
1. Refactor hash join
- Split regular joins into a dedicated stream.
- Introduce the `BatchedJoinProcessor` layer (`JoinIndexMap` is deferred
to PR 2).
2. Implement the contiguous hash table layout behind the new `JoinIndexMap`
abstraction.
3. Split existence joins into a dedicated stream, reusing `JoinIndexMap` and
implementing short-circuiting over matched rows.
--
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]
