andygrove opened a new pull request, #1982:
URL: https://github.com/apache/datafusion-ballista/pull/1982
# Which issue does this PR close?
Closes #1981.
Related to #1980 (denser shuffle compression codec is a complementary,
separate lever for the same disk pressure).
# Rationale for this change
When a query job succeeds, Ballista keeps the entire job directory on the
executors — intermediate inter-stage shuffle data *and* the final-stage output
— and only deletes it after a delay of
`finished_job_data_clean_up_interval_seconds` (default 300s). That delay exists
for a good reason: on success the scheduler returns the final stage's output
partition locations to the client, and in the default (pull) model the client
fetches those results from the executor's job directory *after* the job is
marked successful, so the final output must survive for the client to read it.
The cost of that delay is disk. The intermediate shuffle data — which is the
bulk of a job's on-disk footprint — is dead the moment the job succeeds (every
stage's output has already been consumed by its downstream stage), yet it
lingers for the full interval. In back-to-back runs (for example the TPC-H SF10
CI job, which runs all 22 queries with AQE off and on), intermediate shuffle
from many just-completed queries accumulates at once and intermittently
exhausts the runner's disk (`No space left on device`, os error 28).
This change reclaims the intermediate shuffle data immediately on job
success while retaining the final-stage output for the existing delayed
cleanup, so the client can still fetch results.
# What changes are included in this PR?
- **Protobuf:** a new `repeated uint32 remove_stage_ids` field on
`RemoveJobDataParams` (push path) and `CleanJobDataParams` (pull/poll path).
Empty means "remove the whole job directory" — identical to today's behavior,
so the change is backward compatible on the wire.
- **Executor:** the whole-job `remove_job_dir` becomes stage-aware
`remove_job_data(work_dir, job_id, remove_stage_ids)` — empty removes the whole
job dir; non-empty removes only the given `work_dir/{job_id}/{stage_id}`
subdirectories (missing ones skipped), keeping the existing `is_subdirectory`
safety guard per path.
- **Scheduler:** the cleanup plumbing carries `remove_stage_ids` through
both the push (direct RPC) and pull (`pending_cleanup_jobs` → poll response)
paths. On job success the scheduler computes the intermediate stage set (`all
stages − final stages`, where a final stage has empty `output_links`) and fires
an immediate reclaim via a new `clean_up_intermediate_job_data`, in addition to
the existing delayed whole-job cleanup that now just removes the retained
final-stage output.
Safety invariant: the immediate reclaim only ever deletes stages in `all −
final`; a final-stage output directory can never be reached by it, and any
unresolvable/failed job path degrades to an empty set (no reclaim, existing
delayed cleanup only). Failed-job cleanup is unchanged.
# Are there any user-facing changes?
Behavioral, not API. By default, intermediate shuffle files for a completed
query are removed from executors immediately on job success rather than after
`finished_job_data_clean_up_interval_seconds`. Final query results are
unaffected — they are retained for the existing delayed-cleanup window exactly
as before. No public API or configuration changes; no new configuration is
required.
--
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]
