venkata91 opened a new pull request, #58437: URL: https://github.com/apache/spark/pull/58437
### What changes were proposed in this pull request? On executor loss, `DAGScheduler` unregisters all map outputs on that executor and forces a map-stage recompute. For shuffles whose output is reliably stored off-executor (e.g. a remote shuffle service such as Celeborn), the data survives the executor, so the recompute is wasteful. Reliability is per-shuffle, not app-global. `ShuffleDriverComponents.supportsReliableStorage()` is a single application-wide flag, but under a mixed/fallback setup one shuffle can live on the remote service while another falls back to local disk on the same executor. This PR makes reliability per-shuffle: - `ShuffleHandle.isReliablyStored` (default `false`), overridable by a `ShuffleManager` that routes a shuffle to reliable storage. - `MapOutputTracker` stores the bit per shuffle (`ShuffleStatus`), exposes `isReliablyStored(shuffleId)`, and `removeOutputsOnExecutor` / `removeOutputsOnHost` gain a `skipReliablyStored` overload that leaves reliably-stored shuffles registered. - `DAGScheduler` passes `skipReliablyStored = true` on executor loss (`handleExecutorLost`) and worker loss (`handleWorkerRemoved`); the FetchFailed path keeps `false` so a genuine fetch failure still unregisters everything. - `TaskSetManager.executorLost` honors per-shuffle reliability in its re-run gate. ### Why are the changes needed? With a remote shuffle service in a mixed configuration, losing an executor needlessly recomputes map stages whose output is safely stored on the service. The existing app-global flag cannot express "this shuffle is reliable but that one is not," so it either recomputes reliable shuffles or skips recompute for local-disk fallback shuffles that were genuinely lost. ### Does this PR introduce _any_ user-facing change? No. `ShuffleHandle.isReliablyStored` defaults to `false`, so behavior is unchanged unless a `ShuffleManager` opts in. ### How was this patch tested? - New `MapOutputTrackerSuite` test covering a mixed reliable / local-disk scenario: executor loss preserves the reliably-stored shuffle and drops the local-disk one, while a fetch failure removes both. - Extended `DAGSchedulerSuite` verifications for the executor-loss vs fetch-failure `skipReliablyStored` wiring. - `DAGSchedulerSuite`, `MapOutputTrackerSuite`, and `TaskSetManagerSuite` pass; scalastyle clean. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 4.8) -- 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]
