feat: reclaim intermediate shuffle data on job success#1982
Open
andygrove wants to merge 6 commits into
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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?
repeated uint32 remove_stage_idsfield onRemoveJobDataParams(push path) andCleanJobDataParams(pull/poll path). Empty means "remove the whole job directory" — identical to today's behavior, so the change is backward compatible on the wire.remove_job_dirbecomes stage-awareremove_job_data(work_dir, job_id, remove_stage_ids)— empty removes the whole job dir; non-empty removes only the givenwork_dir/{job_id}/{stage_id}subdirectories (missing ones skipped), keeping the existingis_subdirectorysafety guard per path.remove_stage_idsthrough 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 emptyoutput_links) and fires an immediate reclaim via a newclean_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.