[WIP] Stack/pipelined shuffle pr6 failfast#57361
Open
jerrypeng wants to merge 41 commits into
Open
Conversation
jerrypeng
force-pushed
the
stack/pipelined-shuffle-pr6-failfast
branch
7 times, most recently
from
July 21, 2026 19:34
6e3149c to
4e8c399
Compare
…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
… check; fix a vacuous admission test
Two test-hygiene fixes from an adversarial review:
- assertDataStructuresEmpty now asserts dependentStageMap.isEmpty, per the
DAGScheduler class-doc checklist ("when adding a new data structure, update
DAGSchedulerSuite.assertDataStructuresEmpty ... to catch memory leaks"). The
new dependentStageMap (the pipelined deferred-completion buffer) was omitted;
it is always drained today, so this passes -- and now guards a future leak of
buffered CompletionEvents / a consumer stage left as perpetually-running.
- Retitle/reframe "the group's own running members are not charged against its
admission": it stubbed the occupancy seam to a constant ignoring
excludeStageIds, and the up-front admission passes Set.empty (no job stage
exists yet), so it never exercised member exclusion -- it only asserts a
demand==free-capacity group is admitted (a real boundary). The member-exclusion
invariant is genuinely covered in TaskSchedulerImplSuite; the title/comment now
say what the test actually verifies.
Co-authored-by: Isaac
…n, run per-task effects inline (S5.1) The group-observable completion for a pipelined consumer (spec S5) previously used a coarse model: when a consumer finished ahead of its still-running producer, the WHOLE CompletionEvent was buffered and its side effects were withheld until replay. That contradicts the negotiated spec S5.1, which requires per-task side effects -- accumulator updates and the SparkListenerTaskEnd listener event -- to flow in real time, deferring ONLY the stage/job-completion decision (advancing that early is what would cancel the still-running producer or expose the consumer's output prematurely). Switch to the fine-grained model: - handleTaskCompletion runs a deferred consumer's per-task effects inline (as for any other stage), then buffers the event and returns before the completion bookkeeping. - A new CompletionEvent.finishOnly flag marks a replayed event; on replay only the completion bookkeeping runs, so the per-task effects are never applied twice. - releaseDeferredPipelinedConsumers replays with finishOnly=true; the producer-failure drop path simply discards the buffered events (their TaskEnd already fired inline), removing the old asymmetry where the drop path re-emitted postTaskEnd but not updateAccumulators. Observability improves: a finished consumer's tasks are no longer reported as running for the producer's remaining lifetime, and no late TaskEnd burst appears at group abort. Tests: the two deferral tests now assert per-task TaskEnd fires in real time (before the producer finishes) and exactly once. Reverting the fix makes them fail (TaskEnd count 0 instead of 2 while the producer runs). Full DAGSchedulerSuite green. Co-authored-by: Isaac
…reads as unused" and a phantom symbol The NOTE on isPipelinedGroupMember said it "reads as unused when this change is viewed alone" and listed `pipelinedGroupOf` as a sibling helper. Both are inaccurate in the merged view: the method IS used (its call sites -- TaskSet.isPipelined tagging and the member-FetchFailed group abort -- are part of this stack), and `pipelinedGroupOf` does not exist. Reword to name the actual call sites and drop the phantom reference. Co-authored-by: Isaac
…private[scheduler] The method exists only to back the pipelined-group slot admission check in DAGScheduler; it does not belong on the public TaskSchedulerImpl surface. Its only callers (DAGScheduler and TaskSchedulerImplSuite) are in org.apache.spark.scheduler, so private[scheduler] is sufficient and keeps the new capacity-accounting method out of the public API. Co-authored-by: Isaac
… the retry behavior The gang-admission scaladoc likened the check to barrier's slot check. The likeness is only that both reject before any stage is created and compute demand from the RDD graph; their retry behavior differs and a reader should not infer equivalence. Spell out the contrast: barrier re-posts the job and retries its check up to a max-failures bound, whereas pipelined admission is terminal (one check, then fail) and delegates transient-shortfall retry to the caller -- scheduler-side PG-admission retry being a post-v1 refinement (spec S4.1). 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
…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
…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 22, 2026 18:23
4e8c399 to
7cecf65
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)