kosiew commented on code in PR #25542:
URL: https://github.com/apache/datafusion/pull/25542#discussion_r4094442358
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -3130,6 +2864,103 @@ impl NestedLoopJoinStream {
}
}
+ /// Memory-limited path for handle_emit_left_unmatched.
+ ///
+ /// The global visited bitmap is complete once the last stream has reported
+ /// probe completion, and that stream is the emitter. It streams the left
+ /// spill file one more time and emits the final left rows batch by batch,
+ /// so no left chunk has to be held in memory for it. This mirrors
+ /// `EmitGlobalRightUnmatched` on the right side.
+ fn handle_emit_left_unmatched_memory_limited(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> ControlFlow<Poll<Option<Result<RecordBatch>>>> {
+ // Return any completed batches first
+ if let Some(poll) = self.maybe_flush_ready_batch() {
+ return ControlFlow::Break(poll);
+ }
+
+ let is_emitter = need_produce_result_in_final(self.join_type)
+ && self.is_unmatched_left_emitter;
+ let SpillState::Active(active) = &mut self.spill_state else {
+ unreachable!("memory-limited EmitLeftUnmatched without Active
spill state");
+ };
+
+ // On first entry, the emitter opens its pass over the left spill file
+ if is_emitter && active.left_unmatched_pass.is_none() {
+ let join_metric = self.metrics.join_metrics.join_time.clone();
+ let _join_timer = join_metric.timer();
+ match active.left_spill.open_pass() {
+ Ok(stream) => {
+ // Every chunk is done, which leaves `chunk_reservation`
free
+ // to account for the bitmap this stream now owns.
+ let visited = active.left_spill.take_visited();
+ active.chunk_reservation.grow(visited.len().div_ceil(8));
+ active.left_unmatched_pass = Some(LeftUnmatchedPass {
+ stream,
+ visited,
+ row_offset: 0,
+ });
+ }
+ Err(e) => return ControlFlow::Break(Poll::Ready(Some(Err(e)))),
Review Comment:
Could we release the visited bitmap if `open_pass()` fails here? Once the
final reporter reaches `EmitLeftUnmatched`, a synchronous reopen error from a
custom `SpillFile` can leave the bitmap reservation in `LeftSpillData` until
the plan is dropped. The default spill file opens lazily, so this does not
appear to affect the usual path. It would be fine to handle this in a
follow-up. If you do, a spill file that fails on its second `read_stream()`
call would help test it.
--
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]
