andygrove opened a new issue, #1952: URL: https://github.com/apache/datafusion-ballista/issues/1952
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** Follow-up to #1951, which adds a reduce-side in-flight governor to the shuffle reader. To make a mid-body transport failure retriable, that PR buffers each remote partition body fully into memory (`Vec<RecordBatch>`) before yielding it downstream, turning an open-ended stream into a discrete, idempotent, refetchable unit. The governor bounds concurrent in-flight bytes to `ballista.shuffle.reader.max_bytes_in_flight` (default 48 MiB), which caps memory *approximately* — but there are two gaps the current design intentionally leaves open: 1. **The byte budget is charged in compressed wire bytes, not decoded bytes.** The governor charges each block its `PartitionStats.num_bytes`, which is the LZ4-compressed Arrow IPC file size (what `write_stream_to_disk` records). But the buffered `Vec<RecordBatch>` is *decoded, uncompressed* Arrow data — commonly 2–5× larger. So peak concurrent buffered RAM exceeds the configured byte budget by the compression/expansion ratio. An operator setting a 48 MiB budget can see multiples of that in resident memory. 2. **A single partition larger than the whole budget is buffered in full.** The governor's forward-progress rule lets an oversized block acquire the entire byte budget and run alone, but it is still buffered entirely in memory. One very large partition can therefore exceed the intended cap regardless of the budget. Spark bounds this with `spark.maxRemoteBlockSizeFetchToMem`: blocks above the threshold stream to disk instead of being held in memory, so the reduce side never pins an unbounded amount of RAM. **Describe the solution you'd like** Add a disk-spill path to the shuffle reader's buffered fetch, analogous to `spark.maxRemoteBlockSizeFetchToMem`: - When a partition body's size (from `PartitionStats.num_bytes`, or observed while streaming) exceeds a configurable threshold, stream it to a temporary on-disk file instead of buffering `Vec<RecordBatch>` in memory, then serve batches back from disk. - Keep the fetch idempotent and retriable (re-fetch on transport error) as it is today; a spilled block is just backed by a temp file instead of a `Vec`. - Add a config key (e.g. `ballista.shuffle.reader.max_block_size_fetch_to_mem_bytes`) to set the memory-vs-disk threshold, with a sensible default. - Once decoded memory is bounded this way, the governor's byte budget can be documented as a true memory ceiling rather than a wire-bytes proxy. **Describe alternatives you've considered** - **Charge the governor an estimated decoded size** (apply a compression-ratio multiplier to `num_bytes`). Cheaper, but it only makes the *average* bound tighter — it does not prevent a single oversized partition from being buffered in full, and the multiplier is workload-dependent. - **Stream bodies through instead of buffering** (no full-partition buffer). This removes the memory concern but reintroduces the problem #1951 solved: a mid-body transport failure can no longer be retried transparently because batches have already been emitted downstream. Disk-spill preserves the retriable-block property while bounding memory. **Additional context** Introduced by / discussed in #1951 (see its "Deferred" section). The in-memory buffering there is sufficient for typical (coalesced) TPC-H partition sizes and was validated end-to-end, but disk-spill is the correct fix for large partitions and for making `max_bytes_in_flight` a hard memory bound. -- 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]
