ranflarion commented on code in PR #24820:
URL: https://github.com/apache/datafusion/pull/24820#discussion_r3986296787
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -1187,45 +1245,50 @@ fn is_spillable_oom(
)
}
-/// Write the already-buffered left batches plus the remainder of the same
stream to one spill file.
+/// Write the already-completed chunks plus the remainder of the same stream
to one spill file.
+/// The remainder keeps flowing through the same coalescer, so the file holds
uniformly sized
+/// chunks and the memory-limited replay reads them back at that granularity.
/// Returns `None` when the left side carried no rows at all, which needs no
spill file.
async fn spill_left_input(
spill_manager: SpillManager,
schema: SchemaRef,
- buffered: Vec<RecordBatch>,
- pending: Option<RecordBatch>,
+ chunks: Vec<RecordBatch>,
+ mut coalescer: BatchCoalescer,
mut stream: SendableRecordBatchStream,
metrics: BuildProbeJoinMetrics,
reservation: &MemoryReservation,
) -> Result<Option<LeftSpillData>> {
let mut spill_file =
spill_manager.create_in_progress_file("NestedLoopJoin left spill")?;
- for batch in buffered {
+ for batch in chunks {
if batch.num_rows() > 0 {
spill_file.append_batch(&batch)?;
}
}
- // The in-memory batches are spilled and dropped, so their reservation
goes back to the pool
- // before the rest of the stream is drained.
+ // The in-memory chunks are spilled and dropped, so their reservation goes
back to the pool
+ // before the rest of the stream is drained; only the coalescer's one
in-progress chunk
+ // stays resident past this point.
reservation.free();
-
- for batch in pending.into_iter() {
- if batch.num_rows() > 0 {
- metrics.build_input_batches.add(1);
- metrics.build_input_rows.add(batch.num_rows());
- spill_file.append_batch(&batch)?;
- }
+ while let Some(chunk) = coalescer.next_completed_batch() {
+ spill_file.append_batch(&chunk)?;
}
while let Some(batch) = stream.next().await {
let batch = batch?;
if batch.num_rows() > 0 {
metrics.build_input_batches.add(1);
metrics.build_input_rows.add(batch.num_rows());
- spill_file.append_batch(&batch)?;
+ coalescer.push_batch(batch)?;
+ while let Some(chunk) = coalescer.next_completed_batch() {
+ spill_file.append_batch(&chunk)?;
+ }
Review Comment:
Yes, reproduced. Fixed in cc8badfff by finishing the coalescer's partial
chunk when the load spills (its inputs were all reserved, so that copy is
inside the budget the pool already granted) and writing the rest of the stream
to the spill file batch by batch as it arrives, the way the base did.
--
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]
