apache / apache/datafusion

SortMergeJoinExec reads and sorts the whole buffered side of a partition whose streamed side is empty

Open
#25,205 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### 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 = 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.

Contributor guide

Open the contributing guide

Research direction

Start in joins/sort_merge_join/tests.rs with the empty TestMemoryExec and PanicExec reproduction, then inspect SortMergeJoinExec's stream polling and merge-scan behavior. Done means non-Full joins stop without polling an exhausted streamed side, Inner joins stop when the buffered side is exhausted, and buffered memory is released before the final output batch.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
data-engineering, performance
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.