stuhood opened a new issue, #27109: URL: https://github.com/apache/datafusion/issues/27109
### Is your feature request related to a problem or challenge? `HashJoinStream` unconditionally coalesces matched output rows across probe batches up to `datafusion.execution.batch_size` (`8192`) via `LimitedBatchCoalescer`: https://github.com/apache/datafusion/blob/a7c7c08365a37f0b46bce81b1fd45515f3f18045/datafusion/physical-plan/src/joins/hash_join/stream.rs#L586-L587 When the probe side is expensive to pull from and already ordered for a downstream limit, coalescing across probe batches forces unnecessary work on the probe input before downstream operators can see already-matched rows. For example, when an incremental top-`K` probe scan feeds a chain of `HashJoinExec`s under `ORDER BY ... LIMIT K`, `LimitPushdown` can set `fetch = Some(K)` only on the outermost join; inner joins have `fetch = None`, so their `LimitedBatchCoalescer` buffers matches and repeatedly pulls from the probe scan until `8192` matches accumulate or the scan is exhausted. Relatedly, `datafusion.execution.coalesce_batches` became a no-op after `FilterExec` (#18630), `HashJoinExec` (#18972), and `RepartitionExec` (#19449) internalized `LimitedBatchCoalescer` and #19622 removed the `CoalesceBatches` rule: https://github.com/apache/datafusion/blob/a7c7c08365a37f0b46bce81b1fd45515f3f18045/datafusion/common/src/config.rs#L928-L932 ### Describe the solution you'd like Support disabling coalescing on `LimitedBatchCoalescer` (via `biggest_coalesce_batch_size = Some(0)` so non-empty batches bypass buffering while empty batches are skipped and `fetch` is still enforced), exposed via: 1. A per-operator `HashJoinExec::with_coalesce_batches(bool)` setting (so physical optimizer rules can disable coalescing selectively on demand-driven joins). 2. Wiring `datafusion.execution.coalesce_batches` into operators that use `LimitedBatchCoalescer`. ### Describe alternatives you've considered * Lowering `datafusion.execution.batch_size` globally degrades bulk operators and build-side scans in the same query. * A per-operator `batch_size` override on `HashJoinExec` still buffers batches of `<= batch_size / 2` rows and shrinks the per-step match cap in `lookup_join_hashmap`. ### Additional context None. -- 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]
