korowa commented on code in PR #12082: URL: https://github.com/apache/datafusion/pull/12082#discussion_r1730388571
########## 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: I've finally got your idea (and the fact that the problem is not related to fetching probe side, but to processing already joined buffered side data). Probably for anti join following will be helpful 1) `get_filtered_join_mask` -- maybe it should only update `matched_indices` (in case filters are evaluated as `true` as least once), and data emission logic should be in some other place (currently there is a problem with streamed records without any filter matches will be duplicated for each joined buffered chunk, as "negative" filter results are not tracked across joined batches) 2) filtered anti join (at least filtered one) should return only the records for which buffered-side scanning is completed (as `freeze_streamed` may be called in the middle of buffered-data scanning, due to output batch size), and there were no `true` filters for them (from p.1) -- so, maybe we should split filter evaluation from emitting data in `freeze_streamed` (since the filters should be checked for all matched indices, but in the same time, the last (current) streamed index can be filitered out of output because it has further buffered batches to be joined with)? -- 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: github-unsubscr...@datafusion.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org For additional commands, e-mail: github-h...@datafusion.apache.org