Describe the bug
ClassicPWMJStream (Inner/Left/Right/Full range joins) buffers its output in Arrow's BatchCoalescer. One push_batch can complete several target-sized batches and retain a partial remainder. The state machine, however, treats "the producer is finished" as if it also meant "all output has been emitted", and transitions (or falls back into a producer) while batches are still queued. Three consequences, worst first:
1. The Left/Full unmatched pass restarts forever (reachable from SQL). The drain block in process_unmatched_buffered_batch (classic_join.rs#L326-L346) has no transition for the empty-queue case:
if !self.batch_process_state.continue_process {
if let Some(batch) = /* next_completed_batch() */ {
return Ok(StatefulStreamResult::Ready(Some(batch)));
}
/* finish_buffered_batch() */
if let Some(batch) = /* next_completed_batch() */ {
self.state = PiecewiseMergeJoinStreamState::Completed;
return Ok(StatefulStreamResult::Ready(Some(batch)));
}
} // <- both pops empty: no transition, falls through…
let buffered_data = /* … */ // …into the unmatched-row producer, which runs againWhen the number of unmatched buffered rows is an exact multiple of batch_size, the flush leaves no partial remainder, so after the queue drains both pops return None, neither return fires, and control falls through into the producer. The visited bitmap still reports the same rows unmatched, so the same batches are pushed again — the stream emits duplicates forever and never terminates. Whether you hit it depends only on data cardinality vs. configured batch size.
2. Queued unmatched Right/Full rows can be silently dropped. At scan end, resolve_classic_join flushes all unmatched streamed rows in one push (classic_join.rs#L576-L583), which can queue ≥ 2 completed batches. process_stream_batch then pops one, transitions to FetchStreamBatch, and returns it (classic_join.rs#L291-L303). If the streamed input then ends, a Right/Inner join short-circuits straight to Completed (#L320-L323) — the remaining queued batches become unreachable and rows are missing from the result. The non-final-partition transition (#L230) drops leftovers the same way. Reachability caveat: this needs a streamed batch carrying ≥ 2×batch_size unmatched rows at the terminal transition; in the SQL plans I tried, stream batches arrive chunked to the session batch size, so I could only demonstrate it driving the operator directly — but it is the same missing invariant as bug 1.
3. A zero-row placeholder escapes as stream output. resolve_classic_join returns RecordBatch::new_empty purely as control flow (#L586-L587), and the scan-end path hands it to the consumer (#L306-L309) — an empty join yields one zero-row batch instead of no batches. Harmless to results, but it is StatefulStreamResult::Continue's job, expressed as data.
To Reproduce
Bug 1, via SQL (4 unmatched buffered rows = 2 × batch_size; append LIMIT 20 to see the duplicated rows instead of a non-terminating query):
set datafusion.optimizer.enable_piecewise_merge_join = true;
set datafusion.execution.batch_size = 2;
create table l(v int) as values (10), (20), (30), (40), (1);
create table r(v int) as values (5);
select * from l left join r on l.v <= r.v; -- never terminates
Bug 2, at the operator level: build a PiecewiseMergeJoinExec (Right join, l.v >= r.v) over a buffered side [1, 2] and a streamed side [3, 5] with batch_size = 1. Neither streamed row matches, the flush queues two completed batches, and only the first null-extended row is emitted — collect returns 1 row where 2 are correct.
Expected behavior
A phase transitions only after its coalescer has been finalized and confirmed empty (finish_buffered_batch() + next_completed_batch() returning None): every queued batch is emitted exactly once, the unmatched pass runs once and the stream terminates, and an empty join yields no batches. One drain invariant at the two phase boundaries closes all three defects; matching, null handling, and the visited bitmap need no changes.
Additional context
Describe the bug
ClassicPWMJStream(Inner/Left/Right/Full range joins) buffers its output in Arrow'sBatchCoalescer. Onepush_batchcan complete several target-sized batches and retain a partial remainder. The state machine, however, treats "the producer is finished" as if it also meant "all output has been emitted", and transitions (or falls back into a producer) while batches are still queued. Three consequences, worst first:1. The Left/Full unmatched pass restarts forever (reachable from SQL). The drain block in
process_unmatched_buffered_batch(classic_join.rs#L326-L346) has no transition for the empty-queue case:When the number of unmatched buffered rows is an exact multiple of
batch_size, the flush leaves no partial remainder, so after the queue drains both pops returnNone, neitherreturnfires, and control falls through into the producer. The visited bitmap still reports the same rows unmatched, so the same batches are pushed again — the stream emits duplicates forever and never terminates. Whether you hit it depends only on data cardinality vs. configured batch size.2. Queued unmatched Right/Full rows can be silently dropped. At scan end,
resolve_classic_joinflushes all unmatched streamed rows in one push (classic_join.rs#L576-L583), which can queue ≥ 2 completed batches.process_stream_batchthen pops one, transitions toFetchStreamBatch, and returns it (classic_join.rs#L291-L303). If the streamed input then ends, a Right/Inner join short-circuits straight toCompleted(#L320-L323) — the remaining queued batches become unreachable and rows are missing from the result. The non-final-partition transition (#L230) drops leftovers the same way. Reachability caveat: this needs a streamed batch carrying ≥ 2×batch_sizeunmatched rows at the terminal transition; in the SQL plans I tried, stream batches arrive chunked to the session batch size, so I could only demonstrate it driving the operator directly — but it is the same missing invariant as bug 1.3. A zero-row placeholder escapes as stream output.
resolve_classic_joinreturnsRecordBatch::new_emptypurely as control flow (#L586-L587), and the scan-end path hands it to the consumer (#L306-L309) — an empty join yields one zero-row batch instead of no batches. Harmless to results, but it isStatefulStreamResult::Continue's job, expressed as data.To Reproduce
Bug 1, via SQL (4 unmatched buffered rows = 2 × batch_size; append
LIMIT 20to see the duplicated rows instead of a non-terminating query):Bug 2, at the operator level: build a
PiecewiseMergeJoinExec(Right join,l.v >= r.v) over a buffered side[1, 2]and a streamed side[3, 5]withbatch_size = 1. Neither streamed row matches, the flush queues two completed batches, and only the first null-extended row is emitted —collectreturns 1 row where 2 are correct.Expected behavior
A phase transitions only after its coalescer has been finalized and confirmed empty (
finish_buffered_batch()+next_completed_batch()returningNone): every queued batch is emitted exactly once, the unmatched pass runs once and the stream terminates, and an empty join yields no batches. One drain invariant at the two phase boundaries closes all three defects; matching, null handling, and the visited bitmap need no changes.Additional context
PiecewiseMergeJoinis experimental andenable_piecewise_merge_joindefaults tofalse, so the blast radius is opt-in today — worth fixing before default-enablement is considered.PiecewiseMergeJoinwork in Datafusion #17427.