jayzhan211 commented on code in PR #25206:
URL: https://github.com/apache/datafusion/pull/25206#discussion_r3998541321
##########
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs:
##########
@@ -844,14 +846,46 @@ impl MaterializingSortMergeJoinStream {
}
}
- /// Flush everything that remains once both inputs are exhausted.
+ /// Whether the join can produce no further output. Besides both inputs
being exhausted,
+ /// an exhausted streamed side ends every join but Full, the only one that
emits buffered
+ /// rows without a streamed match, and an exhausted buffered side (which
leaves no key
+ /// group behind) ends an Inner join, which has nothing to match the
remaining streamed
+ /// rows against. Spark's SortMergeJoinExec stops at the same points; an
empty streamed
+ /// partition therefore never polls the buffered side.
+ fn finished(&self) -> bool {
+ (self.streamed_exhausted
+ && (self.buffered_exhausted || self.join_type != JoinType::Full))
+ || (self.buffered_exhausted && self.join_type == JoinType::Inner)
+ }
+
+ /// Drops both inputs and every buffered key group once nothing more can
be joined, so the
+ /// memory they reserve is back in the pool before the final output
batches are emitted
+ /// rather than when the stream is dropped.
+ fn release_inputs(&mut self) {
+ while let Some(buffered_batch) =
self.buffered_data.batches.pop_front() {
Review Comment:
It would be nice to have a helper that wraps these three operations, so it
can be reused elsewhere without risk of missing the cleanup.
```rs
/// Dequeue the head buffered batch, returning its reservation and spill
count.
fn pop_front_buffered_batch(&mut self) -> Option<BufferedBatch> {
let batch = self.buffered_data.batches.pop_front()?;
self.free_reservation(&batch);
if matches!(batch.batch, BufferedBatchState::Spilled(_)) {
self.spilled_batch_count -= 1;
}
Some(batch)
}
```
--
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]
