[WIP] Stack/pipelined shuffle pr6 failfast#57361
Open
jerrypeng wants to merge 59 commits into
Open
Conversation
Introduce PipelinedShuffleDependency, a ShuffleDependency subtype that declares a shuffle's output can be read incrementally -- a consumer stage may begin reading while the producer is still running. This is the first-class marker a later DAGScheduler change will use to co-schedule the producer and consumer (a 'pipelined group') and to select an incremental shuffle implementation. This PR adds only the type. On its own it behaves exactly like its parent ShuffleDependency -- code that matches ShuffleDependency continues to treat it as an ordinary materialized shuffle -- so it changes no existing behavior. The concurrent-scheduling and incremental-shuffle behavior are added separately. The parent's checksumMismatchFullRetryEnabled / checksumMismatchQueryLevelRoll- backEnabled params are intentionally not exposed, so they stay false: their stage-level recompute-and-rerun is moot for a pipelined group (any failure aborts the whole group and the caller reruns from scratch), and would also conflict with a consumer that has already read the output incrementally. Tests: PipelinedShuffleDependency is-a ShuffleDependency, is distinguishable from an ordinary dependency, registers its shuffle with a distinct shuffleId, keeps the checksum retry/rollback flags false, and forwards non-default constructor args (aggregator, mapSideCombine) to the parent correctly. Co-authored-by: Isaac
…eManager by dependency type Introduce spark.shuffle.manager.incremental: when set, SparkEnv instantiates a second ShuffleManager (a peer of the default spark.shuffle.manager, not a wrapper) and SparkEnv.shuffleManagerFor(dep) routes each shuffle to one of the two by dependency type -- a PipelinedShuffleDependency to the incremental manager, every other ShuffleDependency to the default. Routing is a pure function of the dependency, so the driver (registerShuffle) and executors (getReader/getWriter) agree without shared state; each manager only ever sees its own handle type, so no wrapping handle is needed. SparkEnv.get.shuffleManager remains the plain default manager, so nothing is installed in front of it.
…uching the default-manager check Restructure initializeStreamingShuffleOutputTracker so the pre-existing default-manager branch (which matches StreamingShuffleManager or MultiShuffleManager) is left untouched, and the new incremental-manager case (spark.shuffle.manager.incremental == StreamingShuffleManager) is handled by a separate early return. Extract the tracker construction into createStreamingShuffleOutputTracker so both paths share it. This keeps MultiShuffleManager out of this PR's diff entirely: the reference that remains is master's default-slot check, which this routing PR does not modify. Behavior is unchanged (routing suite tracker-init tests still pass). Co-authored-by: Isaac
…callers need no ShuffleManager Block-by-id resolution always uses the default manager (a pipelined shuffle is served out-of-band and produces no block-manager blocks). Add a SparkEnv.shuffleBlockResolver accessor that returns defaultShuffleManager.shuffleBlockResolver, and route the block-serving paths through it instead of SparkEnv.get.defaultShuffleManager.shuffleBlockResolver: - BlockManager: resolver accesses (getBlockData, merged blocks, migratableResolver, corruption diagnosis) go through a private shuffleBlockResolver accessor that preserves the deferred-init wait and the injected-manager (test) path. The manager reference is kept only for the unsupported-resolver error. - BlockManagerMasterEndpoint: the ESS-cleanup path uses the resolver accessor directly. Centralizes the "block resolution uses the default manager" invariant in one place. Updates the BlockManagerSuite mock to stub the new accessor alongside defaultShuffleManager. Co-authored-by: Isaac
…elinedShuffle by kind A ShuffleManager's type now declares what kind of shuffle it implements, instead of the kind being implicit in whether shuffleBlockResolver throws: - BlockingShuffle: materializes output as block-manager-addressed blocks served through a ShuffleBlockResolver (reads, push-merge, decommission migration). shuffleBlockResolver moves off ShuffleManager onto this trait. SortShuffleManager extends it. - PipelinedShuffle: output is read incrementally and served out-of-band, so there is no ShuffleBlockResolver. StreamingShuffleManager extends it and drops its throwing resolver stub. Callers that resolve blocks now match on BlockingShuffle rather than assuming every manager has a resolver: - SparkEnv.shuffleBlockResolver returns Option[ShuffleBlockResolver] (None when the default manager is not a BlockingShuffle). - BlockManager / BlockManagerMasterEndpoint handle the None case; block-serving paths require a BlockingShuffle default (a clear error otherwise, matching the previous throw). - ShuffleWriteProcessor's push-merge path matches BlockingShuffle before touching the resolver. - SparkCoreErrors.unexpectedShuffleBlockWithUnsupportedResolverError takes the resolver directly. MultiShuffleManager stays a plain ShuffleManager (it is slated for deprecation in favor of per-dependency routing) and keeps its own resolver method for the SortShuffleManager it delegates to. New tests assert the manager-kind types and that SparkEnv.shuffleBlockResolver is defined only for a blocking default manager. Co-authored-by: Isaac
…'s shuffle block resolver Deprecate MultiShuffleManager: the cluster-level single-slot approach is superseded by per-dependency routing (spark.shuffle.manager + spark.shuffle.manager.incremental). As a plain ShuffleManager (not a BlockingShuffle), it no longer participates in block-by-id resolution, ESS cleanup, or push-based merge when configured as the default manager -- this is a documented, accepted trade-off for a manager on its way out. Also restore BlockManager.shuffleBlockResolver to a cached lazy val (it had become a def in the trait-split change, recomputing the init-wait and type match on every block access in hot paths like getBlockData). Co-authored-by: Isaac
…ency routing MultiShuffleManager was the cluster-level single-slot way to run streaming and regular shuffles together. Per-dependency routing (spark.shuffle.manager for regular/blocking shuffles and spark.shuffle.manager.incremental for pipelined ones, each routed by dependency type) supersedes it, so remove the class and its suite. This also removes the only in-tree ShuffleManager that was neither a BlockingShuffle nor a PipelinedShuffle: the streaming-tracker init no longer needs to special-case it, and the default manager is now always a BlockingShuffle (built-in) unless a user configures a custom one. Co-authored-by: Isaac
…al ShuffleManager, simplify ESS resolver lookup - Rename BlockingShuffle -> BlockingShuffleManager and PipelinedShuffle -> PipelinedShuffleManager for consistency with the ShuffleManager they extend. - Seal ShuffleManager so it cannot be implemented directly; a manager declares its kind by extending BlockingShuffleManager or PipelinedShuffleManager. - BlockManagerMasterEndpoint's ESS-cleanup path calls SparkEnv.get.shuffleBlockResolver directly instead of a local wrapper val. Co-authored-by: Isaac
…ckingShuffleManager) Per review, keep MultiShuffleManager.scala and its suite rather than deleting them; the new per-dependency routing simply does not reference the class. Since ShuffleManager is now sealed, MultiShuffleManager extends BlockingShuffleManager (it resolves blocks via the inner SortShuffleManager it delegates regular shuffles to). Restore the SparkEnv streaming-tracker-init recognition of a MultiShuffleManager default, and drop a redundant comment in StreamingShuffleManager. Co-authored-by: Isaac
…drop dead endpoint param Address three review comments on the dependency-typed routing PR: - Do not seal ShuffleManager (reverses the earlier seal). sealed would restrict the third-party ShuffleManager extension point from SPARK-45762 (user-jar managers loaded reflectively), and being frontend-only it does not even enforce that against Java or reflective implementations -- its only effect is forcing in-tree subtypes into one file. The BlockingShuffleManager / PipelinedShuffleManager split stays; the "declare your kind" contract is instead enforced by per-slot startup validation (a follow-up commit). - Remove the now-dead _shuffleManager constructor parameter (and its import) from BlockManagerMasterEndpoint: ESS cleanup resolves blocks via SparkEnv.get.shuffleBlockResolver, so nothing consumes it. Drop the argument at all call sites (SparkEnv and three test suites). - Fix the SPARK-45762 test in SparkSubmitSuite. Its user-jar ShuffleManager is compiled by javac at test runtime, so it is not covered by Test/compile; moving shuffleBlockResolver onto BlockingShuffleManager left its @OverRide dangling. The generated class now implements BlockingShuffleManager (its body already stubs every required method), which also exercises a reflectively-loaded BlockingShuffleManager from a foreign classpath. Co-authored-by: Isaac
…by kind, not a default Per review, do not model one manager as "the default" -- that hides the blocking-vs-pipelined distinction the routing is built on, and it left a latent sharp edge: SparkEnv.shuffleBlockResolver resolved through the default slot only, so a blocking manager configured in the incremental slot would have had unreachable blocks. SparkEnv now holds two managers keyed by kind: - blockingShuffleManager (spark.shuffle.manager): a BlockingShuffleManager that serves all regular, materialized shuffles and owns block-by-id resolution. shuffleBlockResolver always resolves through it -- the single, well-typed source for reads, push-merge, and decommission migration. - pipelinedShuffleManager (spark.shuffle.manager.incremental): an optional PipelinedShuffleManager that serves pipelined dependencies out-of-band. initializeShuffleManager validates each slot's kind and fails fast otherwise: a non-blocking manager (including a pipelined-only or kind-less one) is rejected from the default slot, and a blocking manager from the incremental slot. This closes the unreachable-blocks case by construction and is the runtime enforcement of "declare your kind" that replaces the reverted `sealed`. A MultiShuffleManager is still accepted as the blocking manager -- it is blocking-kind and resolves blocks via its inner SortShuffleManager -- so it keeps working as the single-slot streaming+sort manager while routing continues to key on dependency type. The streaming output tracker is initialized when the incremental manager is a StreamingShuffleManager, or the blocking manager is a MultiShuffleManager; the old "StreamingShuffleManager as the default manager" case is gone, since a bare StreamingShuffleManager is pipelined and rejected from the default slot. ShuffleExchangeExec's sort-shuffle detection reads blockingShuffleManager. Routing tests split the recording test double by kind and cover both fail-fast paths. Co-authored-by: Isaac
…after kind validation The kind-based slot validation added to SparkEnv rejects a non-blocking manager in the spark.shuffle.manager (default) slot. The streaming test suites predate the routing model and configured StreamingShuffleManager -- a PipelinedShuffleManager -- as the default, so they now throw IllegalArgumentException at SparkContext startup. These suites are out-of-diff from the validation change, so they broke CI without failing the routing tests. Move StreamingShuffleManager to the incremental slot (spark.shuffle.manager.incremental), which is the correct slot for a pipelined manager and still initializes the StreamingShuffleOutputTracker the reader/writer constructors assert. StreamingShuffleSuite's assertions that the manager is streaming now check pipelinedShuffleManager instead of the (blocking) default. The tracker-init suite gains a slot-aware helper and a case that a MultiShuffleManager default also initializes the tracker. Co-authored-by: Isaac
…ing; add binding policy
Make the built-in streaming shuffle manager the default for spark.shuffle.manager.incremental, the
same way "sort" is the default for spark.shuffle.manager:
- Register "streaming" as a short alias for StreamingShuffleManager in ShuffleManager.resolveShortName
(alongside "sort"/"tungsten-sort"), so the incremental slot accepts the same style of short names.
- Change spark.shuffle.manager.incremental from optional to createWithDefault("streaming"). An empty
value explicitly disables the incremental slot, in which case a pipelined dependency falls back to
the blocking manager. SparkEnv resolves the (now always-present) config, treating empty as
disabled, and the streaming output tracker initializes based on the instantiated pipelined manager
rather than re-reading the config.
Also add .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) to the config, required by
SparkConfigBindingPolicySuite for any new config (a shuffle manager is a physical-execution choice
that does not change how a view/UDF body resolves). Matches the sibling streaming configs.
Tests: the routing suite's fallback and pre-init cases now set the incremental slot empty to
exercise the "no pipelined manager" path explicitly; the tracker-init suite sets the incremental
slot for every case (empty disables it) and drops the case that put a blocking manager in the
incremental slot (now rejected by the kind validation, and already covered by the routing suite's
non-streaming pipelined double).
Co-authored-by: Isaac
…mirroring the blocking one Per review, model the pipelined manager exactly like the blocking manager: it has a built-in default implementation (the streaming shuffle manager), so it is always created rather than optional. - _pipelinedShuffleManager and the pipelinedShuffleManager accessor are now a plain PipelinedShuffleManager (not Option), initialized in initializeShuffleManager the same way as the blocking manager -- resolving spark.shuffle.manager.incremental (default "streaming") and requiring a PipelinedShuffleManager. There is no "disabled" incremental slot. - shuffleManagerFor no longer falls back to the blocking manager for a pipelined dependency: a pipelined dependency is always served by the pipelined manager. - unregister/stop null-guard both managers (both are null until the deferred init runs), matching the blocking manager's handling. Also address the tracker-init review comments: consolidate the streaming-incremental and MultiShuffleManager-default conditions into a single check, and add a TODO to remove the MultiShuffleManager clause once MultiShuffleManager is gone. Tests: drop the now-impossible "incremental disabled -> fallback" and "tracker disabled" cases; add a case that the tracker initializes by default (streaming incremental manager); the pre-init no-op test nulls both manager fields. Co-authored-by: Isaac
…lity, consistent tracker check Address three non-blocking review comments on the dependency-typed routing PR: - Tighten the four new SparkEnv routing accessors (blockingShuffleManager, pipelinedShuffleManager, shuffleBlockResolver, shuffleManagerFor) to private[spark]. They are internal routing plumbing returning private[spark] manager types, and every caller is in-tree under org.apache.spark, so this matches the stated intent and removes a public-method-returns-package-private-type leak on the @DeveloperAPI SparkEnv class. The pre-existing shuffleManager accessor stays public (now @deprecated) since tightening a shipped @DeveloperAPI accessor would itself be a breaking change. - In initializeStreamingShuffleOutputTracker, inspect the instantiated blocking manager (_blockingShuffleManager.isInstanceOf[MultiShuffleManager]) instead of re-reading the config class name, so it is symmetric with the sibling incrementalIsStreaming check and keys on the object actually created. - Fix an inaccurate comment in BlockManagerMasterEndpoint: shuffleBlockResolver is None only before the manager is initialized, not "when the default manager is not blocking" -- the kind validation in initializeShuffleManager makes a non-blocking default manager impossible. Co-authored-by: Isaac
…rom @DeveloperAPI SparkEnv SparkEnv is @DeveloperAPI, so its Scala doc is rendered into Java API docs by unidoc. The new routing accessors documented their return types with [[BlockingShuffleManager]] / [[PipelinedShuffleManager]] / [[ShuffleBlockResolver]] / [[ShuffleManager]], which are all private[spark] and therefore have no Java doc page. Scaladoc [[X]] becomes javadoc {@link X}, and javadoc fails with "reference not found" for a link to a non-public type, breaking the Javaunidoc / doc build. Render those references as backtick code spans instead of doc links (the same idiom the deprecated shuffleManager accessor already uses). The public @DeveloperAPI types PipelinedShuffleDependency and ShuffleDependency keep their [[...]] links, since those resolve. Co-authored-by: Isaac
…nregisterShuffle contract The id-only RemoveShuffle cleanup path (SparkEnv.unregisterShuffleFromAllManagers) cannot tell which manager owns a shuffle, so it now broadcasts unregisterShuffle to every configured manager -- including ones that never handled the shuffle. That relies on an unknown/already-removed id being a harmless no-op, which the built-in managers honor but the ShuffleManager interface never promised. Per review, make it an explicit part of the contract: unregisterShuffle must treat an unknown or already-removed shuffleId as an idempotent no-op (return false) rather than throwing or mutating unrelated state. Documentation-only change to the trait scaladoc. Co-authored-by: Isaac
jerrypeng
force-pushed
the
stack/pipelined-shuffle-pr6-failfast
branch
2 times, most recently
from
July 20, 2026 17:40
a5d70cb to
3864abc
Compare
…cy's active behavior; fix ESS-coexistence note Two doc follow-ups from cloud-fan's review: - PipelinedShuffleDependency scaladoc said it "only declares the capability" and "behaves exactly like its parent". That is stale: construction now routes shuffle registration to the incremental manager (via SparkEnv.shuffleManagerFor) and unconditionally disables push-based shuffle merge (setShuffleMergeAllowed(false)). Describe those active differences instead. - ShuffleManager's External-Shuffle-Service coexistence note was phrased as an incomplete, conditional sentence; reword per the reviewer's suggestion. Comment-only; no behavior change. Co-authored-by: Isaac
…ined shuffle
Native DAGScheduler support for concurrent-stage scheduling over a
PipelinedShuffleDependency (from the prior PR). A pipelined shuffle is
incrementally readable: its consumer stage may begin reading output while the
producer is still running, so the two are co-scheduled ('pipelined group')
instead of the consumer waiting for the producer to materialize.
submitStage: when a stage has missing parents, classify them by their shuffle
dependency type. A parent read through a PipelinedShuffleDependency is a
pipelined parent. The stage is co-scheduled with its producers (its tasks
submitted immediately) only if every missing parent is pipelined AND each is
already running; otherwise it parks in waitingStages exactly as before. A stage
with a regular missing parent, or a pipelined parent not yet running, waits and
is reconsidered later. This is inert for jobs with no pipelined dependency --
the full DAGSchedulerSuite is unchanged.
submitWaitingPipelinedChildStages: the 'producer started running' analog of
submitWaitingChildStages. When a pipelined producer starts, its waiting
consumers are co-scheduled immediately (not only when the producer completes),
so a consumer parked because its producer sat behind a regular shuffle is
co-scheduled as soon as the producer runs. Cascades transitively.
handleJobSubmitted: reject a job that uses a pipelined dependency when
speculation is enabled -- a speculative producer copy would race a consumer
already reading the producer's partial output, with no commit barrier. The check
runs before stage creation so a rejected job leaves no scheduler state behind.
Tests (DAGSchedulerSuite): concurrent submission, inertness for a regular
shuffle, mixed pipelined+regular parents, a deep all-pipelined chain, a pipelined
producer behind a regular shuffle (must not co-schedule early), two pipelined
parents (co-schedule and re-park semantics), fan-out reconsideration to multiple
consumers, transitive cascade, no double-submission on producer completion, and
speculation rejection (plus that a regular job under speculation is not
rejected).
Co-authored-by: Isaac
…uster capacity Best-effort admission check for pipelined groups. All member stages of a group must run concurrently, so if the group's total task demand exceeds the cluster's total concurrent-task capacity it can never be co-resident and, lacking an out-of-band slot reservation, would deadlock (the consumer holds slots waiting for producer output while the producer cannot get slots to produce). We fail fast with a clear CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT error instead. When a pipelined group is about to be co-scheduled, submitStage computes the group's full pipelined-connected component (pipelinedGroupOf) and compares its summed task demand against maxConcurrentTasksForStage (production: sc.maxNumConcurrentTasks for the stage's resource profile). If demand exceeds capacity, the job is aborted; the already-launched producers are torn down by the normal job-abort path (cancelRunningIndependentStages). This is deliberately best-effort, not the atomic gang reservation deferred to a later hardening step: it compares the whole group's demand against TOTAL capacity (not free slots), checked once at co-schedule time. That converts the common under-provisioned case (a group that can never fit) into a clear error rather than a hang; races against other concurrently admitting work are left to the future atomic version. Inert for jobs with no pipelined dependency. maxConcurrentTasksForStage is a protected seam so tests can control reported capacity without changing the cluster's core count. Tests (DAGSchedulerSuite): a group too large to co-fit fails fast with CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT; a group that fits is co-scheduled normally; producer/consumer task sets marked isPipelined; regular task sets are not. Co-authored-by: Isaac
…4.1) Upgrade the gang-admission slot check from total capacity to currently-FREE slots: free = per-profile total capacity minus tasks already running for OTHER work (other groups / regular jobs), excluding the group's own already-running members. Adds TaskSchedulerImpl.runningTasksForOtherWorkInProfile (single-lock, resource-profile-scoped, zombie-filtered) and a DAGScheduler seam. Comparing against total capacity would admit a group that fits in principle but cannot co-fit beside a busy neighbor, then hang; free-slot admission fails fast instead. Also bumps the incremental-manager config .version to 4.3.0.
… add a slot-check disable flag Two updates to the pipelined-group slot admission check (spec S4.1), per an updated spec: - Count outstanding demand, not just running tasks. The occupancy of OTHER work in a resource profile now sums each task set's not-yet-completed tasks (numTasks - tasksSuccessful), i.e. running plus enqueued, instead of only the tasks actively running. A neighbor's queued backlog is real committed demand that can starve a co-scheduled group, so charging only running tasks could admit a group that then hangs once the backlog launches. Renamed TaskSchedulerImpl's helper to outstandingTasksForOtherWorkInProfile and the DAGScheduler seam to outstandingTasksForOtherWork to reflect the new semantics. - Add spark.scheduler.pipelinedGroup.slotCheck.enabled (internal, default true). When false, pipelinedGroupExceedsCapacity is skipped entirely and a group is co-scheduled unconditionally. This mirrors ConcurrentStageDAGScheduler's ability to disable its slot check, for deployments that admit capacity out-of-band (e.g. a slot reservation) and own admission themselves. Tests: the TaskSchedulerImpl unit test now submits more tasks than slots and asserts the count includes the enqueued tasks; a new DAGScheduler test asserts that with the check disabled an over-capacity group is co-scheduled rather than failed. Co-authored-by: Isaac
… the group finishes Group-observable completion for pipelined groups (spec S5). A stage co-scheduled with a still-running pipelined producer (a pipelined consumer) must not have its successful completion processed early: doing so would advance job completion and cancel the still-running producer (via cancelRunningIndependentStages), or make the consumer's output observable before the producer's. handleTaskCompletion: near the top, before any of the event's side effects, if a Success event belongs to a pipelined consumer with unfinished pipelined producers, buffer the whole CompletionEvent and return. This defers the entire event (coarse model), so its side effects (accumulator update, TaskEnd listener event, stage/job completion) run exactly once -- at replay. markStageAsFinished: when a pipelined producer finishes, release its deferred consumers -- but only when the producer's outcome is final. A ShuffleMapStage that finished without error yet is not isAvailable (an output missing; it will be resubmitted) is NOT treated as done, so its consumers stay deferred until the reattempt makes it available. On genuine success the buffered events are replayed; on producer failure they are dropped (the group reruns, S6) but their TaskEnd events are still emitted so listeners do not leak active-task accounting. cleanupStateForJobAndIndependentStages: drop any deferral keyed on a removed stage and remove it from other consumers' pending-producer sets, so no deferral outlives its job (e.g. on abort). assertDataStructuresEmpty also checks the deferral map is empty. Inert for jobs with no pipelined dependency: the deferral map is never populated, the completion check is a always-miss map lookup, and release/cleanup are no-ops. Tests (DAGSchedulerSuite): early-finishing consumer does not end the job or cancel its running producer; normal producer-then-consumer ordering; buffered completions dropped when the producer fails; TaskEnd fired exactly once (no buffer+replay duplication); deferral released only when the producer is genuinely available. Co-authored-by: Isaac
…ts, reject dynamic allocation Two correctness fixes to the pipelined-group slot admission, found in review: - Skip zombie attempts in outstandingTasksForOtherWorkInProfile. The occupancy count summed numTasks - tasksSuccessful over every attempt in taskSetsByStageIdAndAttempt, including zombie (superseded) attempts. A retried/killed stage can have both a zombie and a live attempt in the map at once, and the live attempt already re-runs the zombie's outstanding tasks -- so counting both double-counted that stage's demand, inflating occupancy and potentially failing a pipelined group with CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT that would actually fit. Add the !isZombie filter used elsewhere on this map. - Reject a pipelined-shuffle job under dynamic allocation. A pipelined group is gang-scheduled and its free-slot check measures currently-active executors; under dynamic allocation a job can be submitted before any executor has spun up, so the group would be failed against a transient 0-slot snapshot even though the cluster would soon have capacity. Barrier scheduling forbids the same combination (checkBarrierStageWithDynamicAllocation); pipelined groups now do likewise. The former rejectSpeculationWithPipelinedShuffle is generalized to rejectUnsupportedPipelinedJob, which rejects both speculation and dynamic allocation (single RDD-graph walk, run only when a relevant feature is enabled). Tests: a zombie + live attempt is counted once, not twice; a pipelined job under dynamic allocation is rejected while a regular job under dynamic allocation is not. Co-authored-by: Isaac
…by a later PR in the stack isPipelinedGroupMember is defined here alongside the other group-topology helpers but is first used by the group-atomic failure handling added in a later PR of this stack (TaskSet.isPipelined and the member-failure fail-fast, spec S6). Reviewed in isolation this change reads as introducing an unused private method; note the forward reference in the scaladoc so it is not mistaken for dead code. Co-authored-by: Isaac
…unning plus enqueued The admission check counts OTHER work's OUTSTANDING tasks (running plus enqueued, `numTasks - tasksSuccessful`) against a resource profile's capacity -- as `outstandingTasksForOtherWork` / `TaskSchedulerImpl. outstandingTasksForOtherWorkInProfile` and the config doc already state. Two spots in DAGScheduler still described this as only the tasks "already running", which was stale after the count was widened to include enqueued tasks; the `pipelinedGroupExceedsCapacity` scaladoc even contradicted its own later "running or enqueued" line. Reword both to "outstanding -- running plus enqueued" so the prose matches the code. Doc/comment only; no behavior change. Co-authored-by: Isaac
… admit the group up front v1 (M1, real-time-mode scope) supports a job that is either all-regular or all-pipelined, not a mix. This lets an all-pipelined job's whole stage graph be one pipelined group with no regular prefix, so gang admission can be decided up front -- before any member stage is submitted -- rather than mid-DAG once a producer is already running. Changes in handleJobSubmitted (before createResultStage, so a rejection leaves no partial scheduler state, exactly like the barrier slot check and the speculation/DA reject): - classifyJobShuffleKinds walks the RDD graph and rejects a job that mixes a pipelined shuffle with a regular one, fail-fast. - rejectUnadmittablePipelinedGroup computes the whole all-pipelined group's concurrent-task demand from the RDD graph and checks it against free slots (maxNumConcurrentTasks minus other work's running-plus-enqueued demand, spec S4.1); fails the job if it cannot fit. No scheduler retry -- a transient shortfall is the caller's to retry (e.g. the streaming batch loop reruns the batch). This is true all-or-nothing gang admission: the whole group is admitted, or the job fails before any member runs, so a member is never left running while a sibling cannot get slots. The late slot check in submitStage's co-schedule branch (which measured a mid-flight snapshot after a producer was already running) is removed, along with the now-unused pipelinedGroupExceedsCapacity / pipelinedGroupOf; the capacity/occupancy seams are refactored to be resource-profile-keyed. Tests: mixed jobs rejected up front; all-PG group admitted/rejected up front (2- and 3-stage); the removed mid-DAG prefix tests are dropped as that shape is no longer supported. Co-authored-by: Isaac
…oncurrentStageDAGScheduler Rename the deferred-completion machinery to match the production RTM scheduler (ConcurrentStageDAGScheduler), so reviewers familiar with that code read the same concepts here. Pure rename; no behavior change. - pipelinedConsumerDeferrals -> dependentStageMap - DeferredCompletion -> DependentStageInfo - pendingProducers -> parents - bufferedEvents -> delayedTaskCompletionEvents The pipelined-specific helpers (isPipelinedProducer, isPipelinedGroupMember, submitWaitingPipelinedChildStages, rejectUnadmittablePipelinedGroup) keep their names -- they belong to the type-driven pipelined model and have no clean analog in the reference scheduler. Co-authored-by: Isaac
…t lands on a torn-down stage A deferred pipelined-consumer completion fires its TaskEnd inline on first completion; only the stage/job bookkeeping is re-posted (finishOnly=true) once the producer finishes. The cancelled-stage guard in handleTaskCompletion (which posts TaskEnd for an event whose stage is no longer tracked) ran unconditionally and before the finishOnly check. If the group is torn down (a sibling aborts) between a finishOnly re-post and its dequeue, the replay lands on the removed stage and the guard would re-post TaskEnd -- double-counting it in listeners (AppStatusListener's active-task count could go negative). Skip postTaskEnd for a finishOnly event in that guard: its per-task TaskEnd already fired inline. A regression test on the failure PR (which has the group-atomic abort this needs) drives a finishOnly replay onto a torn-down stage and asserts no extra TaskEnd; reverting this guard makes it fail (TaskEnd count goes up by one). Co-authored-by: Isaac
jerrypeng
force-pushed
the
stack/pipelined-shuffle-pr6-failfast
branch
2 times, most recently
from
July 20, 2026 23:15
0580742 to
20e3aa5
Compare
…cumulator deltas are not un-merged
The drop-path comment implied full cleanliness ("only the deferred completion bookkeeping
was buffered ... must NOT be applied"). But a deferred consumer's per-task effects, including
updateAccumulators, run inline before buffering, so on a producer-failure drop the already-
applied accumulator deltas are NOT un-merged. Note that this is consistent with base Spark
(accumulators are non-transactional across abort+rerun): the rerun re-delivers under the S5
idempotent-sink contract, and RTM SQL metrics are fresh per batch, so there is no cross-batch
double-count. Comment-only.
Co-authored-by: Isaac
jerrypeng
force-pushed
the
stack/pipelined-shuffle-pr6-failfast
branch
from
July 20, 2026 23:41
20e3aa5 to
12ca05c
Compare
…ments Strip internal shorthand that a reader of merged Apache Spark cannot resolve -- version/milestone tags, design-doc spec section numbers, and fine-grained/coarse model terminology -- from the pipelined-shuffle comments, one test title, and the mixed-job error message, rewording to plain behavioral language. Comment, test-title, and error/assertion-string text only; no behavioral change. Co-authored-by: Isaac
…ask failure A pipelined-group member reads/writes a transient shuffle that cannot be re-read in isolation, so a single member task failure must fail the whole group -- which fails the job, letting the caller (the streaming batch loop) rerun -- rather than be retried per-task. TaskSet gains an isPipelined flag, set by DAGScheduler.isPipelinedGroupMember (true if the stage produces a PipelinedShuffleDependency or reads one at a shuffle boundary). For a pipelined task set, TaskSetManager caps attempts at 1 (effectiveMaxTaskFailures) and counts failures that are normally excluded -- most importantly executor loss not caused by the app, which strands the transient output. It deliberately does NOT count the benign reasons TaskKilled (a losing duplicate attempt after another succeeded, or a deliberate kill) and TaskCommitDenied (a speculation commit race): counting those would abort the group spuriously even though the task effectively succeeded. Recovery itself is not handled here: the job aborts and the streaming layer reruns the batch from the last committed offset with fresh stage ids. Inert for non-pipelined task sets: effectiveMaxTaskFailures and the counting condition both reduce to the original expressions when isPipelined is false. Tests (TaskSetManagerSuite): a pipelined set aborts after one genuine failure; counts executor loss; does NOT abort on TaskKilled or TaskCommitDenied; a non-pipelined set is unchanged (executor loss uncounted, retries apply). (DAGSchedulerSuite): both producer and consumer task sets are marked isPipelined; regular task sets are not. Co-authored-by: Isaac
… producer at both layers A pipelined shuffle is transient: a once-through live stream with no retained, addressable output. Unlike a regular shuffle there is nothing for a later or concurrent job to re-read, so a pipelined producer stage must never be reused. Spark can reuse a shuffle-map stage at two independent layers, and both must be closed for a pipelined shuffle (spec S4): 1. shuffleIdToMapStage stage-object reuse: getOrCreateShuffleMapStage fails fast with PIPELINED_SHUFFLE_CROSS_JOB_REUSE if a cached pipelined producer stage would be bound to a second job. A sequential re-run is unaffected: the earlier job's cleanup drops the stage from shuffleIdToMapStage, so a later job mints a fresh producer bound only to itself (this is how an RTM micro-batch reruns). 2. MapOutputTracker-availability reuse: a pipelined shuffle is never registered with the MapOutputTracker as durable output. ShuffleMapStage tracks its completed partitions locally in a monotonic set; numAvailableOutputs and findMissingPartitions read that set for a pipelined stage. Because executor/ host loss only strips MapOutputTracker entries, it can no longer flip a completed pipelined producer to unavailable and trigger a resubmit. Together these fix SC-235532 (an RTM streaming-shuffle query hangs after executor loss: the mapper is resubmitted and its streaming writer blocks forever in awaitTerminationAcks). Two further resubmit routes for a pipelined producer are closed here: - TaskSetManager.executorLost re-enqueues an already-successful map task via the Resubmitted channel on executor decommission (when getMapOutputLocation returns None, which is always true for an unregistered pipelined shuffle). This bypasses handleFailedTask, so the group-atomic abort would never see it. Guarded with !taskSet.isPipelined; a genuine loss still routes through handleFailedTask and fails the group. - handleTaskCompletion records a pipelined success unconditionally, before the "possibly bogus epoch" check. A straggler success whose executor is already marked lost otherwise removed its partition from pendingPartitions without recording it, leaving the stage "done but not available" and driving a resubmit. (Push-based shuffle merge is disabled for a PipelinedShuffleDependency in the routing change that introduces the incremental manager, since that is where a pipelined dependency would otherwise reach the merge path; see that commit.) Inert for jobs without a pipelined dependency: every branch is gated on the dependency subtype or ShuffleMapStage.isPipelined, and the full existing DAGSchedulerSuite / TaskSetManagerSuite pass unchanged. Co-authored-by: Isaac
…ipelined group on a member FetchFailed (SC-233883) A FetchFailed uniquely bypasses the group-atomic maxTaskFailures=1 lever: the base TaskSetManager marks a FetchFailed task successful, zombies the set, and returns without counting the failure (FetchFailed.countTowardsTaskFailures is false), so the whole recovery happens in the DAGScheduler's FetchFailed handler, which resubmits the map stage in isolation and recomputes serially. For a pipelined group that is a deadlock (SC-233883): the transient pipelined shuffle cannot be re-read, and members are co-scheduled, so a lone-stage resubmit is never valid (spec S6, "single-stage resubmit is disabled for group members; any member failure -- including FetchFailed -- fails the whole group"). Route a FetchFailed whose failed stage or map stage is a pipelined group member to abortStage instead: aborting the failed stage tears down the job's co-scheduled member stages via cancelRunningIndependentStages and fails the job, and the caller (the streaming batch loop) reruns the batch from scratch. The teardown is complete because a v1 pipelined group is exactly one job: the producer is registered to the same job (updateJobIdStageIdMaps walks parents), so cancelRunningIndependentStages kills/finishes it whether it is still running or already finished. This coupling to "group == one job" is guaranteed by the cross-job-reuse fail-fast (PIPELINED_SHUFFLE_CROSS_JOB_REUSE); if fan-out or cross-job sharing is ever enabled, this branch must switch to tearing down pipelinedGroupOf(failedStage) explicitly rather than relying on job-atomicity. Inert for jobs without a pipelined dependency: isPipelinedGroupMember returns false for every regular stage (it is a pipelined producer only if its shuffleDep is a PipelinedShuffleDependency, and a consumer only if it reads through one), so the original FetchFailed handling runs unchanged. The full existing DAGSchedulerSuite (including all FetchFailed / shuffle-merger tests) passes. Tests: a member FetchFailed (consumer reading the pipelined edge) fails the job with no single-stage resubmit; a producer FetchFailed reading its regular input (spec S7) also fails the group. Co-authored-by: Isaac
…mit authorization Spec S5 requires that when a pipelined group reruns after a failure, a result stage whose tasks already committed can commit again -- OutputCommitCoordinator otherwise permanently denies re-commit for a committed partition (keyed by stage id; a Success clears nothing). This is already satisfied by M1's existing mechanisms, and this test locks that in: - Group teardown runs the committed result stage through markStageAsFinished -> outputCommitCoordinator.stageEnd, clearing its committer state, so no stale authorization survives the failed attempt (option (b) in S5). - The caller's rerun is a NEW job whose stages get fresh stage ids, so the coordinator has no prior committer for them regardless (option (a) in S5). The test commits a result task in a pipelined group (registering committer state), fails the group via its producer, asserts the coordinator's committer state is cleared on teardown, then reruns the batch as a new job and asserts the rerun's result stage gets a fresh stage id and completes -- re-committing its partitions. No production change: M1's group-atomic-failure + caller-reruns-the-batch design already conforms to S5. This adds the regression test that proves it. Co-authored-by: Isaac
…or-all-PG restriction The M1 all-regular-or-all-PG restriction (PR3) rejects any job with a regular shuffle in a pipelined job up front, including a regular-shuffle prefix feeding a pipelined producer (regularRoot --regular--> producer(pipelined) --pipelined--> consumer). Three tests exercised that now-rejected shape: - "a consumer is a group member even if a REGULAR dep precedes the pipelined one" - "a FetchFailed on a pipelined producer reading its regular input (SC-233883)" - "a group-aborting FetchFailed still unregisters the dead executor's outputs" Remove them: the shape is out of the RTM scope (scan-of-files --pipelined--> stateful, no upstream shuffle). SC-233883's core behavior -- a member FetchFailed fails the whole group rather than resubmitting a single stage -- remains covered by the pure-shape test "a FetchFailed on a group member fails the group, not a single-stage resubmit (SC-233883)". Add a rejection test documenting that a regular-shuffle prefix feeding a pipelined producer is rejected up front. Co-authored-by: Isaac
…test non-vacuous; fix a spec citation
From an adversarial review:
- The SC-235532 executor-loss regression test completed the producer with
makeMapStatus("hostA-exec"), which registers the output under executor id
"hostA-exec-exec" (makeBlockManagerId appends "-exec"), while the test loses
executor "hostA-exec" -- so removeOutputsOnExecutor never matched and the test
passed even with the ShuffleMapStage pipelined-availability fix reverted (it
did not guard its named regression). Use makeMapStatus("hostA") so the lost
executor id matches the registered output. Verified: the test now FAILS when
the fix is reverted.
- Fix a spec citation in the FetchFailed group-abort comment: "registers no map
outputs in the tracker" is a spec-S4 (transient / no durable output) fact,
cited as S4 everywhere else; it was mistakenly attributed to S6 (the
group-atomic failure section).
Co-authored-by: Isaac
… descriptive prose Comments and test names referenced internal Databricks Jira tickets (SC-235532, SC-233883) and internal milestone tags (M1.2 / M1.4 / M1.6 / M1.8), which are meaningless to the Apache community. Replace each with a short description of the behavior it denotes (e.g. the streaming-writer resubmit hang, the group-atomic abort), keeping the technical content and dropping the opaque identifiers. No behavior change; comment/test-name only. Co-authored-by: Isaac
…eap per-job flag isPipelinedGroupMember(stage), used to tag TaskSet.isPipelined on every submitMissingTasks, walks the stage's RDD graph -- pure overhead for a regular job that has no pipelined dependency. Record whether a job uses a PipelinedShuffleDependency once at submission (ActiveJob.hasPipelinedDependency, from the hasPipelined already computed in handleJobSubmitted) and short-circuit the walk when it is false. No behavior change for a pipelined job (the flag is true, so the membership check runs exactly as before); regular jobs now skip the walk entirely. Co-authored-by: Isaac
…er entry is registered but stays empty The availability comment said a pipelined shuffle is "NOT registered with the MapOutputTracker", which could read as contradicting createShuffleMapStage (it does call registerShuffle for every ShuffleDependency, pipelined included). Clarify the precise behavior: the empty shuffle-status entry is registered to keep containsShuffle / unregisterShuffle bookkeeping uniform, but no map output is ever populated into it (registerMapOutput runs only on the non-pipelined path), so availability is read from the stage's local monotonic set instead. Comment-only. Co-authored-by: Isaac
…rd; reconcile a ShuffleMapStage comment
- Regression test for the finishOnly TaskEnd guard (fixed on the scheduling PR): a deferred
consumer's TaskEnd fires inline, the group is then aborted via a member FetchFailed (which
removes the consumer stage), and a finishOnly replay is delivered onto the torn-down stage.
Asserts the replay posts no additional TaskEnd. Lives here because it needs the
group-atomic FetchFailed abort. Reverting the guard makes it fail (TaskEnd count +1).
- Reconcile numAvailableOutputs's inline comment ("not tracked in the MapOutputTracker")
with the field-level comment (which correctly says the entry is registered but never
populated) -- both now say "outputs never populated / entry stays empty".
Co-authored-by: Isaac
…red consumer deferral Deferral-interaction coverage. The deferral buffer (dependentStageMap) is the only mutable state this feature adds and must never outlive its job. This drives the explicit job-cancellation path (JobCancelled -> handleJobCancellation -> failJobAndIndependent- Stages), which was not directly exercised against a buffered deferral, and asserts the entry is fully cleaned up (map empty) with no buffered success applied as a result. Verified non-vacuous: neutering BOTH cleanup mechanisms -- the release-drop in cancelRunningIndependentStages (primary, fires while the producer is running) and the cleanupStateForJobAndIndependentStages sweep (backstop) -- makes the test fail with a leaked deferral entry. Co-authored-by: Isaac
…whole pipelined group Defense-in-depth for the group-atomic failure model at the DAGScheduler layer. The maxTaskFailures=1 mechanism is covered at the TaskSetManager layer, and producer-failure teardown is covered by the deferral-drop test, but there was no DAGScheduler-level test that a CONSUMER (result) task failing tears down a still-running co-scheduled producer. This adds one: it leaves the producer running, asserts it is running, fails the consumer task set (as a maxTaskFailures=1 abort surfaces), then asserts the running producer is torn down and the job fails. Revert-checked: neutering the cancelRunningIndependentStages teardown makes it fail (producer left running). Co-authored-by: Isaac
…ments Strip internal shorthand that a reader of merged Apache Spark cannot resolve -- version/milestone tags, design-doc spec section numbers, "RTM"/"PG" abbreviations, and fine-grained/coarse model terminology -- from the pipelined-shuffle failure-path comments and test comments, rewording to plain behavioral language. Comment and assertion-string text only; no behavioral change. Co-authored-by: Isaac
Reject, with a clear PIPELINED_SHUFFLE_UNSUPPORTED error, the pipelined-shuffle idioms a group cannot support in v1 (spec S9), so a misuse fails fast rather than silently mis-scheduling, hanging, or corrupting a group. Producer-side checks, in checkPipelinedProducerSupported (called from createShuffleMapStage when the shuffle is a PipelinedShuffleDependency), so a rejection throws during stage creation and leaves no partial scheduler state: - Barrier execution in a member stage (exposes output only after a global sync). - Dynamic resource allocation (gang admission needs a stable slot set). - Statically-indeterminate producer (its stage rollback-and-recompute recovery is moot under group-atomic failure -- a group never recomputes a single stage). - Checksum-mismatch full retry (the runtime counterpart to static indeterminism; also moot). Defensive: a PipelinedShuffleDependency does not enable it. - Push-based shuffle merge as the pipelined shuffle (finalize-then-read is the opposite of incremental reads). Defensive backstop: the dependency disables merge in its constructor. - A reliable RDD checkpoint in the member's within-stage chain (durable, lineage- truncated snapshot -> reintroduces cross-time reuse and needs a post-success recompute of the vanished transient input). Keyed on checkpointData being a ReliableRDDCheckpointData, not isCheckpointed (the write has not happened yet). Group-level check, in checkPipelinedGroupsSupportedInRDDGraph (called from handleJobSubmitted before any stage is created, alongside the speculation check, so a rejection fails the job up front with no partial state): - Fan-out: a pipelined producer feeding more than one consumer. 1:N is a supported model deferred to a later version (it needs multicast to N live readers); v1 rejects it. This also closes a group-demand undercount in the slot check when a sibling consumer stage is not yet created. Speculation and cross-job reuse (also S9) are rejected by earlier commits. The remaining S9 group-level rows -- mixed resource profiles and a regular shuffle internal to a group -- are structural invariants that do not arise for the single-profile, prefix*->PG->suffix* shapes v1 targets (a group is single-profile by construction, and groups split at regular-shuffle boundaries); they are deferred to the milestone that makes group formation first-class. Tests: barrier, indeterminate, reliable-checkpoint, and fan-out producers/groups are each rejected; regular (non-pipelined) shuffles using those same idioms are NOT rejected (inertness). Co-authored-by: Isaac
…R chain; widen group-check catch Follow-up from the M1.7 adversarial review: 1. (MEDIUM) The reliable-RDD-checkpoint rejection was scoped to a pipelined PRODUCER's within-stage chain only, but spec S9 covers any group MEMBER's chain -- including a CONSUMER, whose transient input IS the pipelined shuffle, so a reliable checkpoint there would re-read the vanished stream on recompute. Moved the reliable-checkpoint detection into checkPipelinedGroupsSupportedInRDDGraph, which runs at job submission over the whole RDD graph before any stage is created: it now walks both the producer chain (rooted at the produced RDD) and each consumer chain (rooted at a reading RDD). This also removes a partial-state leak the previous stage-creation-time consumer check would have caused (a consumer-stage throw after the producer stage had already registered). 2. (LOW) The group-check was in its own try with a SparkException-only catch, so an incidental non-Spark exception from the RDD-graph walk could escape handleJobSubmitted's job-level handler (with speculation disabled the walk is the first dependency traversal). Moved the call inside the existing createResultStage try, so any exception is handled by the same case e: Exception => jobFailed path. Not changed: fan-out keyed on shuffleId (a third, LOW review note) is correct as-is -- two distinct PipelinedShuffleDependency instances on one producer RDD are two independent 1:1 transient streams, not the multicast fan-out hazard; the check matches the spec's "more than one consumer for the same pipelined shuffle". Tests: a reliable checkpoint in a CONSUMER's chain is now rejected (new test); the producer-chain and all other rejection/inertness tests still pass. Co-authored-by: Isaac
…esses when a producer fails Regression coverage for a review concern: a pipelined consumer with more than one pipelined producer (a join) must DROP its buffered successes when any producer fails, never replay them via a still-pending sibling. On the full stack a member failure is group-atomic (abortStage tears down all members and marks each producer failed), so releaseDeferredPipelinedConsumers sees producerFailed=true for every producer and drops -- the "sibling succeeds -> replay" trigger cannot occur. This test pins that behavior (a producer feeding two consumers would be fan-out and is rejected; two producers feeding one consumer is a supported all-PG join). Co-authored-by: Isaac
…ed doc on where group idioms are checked The scaladoc said group-level idioms (fan-out, mixed resource profiles, internal regular shuffle) are "checked separately where the group is co-scheduled". That is inaccurate: fan-out is rejected up front at job submission by checkPipelinedGroupsSupportedInRDDGraph (before any stage is created), not at co-scheduling time; and mixed resource profiles / an internal regular shuffle are structural invariants that do not arise for v1's single-profile all-pipelined shape and are not separately checked at all. Correct the doc to point at the right place and stop implying checks that do not exist. Co-authored-by: Isaac
… so a group abort cannot leak the stage Follow-up to the fine-grained deferral change (which lives on the scheduling PR): update the abort-path test that previously guarded the coarse model's re-emit-on-teardown behavior. Under the fine-grained model a deferred consumer's TaskEnd is posted in real time when the task finishes -- only the stage/job-completion bookkeeping is deferred -- so the "leak the stage as perpetually running" failure the coarse model had to patch at teardown cannot arise. This test exercises the group-atomic FetchFailed->abort path (which this PR introduces), so it belongs here rather than on the scheduling PR. The test now asserts the consumer's TaskEnd is delivered inline (before the producer's fate is known), and that on the subsequent group abort the deferred completion is dropped (its result is never applied) with no duplicate TaskEnd re-emitted. Co-authored-by: Isaac
… reject mixed resource profiles (S9) Two review items on the pipelined-group fail-fast path: - Typed exception (was: brittle string match). handleJobSubmitted distinguished an up-front idiom rejection from a stage-creation failure by matching on e.getCondition == "PIPELINED_SHUFFLE_UNSUPPORTED" -- a rename or a wrapped cause would silently fall through to the generic "Creating new stage failed" branch and mis-report. Introduce a marker type PipelinedShuffleUnsupportedException (carrying the same error class, so the user-facing message is unchanged) and match on the TYPE. - Mixed-resource-profile fail-fast (spec S9). The gang slot check compares one demand against one profile's capacity (maxNumConcurrentTasks is per profile), so v1 requires a single-profile group. This was asserted in comments as a "structural invariant" but nothing enforced it. checkPipelinedGroupsSupportedInRDDGraph now collects the distinct explicit resource profiles across the group's RDDs in its existing single graph walk and rejects a group spanning more than one. Per-profile accounting remains a follow-up. Updated the two scaladocs that claimed mixed profiles "do not arise / are not checked". Also drops a stray internal milestone reference from a comment. Tests: a new test submits a producer/consumer group with two distinct resource profiles and asserts the up-front rejection; neutering the check makes it fail (non-vacuous). The existing idiom-rejection tests (barrier, indeterminate, checkpoint, fan-out) still pass through the typed-exception catch. Full DAGSchedulerSuite green (193). Co-authored-by: Isaac
…lined group; gate FetchFailed group check
Round-3 review follow-ups:
- Resource-profile admission gap (real). The mixed-profile check collected only distinct
EXPLICIT profiles (getResourceProfile() is null when unset) and rejected on count > 1,
while the slot check measures capacity against the DEFAULT profile. So a group running
uniformly on one non-default profile, or a non-default member mixed with default members,
passed the check yet was admitted against the wrong (default) capacity pool -- the exact
partial-residency deadlock gang admission prevents. Reject if ANY member carries an
explicit non-default profile: v1 requires the whole group on the default profile (spec
S9); per-profile accounting remains a follow-up. Added tests for the two previously-missed
shapes (uniform non-default; non-default + default), alongside the two-distinct case.
- FetchFailed group-membership check now gated on ActiveJob.hasPipelinedDependency, matching
the submitMissingTasks site, so a regular job's FetchFailed no longer walks the stage RDD
graph.
- Dropped an internal-review artifact ("Isaac-review probe") from a test comment.
Co-authored-by: Isaac
…dup a within-stage walk; warn when slot check is off Architecture-review follow-ups (no behavior change on the happy path): - Make the "job == one pipelined group, pre-admitted" invariant explicit and enforced. The submitStage co-schedule branch gang-schedules a group's members with no slot check, trusting that handleJobSubmitted already gang-admitted the whole job -- previously encoded only as a comment. Record ActiveJob.pipelinedGroupAdmitted on the admitted (or check-disabled) path and assert it at the co-schedule branch. This is a tripwire for a future version that relaxes job==group (multiple groups per job): it must move gang admission to a per-group check here and replace the assert, not delete it, or an unadmitted group could be gang-scheduled and deadlock. Verified the assert fires if admission is not recorded, and holds across all existing pipelined tests. - Collapse a duplicated within-stage walk: isPipelinedGroupMember's consumer walk was byte-identical to rddChainReadsPipelinedShuffle (differing only in the root RDD). Delegate to the latter as the single source of truth so a change to "what ends a within-stage chain" need not be made in two places. Behavior-preserving. - Warn (once) when spark.scheduler.pipelinedGroup.slotCheck.enabled=false, since disabling the only gang-admission deadlock check is safe only with out-of-band capacity reservation. Once-per-scheduler (event-loop-confined var) so it does not spam per submitted batch. Co-authored-by: Isaac
…ed, not just an assert The co-schedule branch verified the up-front gang-admission invariant with a bare `assert`, which is elided under -da (JVM assertions off in production). In v1 the invariant always holds, so this is only a forward tripwire -- but if a later version relaxes job==group and forgets to move gang admission to a per-group check here, a production build would silently gang-schedule an unadmitted group and could deadlock. Replace the assert with a fail-closed abortStage (reusing the same terminal jobFailed path as the sibling no-active-job case), so the invariant is enforced even under -da: a violation becomes a clean, rerunnable job failure instead of a silent deadlock. Verified: the guard never fires in normal v1 operation (all pipelined tests green); neutering admission recording makes it abort the group rather than co-schedule. Co-authored-by: Isaac
…ments Strip internal shorthand that a reader of merged Apache Spark cannot resolve -- version/milestone tags, design-doc spec section numbers, "all-PG" abbreviation, and fine-grained/coarse model terminology -- from the pipelined-shuffle fail-fast comments and test comments, rewording to plain behavioral language. Comment and assertion-string text only; no behavioral change. Co-authored-by: Isaac
jerrypeng
force-pushed
the
stack/pipelined-shuffle-pr6-failfast
branch
from
July 21, 2026 19:34
6e3149c to
4e8c399
Compare
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
1 participant
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.
What changes were proposed in this pull request?
This is the top of the pipelined-shuffle stack. On top of concurrent scheduling of a
PipelinedShuffleDependencygroup (#57341), it adds the group-atomic failure model and thefail-fast rejection of unsupported idioms that make co-scheduling safe under failure. (This PR
is stacked on #57286 and #57341; its net-new content is the failure and fail-fast layers described
below.)
A pipelined shuffle is transient and incrementally read: its output is streamed to a co-scheduled
consumer and is never materialized as durable, re-readable map output. That has two consequences the
scheduler must enforce.
Group-atomic failure (a pipelined group succeeds or fails as a whole). A transient shuffle
cannot be recomputed in isolation and its members run concurrently, so the stock "resubmit one
stage" recovery does not apply -- any member failure must fail the whole group, which the caller
then reruns (a streaming query restarts the batch).
TaskSetis taggedisPipelinedand pinnedto
maxTaskFailures = 1, so the first task failure aborts the set instead of retrying in place,which routes to a whole-group abort.
(except benign
TaskKilled/TaskCommitDenied), aborting the group.resubmitting the map stage in isolation (a lone-stage resubmit of a transient shuffle would
deadlock the group). The failed executor's regular outputs are still unregistered from the
MapOutputTracker, exactly as the base handler does, for the benefit of other jobs.ShuffleMapStagerecords its completed partitionslocally and monotonically (never in the
MapOutputTracker) and never flips back to unavailable, solosing an executor that held an already-consumed pipelined output cannot make the scheduler
resubmit the producer -- which would otherwise hang the producer's streaming writer waiting on
termination acks from reducers that already finished. The
TaskSetManager"Resubmitted"re-enqueue loop likewise excludes pipelined sets.
job to read, so binding a pipelined producer stage to a second job fails fast.
Fail-fast on unsupported idioms (spec S9). A pipelined group is rejected up front, before any
stage is created (so a rejection leaves no partial scheduler state), when it uses an idiom v1 cannot
support: a producer feeding more than one consumer (1:N fan-out needs multicast, deferred); a
barrier, statically-indeterminate, checksum-mismatch-retry, or push-merge producer;
a reliable RDD checkpoint anywhere in a member's within-stage chain (it reintroduces cross-time
reuse of a transient edge); or members with differing resource profiles (the gang slot check
compares one demand against one profile's capacity, so v1 requires a single-profile group). These
throw a typed
PipelinedShuffleUnsupportedException(carrying thePIPELINED_SHUFFLE_UNSUPPORTEDerror class), which
handleJobSubmittedmatches by type.Main changes:
DAGScheduler.scala-- the FetchFailed group-abort branch, the no-resubmit handling of apipelined
ShuffleMapStage, cross-job reuse rejection, andcheckPipelinedGroupsSupportedInRDDGraph/
checkPipelinedProducerSupportedfail-fast (typed exception).TaskSetManager.scala--maxTaskFailures = 1and force-counted executor loss for a pipelinedset; exclusion from the "Resubmitted" re-enqueue loop.
ShuffleMapStage.scala-- monotonic local availability for a pipelined shuffle.PIPELINED_SHUFFLE_UNSUPPORTEDandPIPELINED_SHUFFLE_CROSS_JOB_REUSEerror conditions.Why are the changes needed?
Co-scheduling a pipelined group (#57341) is only safe if failure is handled at the granularity of
the whole group: because the shuffle is transient and once-through, the stock per-stage resubmit
recovery would either deadlock the group or hang a producer's streaming writer. This PR makes any
member failure fail the group atomically (so the caller reruns it) and rejects up front the idioms
whose recovery/semantics are incompatible with a transient, concurrently-read shuffle -- turning
what would be a hang or a silently-wrong schedule into a clear, immediate failure.
Does this PR introduce any user-facing change?
No. All new behavior is gated on a job using a
PipelinedShuffleDependency, which nothing constructsyet, so every existing job is scheduled and recovers exactly as before. The new error conditions
(
PIPELINED_SHUFFLE_UNSUPPORTED,PIPELINED_SHUFFLE_CROSS_JOB_REUSE) can only surface for a jobthat uses a pipelined dependency.
How was this patch tested?
New unit tests in
DAGSchedulerSuite,TaskSetManagerSuite, andShuffleDependencySuitecover:maxTaskFailures = 1aborting a member set on the first failure; executorloss force-counted (and benign
TaskKilled/TaskCommitDeniednot force-counted); a memberFetchFailed aborting the whole group rather than resubmitting a single stage;
producer; a completed pipelined producer's availability surviving executor loss; the "Resubmitted"
loop excluding a pipelined set (including the partial-producer-on-decommission case);
checkpoint in a producer's or a consumer's chain (including downstream in the consumer stage), and
a mixed-resource-profile group -- each rejected up front; and that regular-shuffle idioms are NOT
rejected (inertness of the fail-fast for a job with no pipelined dependency).
The full
DAGSchedulerSuite(193),TaskSchedulerImplSuite+TaskSetManagerSuite(188), andShuffleDependencySuitepass.Was this patch authored or co-authored using generative AI tooling?
coauthored-by: Claude Code (Opus 4.8)