andygrove opened a new pull request, #2032:
URL: https://github.com/apache/datafusion-ballista/pull/2032
> **Status: WIP / experimental.** Opened as a draft for early feedback. This
is the first of two stacked PRs; the second builds an opt-in OOM guard on top
of it.
# Which issue does this PR close?
Relates to #2031. It does not close it — this PR only fixes the failure
*classification*; the memory accounting itself is the follow-up.
# Rationale for this change
A task that exhausts its memory budget currently fails the entire job on the
first occurrence. `From<BallistaError> for FailedTask` maps everything that is
not an IO error to `FailedReason::ExecutionError`, and
`ExecutionGraph::update_task_status` fails the stage outright on that reason
with no retry.
Memory exhaustion is often transient. The retried task may land on a
less-loaded executor, or run once its peers on the same executor have drained —
the same reasoning Spark uses. The retry is bounded by `--task-max-failures`
(default 4), so a task that is simply too large still fails the job, but with a
clear message rather than immediately.
While tracing this I found the classification was also *unreachable* for a
large class of stages, which is the second half of the change:
`ShuffleWriterExec`, on the `shuffle_output_partitioning == None` branch,
flattened the stream error with `map_err(|e|
DataFusionError::Execution(format!("{e:?}")))` before it reached the
classifier. That destroys the error type, so `find_root()` sees `Execution` and
the task is classified non-retriable. Per `planner.rs`, the `None`-partitioned
stages are the final stage, the **broadcast-join build side**, the
`CoalescePartitionsExec` input stage, and the `SortPreservingMergeExec` input
stage — i.e. several of the stages most likely to exhaust memory in the first
place. The hash-partitioned branch and the sort-shuffle writer already
propagated the error unchanged.
# What changes are included in this PR?
- `ballista.proto`: a new `ResourcesExhausted` message and a
`resources_exhausted` arm (field 10) on the `FailedTask.failed_reason` oneof.
- `error.rs`: a new arm in `From<BallistaError> for FailedTask` setting
`retryable: true, count_to_failures: true`. It matches through
`DataFusionError::find_root()`, because DataFusion routinely wraps errors in
`Context`, and in `Shared` when propagating one stream error to several output
partitions (`RepartitionExec`) — a bare match on the outer error would miss
most real cases. There are tests for the bare, `Context`-wrapped, and
`Shared`-wrapped forms.
- `shuffle_writer.rs`: preserve the `DataFusionError` on the unpartitioned
write path instead of formatting it into an `Execution` error.
- `handlers.rs`: a new arm in `failed_reason()`, which is an exhaustive
match. Kept exhaustive deliberately — it is display-only, and it is a useful
tripwire for the next `FailedReason` added.
No scheduler logic change is required: `update_task_status`'s existing
`Some(_)` arm already retries any reason that is not `ExecutionError` or
`FetchPartitionError` when `retryable && count_to_failures`. Verified this
holds on both the static and the AQE planner paths.
# Are there any user-facing changes?
Yes, and reviewers should weigh this explicitly: **this changes
default-build behaviour.** A `ResourcesExhausted` — which a bounded
`FairSpillPool` can raise today — is now retried up to `--task-max-failures`
times instead of failing the job immediately. For a *transient* spike that is
the point. For a *deterministic* one, the job now performs the same work up to
4 times before surfacing the identical failure. That trade-off is the main
thing I would like feedback on; if it is unwelcome, the classification could be
gated behind the follow-up PR's cargo feature instead.
The proto change adds a new oneof field; it does not alter existing fields.
--
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]
