jayzhan211 commented on code in PR #25584:
URL: https://github.com/apache/datafusion/pull/25584#discussion_r4105574077
##########
datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:
##########
@@ -912,6 +1003,122 @@ impl BitwiseSortMergeJoinStream {
}
}
+ /// Return a bounded slice length, yielding periodically even if every
+ /// child batch is immediately ready. Pausing the timer excludes scheduling
+ /// time from the join's own work.
+ async fn summary_slice_len(&mut self, remaining: usize) -> usize {
Review Comment:
Work between awaits is already bounded by batch size: every batch boundary
goes through `next_inner_batch`/`next_outer_batch`, where child streams consume
the coop budget. The existing filtered path doesn't yield internally. Slicing
at 1024 rows also repeats `make_comparator`/`to_scalar`/guard evaluation for
every slice. Suggest processing the whole in-batch group range and dropping
`rows_until_yield`, `SUMMARY_WORK_BUDGET` and `summary_slice_len`:
```rs
let batch = self
.inner_batch
.as_ref()
.unwrap()
.slice(from, group_end - from);
let state = self.existence_summary.as_mut().unwrap();
state.summary.update(&batch, &state.reservation)?;
state.inner_rows.add(group_end - from);
self.peak_mem_used.set_max(state.summary.peak_size());
```
--
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]
