jayzhan211 opened a new pull request, #25028: URL: https://github.com/apache/datafusion/pull/25028
## Which issue does this PR close? - Related to #24768 (bounded-memory hash join) and #18942 (hash join operator optimizations); no dedicated issue. ## Rationale for this change For join types that emit build-side rows after the probe side is exhausted (`Left`, `Full`, `LeftAnti`, `LeftSemi`, `LeftMark`), `HashJoinExec` computed the final indices over the *whole* build side and materialized them as **one** `RecordBatch` before handing it to the output coalescer. That batch is not bounded by `batch_size`, and it is not covered by the memory reservation. Worse, `LimitedBatchCoalescer` configures arrow's `BatchCoalescer` with `biggest_coalesce_batch_size = batch_size / 2`, which passes any larger batch through untouched. So a `LEFT ANTI` join over a 10M-row build side with few matches emitted a single ~10M-row batch downstream, regardless of `datafusion.execution.batch_size`. `NestedLoopJoinExec` already emits its unmatched build rows in `batch_size` chunks; this PR brings `HashJoinExec` in line. ## What changes are included in this PR? - `HashJoinStream` gets a new state, `EmitUnmatchedBuildRows`, entered from `ExhaustedProbeSide` by the last probe partition. It holds a `BooleanBuffer` snapshot 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_chunk` scans the snapshot from the cursor and returns at most `batch_size` final indices per call (`LeftMark` emits every row, so its chunks are plain ranges). Null-aware `LeftAnti`/`LeftMark` post-processing and `fetch` handling are unchanged and now run per chunk. - `input_batches` is still bumped once for the final phase and `input_rows` once per chunk, so metric values are identical to before. - `get_final_indices_from_bit_map` / `get_final_indices_from_shared_bitmap` in `joins/utils.rs` had no other callers and are removed. - New benchmark cases in `hash_join_semi_anti.rs` with 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`): | case | main | this PR | change | |---|---|---|---| | left_semi_build1m_h10 | 2.05 ms | 1.97 ms | -4% | | left_anti_build1m_h10 | 6.21 ms | 4.75 ms | -24% | | left_build1m_h10 | 7.37 ms | 6.79 ms | -8% | Peak RSS of the `left_anti_build1m_h10` bench binary: ~369 MB on `main` vs ~160 MB on this branch (`/usr/bin/time -l`, 1M build rows, 900K unmatched). ## Are these changes tested? - New `join_emits_final_build_rows_in_batch_size_chunks` test (Left/Full/LeftAnti/LeftSemi/LeftMark × batch sizes 1/7/8192 × perfect-hash-join on/off) asserts the output rows and that every output batch respects `batch_size`. On `main` 14 of its 30 cases fail the batch-size assertion. - New `join_fetch_stops_final_build_rows_mid_chunk` test checks a `fetch` that is reached in the middle of the final rows. - Existing hash join unit tests (493), join sqllogictests (`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_size` instead of arriving as one batch holding every unmatched build row. -- 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]
