sunchao commented on code in PR #24820:
URL: https://github.com/apache/datafusion/pull/24820#discussion_r3970799197
##########
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:
**[P2] Bound spill coalescing by memory**
After `reservation.free()`, this loop keeps accumulating input until the
coalescer reaches its row target or EOF. For wide strings arriving in small
batches, this retains substantial unreserved build data and creates oversized
spill batches that replay must accept over budget.
I reproduced this on base `5e168c9d0` and head `a5bbd22a0` using a lazy
source that emits 512 one-row `Utf8` batches containing 64 KiB strings, a 256
KiB memory pool, the default 8192-row target, and an empty right input:
- Base: **0.51 MiB** peak live allocations.
- Head: **64.18 MiB** peak live allocations, with **32.17 MiB** still live
just before the left source returns EOF while the pool reports **0 bytes
reserved**.
These are tracked live Rust allocations relative to the pre-execution
baseline, not RSS. The source retains no previous batches; both runs spill once
and return zero rows. Setting the head's batch target to 1 restores
approximately base memory usage, confirming that the amplification follows the
new coalescing.
Please flush partial chunks according to a byte budget, or preserve the
input-batch spill granularity, so spilling does not accumulate much more memory
than the configured limit. Otherwise this workload can exhaust process memory
despite spilling.
--
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]
