[SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler by jerrypeng · Pull Request #57341 · apache/spark

jerrypeng · 2026-07-17T20:13:19Z

What changes were proposed in this pull request?

This PR adds native DAGScheduler support for concurrently scheduling stages connected by a
PipelinedShuffleDependency (added earlier in the stack), together with the admission and
completion semantics that make co-scheduling correct.

A pipelined shuffle is incrementally readable: a consumer stage may begin reading the producer's
output while the producer is still running. The stock scheduler runs a consumer only after its
producer has fully materialized; this PR teaches the scheduler to co-schedule a producer and its
pipelined consumer as a pipelined group instead. Every new path is gated on a job actually
using a pipelined dependency, so behavior is unchanged for all existing jobs (the existing
DAGSchedulerSuite is unaffected).

This PR supports a job that is all-regular or all-pipelined (a job mixing the two is rejected
up front). So an all-pipelined job's whole stage graph is one pipelined group, which lets admission
be decided once, up front.

Up-front gang admission (handleJobSubmitted). All members of a pipelined group must run
concurrently, so a group that cannot fit would deadlock (a consumer holds slots waiting for
producer output while the producer cannot get slots to produce). Before any stage is created, the
group's total task demand (computed from the RDD graph) is compared against the currently free
slots of its resource profile -- total capacity (maxNumConcurrentTasks) minus the outstanding
(running plus enqueued) task demand of other work in the same profile. Counting enqueued, not
just running, demand prevents two groups from each passing the check yet failing to co-fit. If the
group does not fit, the job fails fast with CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT before any
member runs -- true all-or-nothing admission that leaves no partial scheduler state (like the
barrier slot check). There is no scheduler-side retry: a transient shortfall is the caller's to
retry (a streaming query's batch loop reruns the batch). The check can be disabled with
spark.scheduler.pipelinedGroup.slotCheck.enabled=false for deployments that admit capacity
out-of-band (e.g. a slot reservation).
Co-scheduling (submitStage). A stage's missing parents are classified by shuffle dependency
type; a parent reached through a PipelinedShuffleDependency is a "pipelined parent". The stage is
co-scheduled with its producers (tasks submitted immediately) only if every missing parent is
pipelined and each is already running; otherwise it parks in waitingStages exactly as before.
submitWaitingPipelinedChildStages is the "producer started running" analog of the existing
"producer completed" hook: when a pipelined producer starts, its waiting consumers are reconsidered
immediately, cascading transitively down a chain.
Deferred completion for a co-scheduled consumer (handleTaskCompletion /
markStageAsFinished). A consumer co-scheduled with a still-running producer can finish first.
Advancing its stage/job completion early would end the job and cancel the still-running producer,
or make the consumer's output observable before the producer's. So only the stage/job-completion
decision is deferred: the consumer's per-task side effects -- accumulator updates and the
SparkListenerTaskEnd listener event -- run inline as its tasks finish, exactly as for any other
stage, and only the completion bookkeeping is buffered. It is replayed when the producer's outcome
is final: applied on genuine producer success, dropped on producer failure (the group reruns).
Deferrals are cleaned up on job end/abort so none outlive their job.
Other guards. A job using a pipelined dependency is rejected up front when speculation is
enabled (a speculative producer copy would race a consumer already reading partial output, with no
commit barrier) or when dynamic allocation is enabled (gang admission needs a stable slot set);
and a PipelinedShuffleDependency cannot be submitted as a map-stage job (no durable map output to
compute statistics from).

Main changes:

DAGScheduler.scala -- job classification, up-front gang admission, co-scheduling, and
completion deferral. A one-pass RDD-graph walk (rddGraphHasPipelinedDependency /
classifyJobShuffleKinds) keeps every new path inert for a job with no pipelined dependency.
TaskSchedulerImpl.outstandingTasksForOtherWorkInProfile -- a resource-profile-scoped
outstanding-task (running + enqueued) count for the admission check (private[scheduler]).
spark.scheduler.pipelinedGroup.slotCheck.enabled -- a new internal() config (default true).
CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT -- a new error condition.

This is a follow-up to PipelinedShuffleDependency and the dependency-typed shuffle-manager
routing (SPARK-58185, already in master); it is the first PR that makes the scheduler behave
differently for a pipelined dependency. Group-atomic failure/rerun and additional fail-fast checks
for unsupported idioms follow in later PRs of the stack.

Why are the changes needed?

PipelinedShuffleDependency and its incremental shuffle routing let a consumer read a producer's
output as it is produced, but nothing takes advantage of that until the scheduler co-schedules the
two stages -- otherwise the consumer still waits for the producer to fully materialize and the
pipelining is never realized. Co-scheduling in turn requires admission control (a group that cannot
co-fit must fail fast, not deadlock) and completion control (a fast-finishing consumer must not end
the job or cancel its producer). This PR provides both.

Does this PR introduce any user-facing change?

No. All new behavior is gated on a job using a PipelinedShuffleDependency, which nothing constructs
yet, so for every existing job the scheduler behaves exactly as before. The new
spark.scheduler.pipelinedGroup.slotCheck.enabled config is internal() and defaults to true,
and CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT is a new error condition that can only surface for a job
that uses a pipelined dependency.

How was this patch tested?

New unit tests in DAGSchedulerSuite cover:

concurrent submission of a pipelined producer/consumer; inertness for a regular shuffle; a job
mixing pipelined + regular shuffles rejected up front; a deep all-pipelined chain co-scheduled;
transitive cascade when a producer starts; no double-submission on producer completion;
speculation and dynamic-allocation rejection for a pipelined job (and that a regular job under
speculation is not rejected); a pipelined dependency submitted as a map-stage job rejected;
up-front admission: a group too large to co-fit failing fast with
CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT; a group that fits total capacity but not free slots
failing fast; a group whose demand exactly equals free capacity admitted; other work's outstanding
demand charged against admission; the slot check disabled admitting an over-capacity group;
deferred completion: an early-finishing consumer not ending the job or cancelling its running
producer; its TaskEnd fired in real time (before the producer finishes) and exactly once (no
buffer+replay duplication); normal producer-then-consumer ordering; and the deferral released only
when the producer is genuinely available.

TaskSchedulerImplSuite covers outstandingTasksForOtherWorkInProfile counting running + enqueued
tasks, excluding given stages, being resource-profile-scoped, and not double-counting a
zombie + live attempt.

Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Code (Opus 4.8)

jerrypeng · 2026-07-22T18:43:42Z

-      // Update rdd blocks' visibility status.
-      blockManagerMaster.updateRDDBlockVisibility(
-        event.taskInfo.taskId, visible = event.reason == Success)
+    // Group-observable completion for a pipelined consumer: its per-task side effects (accumulator


This logic here can be greatly simplified if we do eagerly emit task completion events and delay stage / job completion events.

…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

jerrypeng · 2026-07-23T01:02:24Z

@cloud-fan @mridulm please review.

cloud-fan

1 blocking, 0 non-blocking, 0 nits.
The scheduling and completion design is well covered, but the admission check can consult the wrong capacity for jobs using custom resource profiles.

Correctness (1)

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:1557: Use the pipelined stages' actual resource profile for admission, or reject groups that are not single-profile. This currently always measures the default profile, but stage construction derives each stage's profile from its RDDs. A pipelined job using a custom ResourceProfile can therefore pass admission based on unrelated default-profile capacity and then queue or deadlock in the profile where its tasks actually run. Please derive and validate the group profile before comparing demand and occupancy, and add a custom-profile test. -- see inline

Verification

Traced admission from handleJobSubmitted through rejectUnadmittablePipelinedGroup, then compared its fixed default-profile lookup with createShuffleMapStage/createResultStage, which derive stage profiles from RDD resource profiles. Also audited deferred completion ordering, task-side-effect replay guards, task-set occupancy accounting, configuration inversion coverage, and all three scanner manifests.

cloud-fan · 2026-07-23T05:04:16Z

+    if (!sc.conf.get(config.PIPELINED_GROUP_SLOT_CHECK_ENABLED)) {
+      return false
+    }
+    val rp = sc.resourceProfileManager.defaultResourceProfile


Please admit against the pipelined stages' actual resource profile, or reject groups that are not single-profile. Stage construction derives profiles from the RDD graph, so a custom-profile job can pass this default-profile check using unrelated capacity and then queue or deadlock where its tasks actually run. A focused custom-profile test would cover the mismatch.

jerrypeng force-pushed the stack/pipelined-shuffle-pr3-scheduling branch from 9aefc6d to f7a6c5e Compare July 17, 2026 21:54

jerrypeng changed the title ~~[WIP] Stack/pipelined shuffle pr3 scheduling~~ [WIP] [SPARK-XXXXX][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler Jul 17, 2026

jerrypeng mentioned this pull request Jul 20, 2026

[WIP] Stack/pipelined shuffle pr6 failfast #57361

Open

jerrypeng force-pushed the stack/pipelined-shuffle-pr3-scheduling branch 2 times, most recently from 253308e to 4df7ef5 Compare July 22, 2026 18:23

jerrypeng commented Jul 22, 2026

View reviewed changes

jerrypeng changed the title ~~[WIP] [SPARK-XXXXX][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler~~ [WIP] [SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler Jul 22, 2026

jerrypeng changed the title ~~[WIP] [SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler~~ [SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler Jul 22, 2026

jerrypeng added 18 commits July 22, 2026 23:34

jerrypeng force-pushed the stack/pipelined-shuffle-pr3-scheduling branch from 4df7ef5 to 89b29a8 Compare July 22, 2026 23:47

cloud-fan reviewed Jul 23, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler#57341

[SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler#57341
jerrypeng wants to merge 18 commits into
apache:masterfrom
jerrypeng:stack/pipelined-shuffle-pr3-scheduling

jerrypeng commented Jul 17, 2026 •

edited

Loading

Uh oh!

jerrypeng Jul 22, 2026

Uh oh!

jerrypeng commented Jul 23, 2026

Uh oh!

cloud-fan left a comment

Uh oh!

cloud-fan Jul 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

jerrypeng commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

jerrypeng Jul 22, 2026

Choose a reason for hiding this comment

Uh oh!

jerrypeng commented Jul 23, 2026

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Correctness (1)

Verification

Uh oh!

cloud-fan Jul 23, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jerrypeng commented Jul 17, 2026 •

edited

Loading