Weijun-H commented on code in PR #25910:
URL: https://github.com/apache/datafusion/pull/25910#discussion_r4243787355
##########
datafusion/physical-plan/src/spill/in_progress_spill_file.rs:
##########
@@ -245,11 +273,453 @@ impl InProgressSpillFile {
}
}
+/// Compacts `batch` for spilling and, when `max_batch_bytes` is set, splits
it into row
+/// ranges of at most that size by recursive halving. Returns each piece with
the size that
+/// a reader measures when it decodes the piece.
+///
+/// Without a bound, the batch is written whole after [`gc_view_arrays`], as
before.
+///
+/// With a bound, the split is decided on [`referenced_size`], which models
what the IPC
+/// writer encodes for a row range, and each range is then compacted by
[`compact_array`]
+/// so that it matches that model. The sliced size of an uncompacted slice
counts every
+/// buffer the slice keeps alive, so a slice of a large view array would
report the whole
+/// parent, and no split would ever seem to help.
+///
+/// When a batch is split, each piece also drops the dictionary values that
none of its keys
+/// use, so a large dictionary is divided between the pieces instead of
written whole with
+/// each one. The IPC stream then sends a replacement dictionary before each
piece.
+fn split_for_spill(
+ batch: &RecordBatch,
+ max_batch_bytes: Option<usize>,
+) -> Result<Vec<(RecordBatch, usize)>> {
+ let Some(max_batch_bytes) = max_batch_bytes else {
+ let gc_batch = gc_view_arrays(batch)?;
+ let size = gc_batch.get_sliced_size()?;
+ return Ok(vec![(gc_batch, size)]);
+ };
+ let size = referenced_batch_size(batch)?;
+ let mut ranges = Vec::new();
+ split_ranges(batch, size, max_batch_bytes, &mut ranges)?;
+ ranges.iter().map(compact_piece).collect()
Review Comment:
Preserve shared ListView children when splitting
For 1,024 ListView rows sharing one 64 KiB child, a 1 MiB limit expands a 74
KB input into 68 MB of simultaneously live pieces before the first write,
outside the memory pool; the spill file also grows from 74 KB to 67 MB.
Preserve overlapping child ranges and compact/write pieces incrementally;
incremental writing alone leaves the disk and merge-reservation amplification.
--
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]
