perf: emit unmatched build rows in batch_size chunks in HashJoinExec - #25028
perf: emit unmatched build rows in batch_size chunks in HashJoinExec#25028jayzhan211 wants to merge 1 commit into
Conversation
For join types that emit build-side rows after the probe side is exhausted (Left, Full, LeftAnti, LeftSemi, LeftMark), HashJoinExec materialized every final build row as one RecordBatch. That batch is unbounded by batch_size and not covered by the memory reservation, and LimitedBatchCoalescer passes batches larger than batch_size / 2 through untouched, so a selective anti join over a large build side emitted a single build-side-sized batch downstream. Add an EmitUnmatchedBuildRows stream state that snapshots the visited bitmap once (after every probe partition reported completion) and emits at most batch_size final rows per poll, matching NestedLoopJoinExec. Remove the now-unused get_final_indices_from_bit_map helpers. Add tests for chunked emission across join types and batch sizes and for a fetch reached mid-chunk, plus benchmark cases with a 1M-row build side.
| let emit_visited = join_type == JoinType::LeftSemi; | ||
| let mut build_indices = Vec::with_capacity(batch_size.min(num_rows - start)); | ||
| let mut idx = start; | ||
| while idx < num_rows && build_indices.len() < batch_size { |
There was a problem hiding this comment.
Unmatched build-side rows are now emitted in batch_size chunks instead of a single batch over the whole build side.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #25028 +/- ##
==========================================
- Coverage 81.71% 81.71% -0.01%
==========================================
Files 1127 1127
Lines 416072 416150 +78
Branches 416072 416150 +78
==========================================
+ Hits 339997 340058 +61
- Misses 56090 56102 +12
- Partials 19985 19990 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
@jayzhan211, thanks for working on this. This looks good to me.
I like the approach of introducing a dedicated final-build-row emission state and snapshotting the visited bitmap once probing is complete. Emitting the remaining build-side rows in batch-size chunks avoids materializing the entire final result at once and keeps the output bounded by the configured batch size.
The chunking and fetch-limit tests cover the new behavior well, and the build-side-output benchmarks are a useful addition. The cleanup of the now-unused index helpers also makes sense with the new incremental approach.
I don't see any blocking issues or additional changes needed from my side. Thanks!
|
Thanks @kosiew 🚀 |
3 participants
Which issue does this PR close?
Rationale for this change
For join types that emit build-side rows after the probe side is exhausted (
Left,Full,LeftAnti,LeftSemi,LeftMark),HashJoinExeccomputed the final indices over the whole build side and materialized them as oneRecordBatchbefore handing it to the output coalescer.That batch is not bounded by
batch_size, and it is not covered by the memory reservation. Worse,LimitedBatchCoalescerconfigures arrow'sBatchCoalescerwithbiggest_coalesce_batch_size = batch_size / 2, which passes any larger batch through untouched. So aLEFT ANTIjoin over a 10M-row build side with few matches emitted a single ~10M-row batch downstream, regardless ofdatafusion.execution.batch_size.NestedLoopJoinExecalready emits its unmatched build rows inbatch_sizechunks; this PR bringsHashJoinExecin line.What changes are included in this PR?
HashJoinStreamgets a new state,EmitUnmatchedBuildRows, entered fromExhaustedProbeSideby the last probe partition. It holds aBooleanBuffersnapshot of the visited bitmap (taken once, after every partition reported completion, so the lock is not held while emitting) and a cursor.next_final_indices_chunkscans the snapshot from the cursor and returns at mostbatch_sizefinal indices per call (LeftMarkemits every row, so its chunks are plain ranges). Null-awareLeftAnti/LeftMarkpost-processing andfetchhandling are unchanged and now run per chunk.input_batchesis still bumped once for the final phase andinput_rowsonce per chunk, so metric values are identical to before.get_final_indices_from_bit_map/get_final_indices_from_shared_bitmapinjoins/utils.rshad no other callers and are removed.hash_join_semi_anti.rswith a 1M-row build side and a 100K-row probe side (left_semi_build1m_h10,left_anti_build1m_h10,left_build1m_h10).Benchmark (this branch vs.
main, Apple Silicon,cargo bench --bench hash_join_semi_anti -- build1m):Peak RSS of the
left_anti_build1m_h10bench binary: ~369 MB onmainvs ~160 MB on this branch (/usr/bin/time -l, 1M build rows, 900K unmatched).Are these changes tested?
join_emits_final_build_rows_in_batch_size_chunkstest (Left/Full/LeftAnti/LeftSemi/LeftMark × batch sizes 1/7/8192 × perfect-hash-join on/off) asserts the output rows and that every output batch respectsbatch_size. Onmain14 of its 30 cases fail the batch-size assertion.join_fetch_stops_final_build_rows_mid_chunktest checks afetchthat is reached in the middle of the final rows.joins,join_limit_pushdown,join_disable_repartition_joins,subquery), and the core join fuzz tests pass.Are there any user-facing changes?
No result changes. Output batches of the affected join types are now bounded by
batch_sizeinstead of arriving as one batch holding every unmatched build row.