We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
1 parent 1e5aa10 commit e3e1454Copy full SHA for e3e1454
1 file changed
datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs
@@ -600,6 +600,16 @@ impl Stream for MaterializingSortMergeJoinStream {
600
.filter_mask
601
.len();
602
if accumulated >= self.batch_size {
603
+ // Ensure required spilled batches are restored to memory
604
+ // before processing, as this path invokes freeze_all().
605
+ let needed = self.get_required_batch_indices(
606
+ self.buffered_data.batches.len(),
607
+ );
608
+ if let Err(e) = ready!(
609
+ self.poll_spilled_batches(cx, &needed)
610
+ ) {
611
+ return Poll::Ready(Some(Err(e)));
612
+ }
613
match self.process_filtered_batches()? {
614
Poll::Ready(Some(batch)) => {
615
return Poll::Ready(Some(Ok(batch)));
0 commit comments