andygrove opened a new pull request, #2659: URL: https://github.com/apache/datafusion-ballista/pull/2659
# Which issue does this PR close? Closes #2656. Closes #2658. # Rationale for this change TPC-H q5 at SF1000 could not complete. In the stage that sort-merge joins `orders` and `lineitem`, executors were OOM-killed while the memory pool reported only ~9 GB reserved. A heap profile showed 86% of the executor's memory at peak was remote shuffle partitions that `fetch_partition_buffered` had decoded in full with `try_collect` (#2656). That memory is not charged to the pool, and the in-flight governor only limits when a fetch starts, not how large the fetched partition is, so nothing stopped the growth. Fixing that exposed a second limit (#2658): with memory bounded, the stage spills as intended, but each executor's sort spills more than DataFusion's default 100 GB per runtime, and executors had no way to raise it. `datafusion.runtime.*` settings only apply to the client's local runtime. # What changes are included in this PR? **Stream remote shuffle fetches** (`shuffle_reader.rs`) - `fetch_partition_buffered` is replaced by `open_remote_partition`, which opens the fetch and receives the first batch, and `remote_body_stream`, which yields that batch and then pulls the rest of the body as the consumer reads it. - Opening the fetch is still wrapped in `with_retry`: nothing has been yielded downstream yet, so a failure there can be retried as before. - A failure after the first batch is reported as `FetchFailed` (carried through `DataFusionError::External`, which `FailedTask` already unwraps), so the scheduler re-runs the task, matching Spark's fetch-failure handling, rather than the reader retrying and yielding duplicate batches. - The governor permits are still held by `GovernedStream` for the life of the stream, so they now cover an open stream rather than a buffered `Vec`. - `decoded_bytes` is counted as batches stream; `fetch_time` now covers opening the fetch and the first batch. - tonic's channel `timeout` bounds only the wait for response headers, not the streaming body, so long-lived streams are not cut off. **Configurable spill limit** (`executor_process.rs`, `config.rs`) - New executor flag `--max-temp-directory-size` (same size syntax as `--memory-pool-size`). It wraps the base runtime producer, so it also applies to a custom `override_runtime_producer` such as the standalone binary's S3-enabled runtime. Unset keeps DataFusion's default. - The limit lives on each runtime's `DiskManager`, so it applies per session runtime (shared by that session's tasks); the flag help and docs say so. **Docs**: the tuning guide explains the spill limit, the new flag, and that the client-side DataFusion setting does not reach executors. **Tests** - `remote_body_stream_pulls_batches_on_demand`: opening a fetch reads only the first batch, each later batch is read only when the consumer asks for it, and `decoded_bytes` sums correctly. - `remote_body_stream_reports_mid_body_error_as_fetch_failed`: an error after the first batch surfaces as `FetchFailed` with the partition's executor, stage and partition ids. - `applies_spill_limit_to_custom_runtime_producer`. **Validation**: TPC-H q5 at SF1000 on 2 executors (8 vCPU, 56 GiB limit, `--memory-pool-size=48GB`, `--max-temp-directory-size=1TB`): | | Before | This PR | |---|---|---| | Peak executor memory | OOM-killed at ~55 GB (pool reserved ~9 GB) | 14.7 GB | | Executor restarts | both, repeatedly | none | | Result | job never completes | completes in 361 s, 5 rows | # Are there any user-facing changes? - New executor CLI flag `--max-temp-directory-size`. - `ExecutorProcessConfig` has a new public field, `max_temp_directory_size: Option<u64>` (defaults to `None`). Code that builds the struct with a full literal instead of `..Default::default()` needs to add it. - A shuffle fetch that fails partway through a partition now fails the task (and the scheduler re-runs it) instead of being retried in place by the reader. -- 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]
