ranflarion opened a new pull request, #25206: URL: https://github.com/apache/datafusion/pull/25206
## Which issue does this PR close? - Closes #25205. ## Rationale for this change A sort-merge join partition whose streamed side is empty still fetches, sorts and scans its entire buffered side, because the stream loads a key group from the buffered input before comparing anything and merge-scans until both inputs are exhausted. Only a Full join can emit buffered rows without a streamed match, so for every other join type that work produces nothing. On a hash-partitioned join of a large table against a sparse one, most partitions look like this: in the production job that motivated this, 1,371 of 2,000 partitions had an empty streamed side, and the join consumed 6,231,306,982 buffered rows to emit 6,144, taking 456 s for the stage where Spark, whose `SortMergeJoinExec` exits as soon as the streamed side is empty, took 194 s. The symmetric case, an Inner join whose buffered side is exhausted with no key group left, drains and drops the remaining streamed rows for the same reason. ## What changes are included in this PR? The materializing stream (Inner/Left/Right/Full) gets a `finished()` predicate that replaces "both inputs exhausted" as the loop condition: an exhausted streamed side finishes every join but Full, and an exhausted buffered side finishes an Inner join. The first buffered key group is only loaded when the join is not already finished, so an empty streamed partition never polls the buffered input, and the `SortExec` beneath it never runs. `on_children_exhausted` now releases both inputs and every remaining buffered key group right after freezing the pending pairs, so the memory the join reserved is back in the pool before its final batches are emitted rather than when the stream is dropped; without that, a parent operator processing the final batch can fail with `ResourcesExhausted` against a reservation the finished child still holds. The bitwise stream (Semi/Anti/Mark) already returns on an empty outer side; it now also stops draining the outer side for Semi joins once the inner si de is exhausted, since the remaining rows cannot be emitted. ## Are these changes tested? New tests in `joins/sort_merge_join/tests.rs`: - `join_empty_streamed_side_never_polls_buffered_side` drives the buffered side with `PanicExec`, so it passes only if that input is never polled, for Inner/Left/LeftSemi/LeftAnti/LeftMark and the mirrored Right/RightSemi/RightAnti. It fails on `main` with the `PanicExec` panic. - `join_full_empty_left_still_emits_buffered_rows` pins that Full still emits the buffered rows. - `join_inner_and_semi_finish_once_buffered_side_is_exhausted` covers the Inner and Semi exit and that Anti still emits the unmatched streamed rows. - `join_left_releases_buffered_reservation_before_final_batch` asserts the pool holds no reservation when the final batch is returned, and `join_chain_reuses_memory_released_by_completed_child` runs a two-join plan under a 2 MB pool with spilling disabled, which fails with `ResourcesExhausted` if the child keeps its reservation. All existing `joins::sort_merge_join` tests pass. ## Are there any user-facing changes? No API or result changes. Sort-merge joins read less input in the cases above, and their `input_rows` metric drops accordingly. -- 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]
