ranflarion opened a new issue, #25205:
URL: https://github.com/apache/datafusion/issues/25205
### Describe the bug
`SortMergeJoinExec` reads and sorts the whole buffered side of a partition
even when the streamed side of that partition is empty. The stream loads the
first streamed batch and the first buffered key group unconditionally, then
merge-scans until *both* inputs are exhausted, so once the streamed side runs
out it keeps pulling buffered key groups (the `Greater` arm) until the buffered
input ends, although only a Full join can emit anything from them. The same
happens the other way round for Inner joins: with the buffered side exhausted
and no key group left, the remaining streamed rows are pulled and dropped one
by one.
In a hash-partitioned join of a large table against a sparse one this is
most of the work. On a production Spark job we run through DataFusion (6.2B-row
history side, a few thousand rows on the other side, 2,000 partitions), 1,371
of 2,000 partitions had an empty streamed side; every one of them still fetched
its ~3M-row buffered partition, sorted it (the `SortExec` below the join runs
on first poll) and scanned it against nothing: 6,231,306,982 rows consumed for
6,144 output rows, 456 s for the stage against 194 s in Spark, whose
`SortMergeJoinExec` returns as soon as the streamed side is empty and never
polls the buffered side.
### To Reproduce
Against `main`, in `joins/sort_merge_join/tests.rs`: join an empty
`TestMemoryExec` left table with a right side built from
`PanicExec::new(schema, 1)` (panics on first poll) for `Inner`, `Left`,
`LeftSemi`, `LeftAnti` or `LeftMark`, or the mirror image (`PanicExec` on the
left, empty right) for `Right`, `RightSemi`, `RightAnti`:
```rust
let left = build_table(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
let right: Arc<dyn ExecutionPlan> = Arc::new(PanicExec::new(schema, 1));
join_collect(left, right, on, Inner).await?; // panics: PanicExec polled
```
### Expected behavior
When the streamed side of a partition is exhausted, the join should finish
without polling the buffered side any further (never, for an empty streamed
partition), for every join type except Full, which still has to emit unmatched
buffered rows. When the buffered side is exhausted with no key group left, an
Inner join should finish without draining the streamed side. Whatever the join
has buffered at that point should be released back to the memory pool before
its final output batch is emitted, not when the stream is dropped.
### Additional context
Spark's `SortMergeJoinExec` behaves this way, and the sorts and shuffle
reads under the join are lazy, so skipping the poll skips their work as well.
We run this fix in production against 54.1.0 and can offer it as a PR.
--
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]
