Repository navigation
perf: emit unmatched build rows in batch_size chunks in HashJoinExec - #25028
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
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 🚀 |
…tion saw NULL RE-IMPLEMENTATION, not a faithful cherry-pick. apache#25076 (5747e87) does not apply here: this fork's base predates the null-aware branching refactor (apache#24957, `NullAwareMode`) and the chunked final emission (apache#25028, `EmitUnmatchedBuildRowsState`), and `git show 5747e87 | git apply --check` fails on `exec.rs`, `stream.rs` and the three manifests. The semantics below are the upstream ones; the shape is this fork's. The bug: the final stage read the shared "probe side saw NULL" flag and only then decremented the probe-partition counter, so a partition could read `false`, let a sibling record a NULL and finish, then become the last partition itself and emit build rows that `NOT IN` must suppress. The counter also decremented with `Relaxed`, so nothing ordered a sibling's store before the last partition's load. Changes: - New `hash_join/probe_completion.rs`, ported nearly verbatim: `ProbeCompletion` owns the probe-partition counter and the two shared flags, `report_completed()` decrements with `AcqRel` and returns a `ProbeSideSummary` only to the last partition, so the final stage cannot read the flags before its own decrement. Its sequential unit tests and the `--cfg datafusion_loom` model-checked tests come along with the workspace `loom` dependency and the `cfg(datafusion_loom)` check-cfg registration. - `JoinLeftData` holds one `ProbeCompletion` instead of `probe_threads_counter`/`probe_side_non_empty`/`probe_side_has_null`. The probe phase records through `record_probe_batch()` and keeps a best-effort `probe_side_has_null_hint()` early exit. - `process_unmatched_build_batch` decrements first and hands the resulting summary to the new `null_aware_left_anti_final_indices`, which holds the `LeftAnti` rules: a probe NULL seen by any partition suppresses every build row; otherwise a NULL build key is dropped unless the probe side was empty. - This fork has no null-aware `LeftMark`, so the mark half of the upstream PR has no counterpart here. Tests (`hash_join/exec.rs`): - `test_null_aware_anti_join_probe_null_in_other_partition`: two probe partitions, only partition 0 sees the NULL, drained in both finishing orders; no rows may be emitted. - `test_null_aware_anti_join_probe_logical_null_in_other_partition`: the same with a dictionary key whose NULL is logical only. - `test_null_aware_anti_join_empty_probe_keeps_null_build_key`: pins `NULL NOT IN (empty) = TRUE` (added here; upstream covers it via slt only). (backport of apache#25076, 5747e87) Signed-off-by: discord9 <[email protected]> (cherry picked from commit 3fb9a3084e39781b90d5dbffa01f2664f64dcc2d)
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.