mbutrovich commented on code in PR #26277:
URL: https://github.com/apache/datafusion/pull/26277#discussion_r4316615665
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -722,18 +791,20 @@ impl MaterializingSortMergeJoinStream {
/// paused. Returns true when the group scan is complete and there is
/// room for more pairs.
fn pair_streamed_row_with_group(&mut self) -> bool {
- while !self.buffered_data.scanning_finished()
- && self.num_unfrozen_pairs() < self.batch_size
- {
- let scanning_idx = self.buffered_data.scanning_idx();
- self.streamed_batch.append_output_pair(
- Some(self.buffered_data.scanning_batch_idx),
- Some(scanning_idx),
+ let mut room = self.batch_size - self.num_unfrozen_pairs();
Review Comment:
On `main`, the loop condition `num_unfrozen_pairs() < self.batch_size` means
a call with a full batch appends nothing and returns false, so the caller
freezes and calls again. Here that state underflows the subtraction. Debug
builds panic. In release, `room` wraps to near `usize::MAX`, and the loop
appends the rest of the group past `batch_size`. As I read the callers, the
`Ordering::Less` arm and the `while !self.pair_streamed_row_with_group()` loop
keep the count below `batch_size`, so this only matters if a later change
breaks that. Should we keep `main`'s behavior here with `saturating_sub`?
```suggestion
let mut room =
self.batch_size.saturating_sub(self.num_unfrozen_pairs());
```
--
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]
