comphead commented on code in PR #12082:
URL: https://github.com/apache/datafusion/pull/12082#discussion_r1733579129
##########
datafusion/physical-plan/src/joins/sort_merge_join.rs:
##########
@@ -1356,16 +1392,82 @@ impl SMJStream {
pre_mask.clone()
};
+ // Try to calculate if the buffered batch we scan is the
last one for specific stream row and join key
+ // for Batchsize == 1
self.buffered_data.scanning_finished() works well
+ // For other scenarios its an attempt to figure out there
is no more rows matching the same join key
+ let last_batch = if self.batch_size == 1 {
Review Comment:
Another option just came to my mind. We know in advance number of matched
rows. We can calc it as
```
let matched_indices = &self.buffered_data.batches.iter().map(|b| b.range.end
- b.range.start).count();
```
No matter how SMJ distributes data, we can just take number of
`buffered_indices.len()` per each iteration and substract it from
`matched_indices`. Once we have hit 0 means no more matched rows expected
--
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]
