WIP: feat: treat ResourcesExhausted as a retriable task failure

[andygrove]

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.