andygrove opened a new pull request, #2170:
URL: https://github.com/apache/datafusion-ballista/pull/2170
# Which issue does this PR close?
<!-- No existing issue; happy to file one if preferred for changelog
tracking. -->
No linked issue.
# Rationale for this change
When a job completes, the scheduler fires `JobFinished` as soon as the final
stage's tasks finish on the executors — independent of whether the client has
fetched the results. It then schedules the final-stage output for deletion only
after `finished_job_data_clean_up_interval_seconds` (default 300s). As a
result, a job's result files linger on executor disk for up to ~5 minutes after
the client has actually finished consuming them, because nothing tells the
scheduler the client is done.
The scheduler already exposes a `CleanJobData(job_id)` RPC that reclaims a
job's on-disk data on executors, but nothing calls it. This PR wires the client
to call it as soon as it finishes (or abandons) consuming results, so the data
is reclaimed promptly. The scheduler's timed cleanup remains as a safety net,
and scheduler-side job *state* is deliberately left on its own timer so jobs
still appear in history/status.
# What changes are included in this PR?
- `DistributedQueryExec` now wraps its result stream in a `GuardedStream`
whose drop-guard fires the existing `CleanJobData` RPC (fire-and-forget) when
the client finishes consuming results or drops the stream early (e.g. `LIMIT`,
or a mid-stream error). This covers both the pull and push job-tracking paths.
- The RPC sends an empty `remove_stage_ids`, i.e. "remove the whole job
dir" — reclaiming the final-stage output. Scheduler job state is untouched.
- The guard captures a `tokio::runtime::Handle` in async context and
spawns the RPC on drop, so dropping never blocks; RPC errors are best-effort
(logged at `warn!`). `GuardedStream` declares `inner` before the guard, so the
result-fetch substreams tear down before cleanup fires — no file is deleted out
from under an in-flight fetch.
- New config flag `ballista.job.client_side_cleanup` (Boolean, default
`true`) gates the behavior. When disabled, the stream is returned unchanged and
only the scheduler's timer reclaims data.
- No scheduler or executor source changes — the RPC and its executor-side
handling already existed.
- Adds a standalone integration test verifying the job dir is removed
promptly after the client consumes results (and is retained when the flag is
off).
# Are there any user-facing changes?
Yes:
- New config option `ballista.job.client_side_cleanup` (default `true`),
documented in the generated config reference.
- Behavior change: by default, a job's on-disk result data on executors is
now reclaimed shortly after the client finishes fetching results, rather than
after the scheduler's ~300s timer. This is benign (faster reclamation; the
timer remains a safety net), but if you have an out-of-band process that reads
executor result files *after* the client has consumed them, set
`ballista.job.client_side_cleanup=false` to preserve the old timing.
No breaking public API changes (the additions are a new config key and a
getter).
--
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]
