buraksenn opened a new issue, #24297: URL: https://github.com/apache/datafusion/issues/24297
### 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`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#L326-L346)) has no transition for the empty-queue case: ```rust 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 again ``` 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 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`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/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`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#L291-L303)). If the streamed input then ends, a Right/Inner join short-circuits straight to `Completed` ([`#L320-L323`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#L320-L323)) — the remaining queued batches become unreachable and rows are missing from the result. The non-final-partition transition ([`#L230`](https://githu b.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#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`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#L586-L587)), and the scan-end path hands it to the consumer ([`#L306-L309`](https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs#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): ```sql 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 - `PiecewiseMergeJoin` is experimental and `enable_piecewise_merge_join` defaults to `false`, so the blast radius is opt-in today — worth fixing before default-enablement is considered. - Part of the classic-join hardening for #17427. -- 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]
