From 228689273fef21e53ce20793ed1b23a20af3a6c9 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Tue, 7 Jul 2026 20:57:00 +0000
Subject: [PATCH 01/10] [SPARK-XXXXX][DOC] Pipelined shuffle dependency &
 concurrent stage scheduling spec

Baseline of the design spec for native concurrent stage scheduling in the
DAGScheduler, as reviewed on PR apache/spark#56055. Subsequent commits layer
reviewer feedback one point at a time for a clean per-round diff.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 209 +++++++++++++++++++++++++++
 1 file changed, 209 insertions(+)
 create mode 100644 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
new file mode 100644
index 000000000000..b3e47bda8d33
--- /dev/null
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -0,0 +1,209 @@
+# Pipelined Shuffle Dependency & Concurrent Stage Scheduling
+
+A spec for running data-dependent stages of a single job concurrently, connected by a shuffle the
+consumer reads incrementally.
+
+---
+
+## 1. Motivation
+
+Today a multi-stage job runs one stage at a time: each shuffle is fully materialized before the next
+stage starts. Some workloads need the stages of a single job to run **concurrently**, connected by a
+shuffle whose consumer reads the producer's output **as it is produced** rather than after the
+producer finishes. This spec introduces the scheduler primitives to express and run that.
+
+"Run these stages concurrently" and "the connecting shuffle is incremental" are the same decision
+seen from two sides: co-scheduling a producer and consumer is only useful if the edge is readable
+before the producer completes.
+
+---
+
+## 2. Primitives
+
+### 2.1 Pipelined shuffle dependency (PSD)
+
+A shuffle dependency declared **incrementally readable**: a consumer stage may begin reading its
+output while the producer stage is still running.
+
+- It is a shuffle dependency (has a `shuffleId`, partitioner, map/reduce sides); the *pipelined*
+  property is a binding part of the scheduler contract, not an advisory hint.
+- The property is set during **physical planning** (an execution concern, not a logical-plan one)
+  and carried into the `ShuffleDependency` the `DAGScheduler` reads at stage-creation time.
+
+A **regular shuffle dependency (RSD)** is an ordinary shuffle dependency: its output must be fully
+materialized before any consumer reads it.
+
+### 2.2 Pipelined group (G)
+
+The set of stages connected to one another through pipelined edges — the connected component of the
+stage DAG when only pipelined edges are considered.
+
+- A stage with no incident pipelined edge is a **singleton group** and behaves exactly as a normal
+  stage today.
+- The group — not the edge or the individual stage — is the unit of **admission**, **slot
+  checking**, **completion**, and **failure**.
+
+**External input of G:** a regular shuffle dependency whose consumer is in G and whose producer is
+not — i.e. a normal materialized parent of the group.
+
+---
+
+## 3. Group formation
+
+- **Stage decomposition is unchanged.** A pipelined dependency introduces a shuffle boundary exactly
+  as a regular one does; the set of stages and their partitioning are identical. The pipelined
+  property changes only *when* stages run relative to one another, never *how the plan is cut into
+  stages*.
+- **Group = connected component over pipelined edges.** As stages are created, two stages joined by a
+  pipelined edge are placed in the same group; the group is the transitive closure.
+- **Every stage belongs to exactly one group** (singletons included). Group membership is fixed at
+  stage-creation time.
+
+---
+
+## 4. Scheduling & admission
+
+- **A pipelined edge is non-sequencing.** The consumer of a pipelined dependency does not wait for
+  its producer to materialize. (A regular-dependency consumer still waits — the default behavior.)
+- **Group readiness.** A group is ready to be admitted when every external input of the group (its
+  regular materialized parents) is available — the same precondition a normal stage has today, lifted
+  to the group. Pipelined parents inside the group impose no readiness precondition.
+- **Gang admission (all-or-nothing).** A group is admitted only if the cluster can currently run all
+  tasks of all member stages **concurrently**; admission then submits every member stage at once.
+  There is no partial admission — a group is never left with some members running while others wait
+  on slots the running members occupy.
+- **Slot check.** The group's aggregate concurrent-task demand — the sum of `numTasks` over member
+  stages — is compared against the number of available slots in the cluster (a slot is one task's
+  worth of capacity, so this is the maximum number of tasks that can run at once). If the group
+  needs more slots than the cluster can offer, the submission fails fast, since the group could
+  never become fully co-resident.
+- **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
+- **Single ownership.** A pipelined group belongs to exactly one job; a pipelined producer is not
+  shared across jobs.
+
+---
+
+## 5. Completion
+
+- **Group-observable completion.** A member stage is not job-observably finished until **all** member
+  stages of its group have completed successfully. A result stage's job therefore completes only when
+  its entire group has, and a member that finishes ahead of the others cannot advance job or stage
+  completion until the group does.
+- **Defer the finish decision, not per-task work.** When a member finishes before its group, only its
+  stage-finish / job-finish transition is deferred until group completion. Per-task side effects that
+  must always run — accumulator updates, output-commit coordination, task-end listener events — run
+  immediately.
+- **Replay window.** There is a window between a member finishing and group completion. A failure in
+  that window is a group failure (§6): the deferred finish transitions are dropped and the group
+  reruns.
+
+### 5.1 Observable completion events (listener bus)
+
+The listener bus is an external contract that monitoring tools depend on, so it is worth stating
+exactly when each event is delivered. The rule follows directly from atomic commit: **task-level
+events flow in real time, but stage-completion and job-completion events are held until the group
+commits — so a listener never observes a member as *successfully completed* before the group as a
+whole has.**
+
+| Event | Timing | Rationale |
+|-------|--------|-----------|
+| `SparkListenerTaskStart` / `SparkListenerTaskEnd` | Real time, as they occur | Per-task facts are true when they happen; a group's members genuinely run concurrently. Deferring these would freeze a member's live progress and metrics for the whole group's duration. Note a successful `TaskEnd` means "this task finished," not "its output is committed" — already true in Spark, since a stage attempt can later be discarded. |
+| `SparkListenerStageSubmitted` | Real time, at group admission | All member stages are submitted together (§4); a monitor should show them active simultaneously. |
+| `SparkListenerStageCompleted` | Deferred to group commit; on group failure, emitted with a failure reason | "Completed" should track commit, which is atomic at the group level. A member whose tasks finish early is reported as still running until the group commits — which matches the truth that its results are not usable until then. This avoids emitting a success-shaped completion for an attempt that a later group failure would discard. |
+| `SparkListenerJobEnd(JobSucceeded)` | At group commit only | Job completion delivers results to the caller and cancels sibling stages; emitting it before the group commits risks double/inconsistent result delivery if the group later fails. Non-negotiable. |
+| `SparkListenerJobEnd(JobFailed)` | On group failure | Group-atomic failure (§6): buffered success transitions are dropped, never replayed as success. |
+
+Failure-path consequence: because stage- and job-completion are deferred (never emitted early as
+success), a group failure needs only to emit them as *failure* — there is no premature success event
+to retract. Already-emitted task events stand as-is (they were true), consistent with how Spark
+treats task events from a stage attempt that is later discarded.
+
+Open question for reviewers: whether `SparkListenerStageCompleted` for a member should be deferred to
+group commit (above) or emitted in real time when the member's tasks finish. Deferring keeps the
+listener's notion of "completed" consistent with the atomic-commit model at the cost of an inflated
+stage duration on the timeline; real-time emission gives crisper per-stage timings but reports a
+member as completed before the group has committed.
+
+---
+
+## 6. Failure
+
+- **Failure is group-atomic.** Any task failure in any member stage — including a fetch failure —
+  fails the whole group. There is no independent single-stage failure within a group. Two mechanisms
+  make this hold:
+  - *No single-stage resubmit inside a group.* The ordinary "resubmit one failed stage" path is
+    invalid for a group member: the transient pipelined shuffle cannot be re-read, and members are
+    co-scheduled. A member failure routes to group failure instead.
+  - *Teardown is by group membership, not producer availability.* At the end of a batch the producer
+    finishes and registers all its map outputs — becoming "available" — while the consumer is still
+    draining the remainder, so a producer-available-while-consumer-still-running window is normal,
+    not rare. For a transient pipelined shuffle, "available" does not mean the consumer is safe: the
+    data is not durably stored and the consumer is still actively reading it, so a producer failure
+    in that window still strands the consumer. Failure propagation that keys off producer
+    availability would miss the consumer here; teardown is therefore keyed on group membership, so a
+    producer failure always tears down its co-scheduled consumers regardless of the producer's
+    availability at the failure instant.
+- **Recovery.** Group failure tears down all member stages atomically and discards their deferred
+  completions. Because the shuffle is transient it cannot be partially recovered; the job fails and
+  the caller re-runs it.
+
+---
+
+## 7. Interaction with regular dependencies
+
+- **Regular shuffle into a group — supported.** A regular dependency from outside the group to a
+  member is a normal materialized parent (an external input); the group waits for it.
+- **Regular shuffle out of a group — supported.** A regular dependency from a member to a stage
+  outside the group produces a materialized output that a downstream group consumes with normal
+  sequencing; the downstream waits for the group to complete.
+- **Regular shuffle internal to a group — fail-fast / unsupported.** If a regular dependency's
+  producer and consumer both fall in the same group, the group would co-schedule them concurrently
+  while the regular edge demands the producer be materialized first — a contradiction. Reject at
+  stage creation.
+- Pipelined edges are intra-group by construction, so they never cross a group boundary.
+
+---
+
+## 8. Activation
+
+- **Activation is by group membership.** Every behavioral change — admission, completion, and
+  failure handling — keys off a single question: is this stage a member of a pipelined group? There
+  is no separate enable flag and no half-activated state.
+- **Opt-in is expressed by the presence of a pipelined dependency**, which a physical-planning rule
+  sets on the relevant exchanges. The scheduler primitive is generic; any feature can write such a
+  rule to opt a job into concurrent stage scheduling over an incremental shuffle.
+
+---
+
+## 9. Fail-fast on unsupported idioms
+
+A pipelined group that involves any of the following is rejected at stage/group creation with a clear
+error, rather than silently mis-scheduled. Each is a documented limitation of the first version.
+
+The rejected idioms fall into two kinds. Some are **moot** under the group failure model (§6):
+Spark's stage-level recompute/rollback mechanisms never fire inside a group, because any failure
+aborts the whole group and the caller reruns from scratch — so a mechanism whose only job is to
+recompute or roll back a stage after a partial failure is never reached. We reject these rather than
+carry dead, self-contradictory machinery. The rest are **incompatible** with concurrent execution
+itself and would corrupt or hang a group that never fails.
+
+| Rejected condition | Kind | Why rejected |
+|--------------------|------|--------------|
+| Barrier execution in a member stage | incompatible | Barrier exposes output only after a global sync, contradicting concurrent partial reads. |
+| Dynamic resource allocation | incompatible | Gang admission needs a stable slot set; reclaiming executors from a pinned-open group can deadlock it. |
+| Speculative execution | incompatible | A speculative producer copy races a consumer already reading partial output; no commit barrier protects the read. |
+| Push-based shuffle merge | incompatible | Push-based shuffle merges each partition's blocks into large pre-sealed files on merger services, and those files become readable only after a post-map-completion "finalize" step — the exact opposite of a pipelined shuffle, whose consumer reads before the producer finishes, so the two can never apply to the same edge. |
+| Statically-indeterminate producer | moot | Its recovery is stage rollback-and-recompute, which a group never performs (§6: any failure aborts the group). The mechanism is never reached; rejected so the inapplicability is explicit rather than latent. (A producer whose output RDD is classified `INDETERMINATE` — a rerun can yield different data, not just a different order — determinable from the RDD graph at stage-creation time.) |
+| Checksum-mismatch full retry | moot | The runtime counterpart to static indeterminism: it checksums each map task's output and, on a cross-attempt mismatch, rolls back and re-runs the succeeding stages. A group never keeps succeeding stages across a retry (§6), so this never fires; rejected to keep the inapplicability explicit. |
+| Cached/persisted RDD in a member's within-stage chain | incompatible | Would capture partial output read mid-run and serve it as a complete result on reuse. Scoped to the member's within-stage (narrow-dependency) chain: a cached *complete* input reached via a broadcast variable or across a materialized shuffle — e.g. the static side of a stream-static join — is outside that chain and is unaffected. |
+| Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. |
+| Regular shuffle internal to a group | incompatible | Concurrent co-scheduling contradicts materialize-before-read (§7). |
+| Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
+
+The AQE row forbids AQE's *intra-batch, mid-stage* replanning — reading a running producer's statistics
+within one execution and reshaping its consumers — not statistics-driven adaptivity in general. A
+streaming consumer may still reshape a later batch's plan from an earlier batch's completed statistics:
+those statistics are final (the earlier batch fully materialized), and the reshaping happens at planning
+time, before the later batch's stages are co-scheduled — so it composes with pipelining rather than
+contradicting it. That cross-batch feedback is the compatible way to get AQE-like benefits (partition
+coalescing, skew handling, join-strategy selection) in a pipelined streaming query.

From 177a57e68eb4701b08a3d9fd33db071387bafb60 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Tue, 7 Jul 2026 23:01:10 +0000
Subject: [PATCH 02/10] [SPARK-XXXXX][DOC] Address review round 1 (cloud-fan):
 admission, commit, failure framing

Layer the first round of reviewer feedback on PR apache/spark#56055 onto the
baseline spec:

- Cross-group admission and starvation (new 4.1): capacity measured against free
  slots; a group that does not fit fails admission rather than queuing (no FIFO,
  no partial reservation); retry is the caller's decision, not the scheduler's.
- Slot check: reuse maxNumConcurrentTasks (not defaultParallelism); require a
  single resource profile per group, reject mixed (fail-fast).
- Completion: in-group result-stage side effects must be idempotent (the
  streaming at-least-once model); deferring commit noted as a future augmentation.
- Failure framing: state it as a mechanism -- single-stage resubmit is disabled
  for group members, so any member failure fails the group -- rather than
  enumerating a FetchFailed path that does not exist on a streaming edge; note the
  coarser-grained recovery vs. base Spark.
- Naming note: "pipelined" is deliberate, not "streaming"; the capability is
  general and streaming is only the first caller.
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 63 +++++++++++++++++++++++++---
 1 file changed, 57 insertions(+), 6 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index b3e47bda8d33..c999ee9ec91d 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -33,6 +33,12 @@ output while the producer stage is still running.
 A **regular shuffle dependency (RSD)** is an ordinary shuffle dependency: its output must be fully
 materialized before any consumer reads it.
 
+Note: the name *pipelined* is deliberately chosen over *streaming*. The property is that a consumer
+reads producer output as it is produced — software-pipelining of dependent stages — which is a
+general execution capability. Streaming / real-time mode is the first caller, but nothing about the
+primitive is streaming-specific, so the dependency and the group are named for the capability, not
+the caller.
+
 ### 2.2 Pipelined group (G)
 
 The set of stages connected to one another through pipelined edges — the connected component of the
@@ -77,10 +83,43 @@ not — i.e. a normal materialized parent of the group.
   worth of capacity, so this is the maximum number of tasks that can run at once). If the group
   needs more slots than the cluster can offer, the submission fails fast, since the group could
   never become fully co-resident.
+  - *What "available slots" is:* the cluster's concurrent-task capacity — the number of tasks it can
+    run at once — reusing the value the barrier slot check already uses (`sc.maxNumConcurrentTasks`),
+    not `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run
+    concurrently).
+- **Single resource profile per group (v1).** A resource profile is the executor/task resource
+  requirement (cores, memory, GPUs, ...) a stage runs under; the number of concurrent slots is
+  defined *per profile* (a cluster may run many concurrent CPU tasks but few GPU tasks). Comparing
+  one demand against one capacity is therefore only well-defined when all member stages of a group
+  share a single profile. v1 requires that and rejects a mixed-profile group (fail-fast, §9);
+  per-profile accounting — checking each profile's demand against that profile's own capacity — is a
+  follow-up, not needed for the streaming shapes whose members share the default profile.
 - **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
 - **Single ownership.** A pipelined group belongs to exactly one job; a pipelined producer is not
   shared across jobs.
 
+### 4.1 Cross-group admission (multiple groups / groups vs. regular jobs)
+
+A pipelined group holds its slots for its whole run, so admission is decided against currently-free
+slots, and a group that does not fit is failed rather than queued.
+
+- **Capacity is free slots, not total.** The slot check counts only slots not occupied by running
+  tasks — running groups' and regular jobs' tasks are subtracted — so a group is admitted only if its
+  full demand fits in the slots free at admission time.
+  - *How free capacity is measured:* the cluster's concurrent-task capacity (the number of tasks it
+    can run at once — what an all-at-once gang admission must check against, see §4) minus the tasks
+    currently running.
+- **No waiting queue, no partial reservation.** A group that doesn't fit fails its admission; it never
+  sits in a queue holding slots incrementally. This keeps the scheduler change minimal and cannot
+  deadlock (a group never occupies slots a sibling is blocked on), matching barrier, which also fails
+  its slot check rather than queuing.
+- **Two failure reasons, one path.** Demand > total capacity can never fit (a plan/sizing error);
+  demand > free slots but ≤ total could fit later. Both fail admission.
+- **Retry is the caller's decision.** The scheduler does not automatically retry a failed admission —
+  it fails the job. Whether and how to retry is up to the caller. For a streaming query this is the
+  batch-execution loop restarting the batch (with its own backoff), so a transiently-busy cluster is
+  handled by the caller's restart policy, not a scheduler retry loop.
+
 ---
 
 ## 5. Completion
@@ -96,6 +135,13 @@ not — i.e. a normal materialized parent of the group.
 - **Replay window.** There is a window between a member finishing and group completion. A failure in
   that window is a group failure (§6): the deferred finish transitions are dropped and the group
   reruns.
+- **In-group result-stage side effects must be idempotent.** Per-task side effects run immediately
+  (above), including a result stage's output commit. If a result task commits and a sibling then
+  fails in the replay window, the group reruns and re-delivers that output. This is the standard
+  streaming model — a batch is re-delivered on recovery and the sink must absorb it — so v1 requires
+  an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink that cannot
+  absorb re-delivery (§9). Deferring the commit itself to group completion is a possible future
+  augmentation if a non-idempotent committer ever needs to be supported.
 
 ### 5.1 Observable completion events (listener bus)
 
@@ -128,12 +174,15 @@ member as completed before the group has committed.
 
 ## 6. Failure
 
-- **Failure is group-atomic.** Any task failure in any member stage — including a fetch failure —
-  fails the whole group. There is no independent single-stage failure within a group. Two mechanisms
-  make this hold:
-  - *No single-stage resubmit inside a group.* The ordinary "resubmit one failed stage" path is
-    invalid for a group member: the transient pipelined shuffle cannot be re-read, and members are
-    co-scheduled. A member failure routes to group failure instead.
+- **Failure is group-atomic.** Any member task failure, for any reason, fails the whole group; there
+  is no independent single-stage failure within a group.
+  - *Single-stage resubmit is disabled for group members.* That isolated-resubmit branch is turned
+    off for a member of a pipelined group: the transient pipelined shuffle cannot be re-read and
+    members are co-scheduled, so a lone-stage resubmit is never valid. With it disabled, every member
+    failure necessarily routes to group failure.
+    - *Note this differs from the base scheduler,* which handles a task failure by resubmitting just
+      that one stage — the `FetchFailed` -> resubmit path, and retry paths generally — recomputing
+      the failed stage in isolation and resuming.
   - *Teardown is by group membership, not producer availability.* At the end of a batch the producer
     finishes and registers all its map outputs — becoming "available" — while the consumer is still
     draining the remainder, so a producer-available-while-consumer-still-running window is normal,
@@ -153,6 +202,7 @@ member as completed before the group has committed.
 
 - **Regular shuffle into a group — supported.** A regular dependency from outside the group to a
   member is a normal materialized parent (an external input); the group waits for it.
+  - A lost external durable input reruns the group rather than being recomputed in isolation.
 - **Regular shuffle out of a group — supported.** A regular dependency from a member to a stage
   outside the group produces a materialized output that a downstream group consumes with normal
   sequencing; the downstream waits for the group to complete.
@@ -198,6 +248,7 @@ itself and would corrupt or hang a group that never fails.
 | Cached/persisted RDD in a member's within-stage chain | incompatible | Would capture partial output read mid-run and serve it as a complete result on reuse. Scoped to the member's within-stage (narrow-dependency) chain: a cached *complete* input reached via a broadcast variable or across a materialized shuffle — e.g. the static side of a stream-static join — is outside that chain and is unaffected. |
 | Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. |
 | Regular shuffle internal to a group | incompatible | Concurrent co-scheduling contradicts materialize-before-read (§7). |
+| Members with differing resource profiles | incompatible | The gang slot check compares one demand against one capacity (§4); v1 requires a group to be single-profile. Per-profile accounting is a follow-up. |
 | Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
 
 The AQE row forbids AQE's *intra-batch, mid-stage* replanning — reading a running producer's statistics

From 18152894cf4ed119ad4f806ec7a214e346cc46be Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Tue, 7 Jul 2026 23:39:29 +0000
Subject: [PATCH 03/10] [SPARK-XXXXX][DOC] Address review round 2 (tgravescs):
 gang-admission scoping, per-dependency shuffle manager

- Gang admission (4): scope the all-or-nothing slot requirement to a pipelined
  group explicitly; add a sub-bullet stating a non-pipelined (singleton) group is
  unaffected and runs as a normal stage does today. Clarifies that gang scheduling
  applies only to a group of two or more stages connected by pipelined edges.
- Pipelined shuffle dependency (2.1): note the property is also the per-dependency
  selector for the shuffle implementation -- the shuffle layer maps a pipelined
  dependency to an incremental ShuffleManager and everything else to the default,
  so a mixed job routes each group to the right implementation per-dependency, not
  per-job. Add a config example (spark.shuffle.manager.incremental).
- Minor: drop the "distinguished in diagnostics" clause from the two-failure-reasons
  bullet (4.1).

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 20 ++++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index c999ee9ec91d..8a714c6c90d8 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -29,6 +29,14 @@ output while the producer stage is still running.
   property is a binding part of the scheduler contract, not an advisory hint.
 - The property is set during **physical planning** (an execution concern, not a logical-plan one)
   and carried into the `ShuffleDependency` the `DAGScheduler` reads at stage-creation time.
+- It is also the **per-dependency selector** for the shuffle implementation: the shuffle layer maps a
+  pipelined dependency to an incremental `ShuffleManager` and everything else to the default, so one
+  job with both regular and pipelined groups uses the right implementation for each — selected
+  per-dependency, not per-job. The scheduler construct stays generic; the shuffle implementation
+  stays pluggable.
+  - For example, an additional conf (e.g. `spark.shuffle.manager.incremental`) can be introduced to
+    specify the incremental shuffle implementation used for pipelined shuffle dependencies, alongside
+    the existing `spark.shuffle.manager` for the default.
 
 A **regular shuffle dependency (RSD)** is an ordinary shuffle dependency: its output must be fully
 materialized before any consumer reads it.
@@ -74,10 +82,14 @@ not — i.e. a normal materialized parent of the group.
 - **Group readiness.** A group is ready to be admitted when every external input of the group (its
   regular materialized parents) is available — the same precondition a normal stage has today, lifted
   to the group. Pipelined parents inside the group impose no readiness precondition.
-- **Gang admission (all-or-nothing).** A group is admitted only if the cluster can currently run all
-  tasks of all member stages **concurrently**; admission then submits every member stage at once.
-  There is no partial admission — a group is never left with some members running while others wait
-  on slots the running members occupy.
+- **Gang admission (all-or-nothing).** A pipelined group is admitted only if the cluster can currently
+  run all tasks of all member stages **concurrently**; admission then submits every member stage at
+  once. There is no partial admission — a pipelined group is never left with some members running
+  while others wait on slots the running members occupy.
+  - *A non-pipelined (singleton) group is unaffected:* it is admitted exactly as a normal stage is
+    today — submitted when its parents are available, filling slots incrementally, with no all-at-once
+    requirement. Gang admission applies only to a group of two or more stages connected by pipelined
+    edges.
 - **Slot check.** The group's aggregate concurrent-task demand — the sum of `numTasks` over member
   stages — is compared against the number of available slots in the cluster (a slot is one task's
   worth of capacity, so this is the maximum number of tasks that can run at once). If the group

From 753ca0103986e618d949f1a34cd60c391a0862ed Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Wed, 8 Jul 2026 20:40:25 +0000
Subject: [PATCH 04/10] addressing @cloud-fan comments

---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 33 ++++++++++++++--------------
 1 file changed, 17 insertions(+), 16 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 8a714c6c90d8..78abbd0f8c18 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -91,14 +91,16 @@ not — i.e. a normal materialized parent of the group.
     requirement. Gang admission applies only to a group of two or more stages connected by pipelined
     edges.
 - **Slot check.** The group's aggregate concurrent-task demand — the sum of `numTasks` over member
-  stages — is compared against the number of available slots in the cluster (a slot is one task's
-  worth of capacity, so this is the maximum number of tasks that can run at once). If the group
-  needs more slots than the cluster can offer, the submission fails fast, since the group could
-  never become fully co-resident.
-  - *What "available slots" is:* the cluster's concurrent-task capacity — the number of tasks it can
-    run at once — reusing the value the barrier slot check already uses (`sc.maxNumConcurrentTasks`),
-    not `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run
-    concurrently).
+  stages — is compared against the number of **currently-free** slots: the cluster's total
+  concurrent-task capacity minus the tasks already running. If the group needs more than are free,
+  admission fails fast, since the group could never become fully co-resident. (Free rather than
+  total is essential once more than one group can be admitted; §4.1 explains why.)
+  - *How capacity is measured:* the total concurrent-task capacity is the value barrier's slot check
+    uses (`sc.maxNumConcurrentTasks` — task slots summed over active executors), not
+    `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run
+    concurrently). Note this diverges from barrier in one way: barrier compares demand against that
+    *total*, whereas a pipelined group compares against *free* capacity (total minus running tasks).
+    Computing free capacity is therefore a new computation, not barrier's check reused verbatim.
 - **Single resource profile per group (v1).** A resource profile is the executor/task resource
   requirement (cores, memory, GPUs, ...) a stage runs under; the number of concurrent slots is
   defined *per profile* (a cluster may run many concurrent CPU tasks but few GPU tasks). Comparing
@@ -115,16 +117,15 @@ not — i.e. a normal materialized parent of the group.
 A pipelined group holds its slots for its whole run, so admission is decided against currently-free
 slots, and a group that does not fit is failed rather than queued.
 
-- **Capacity is free slots, not total.** The slot check counts only slots not occupied by running
-  tasks — running groups' and regular jobs' tasks are subtracted — so a group is admitted only if its
-  full demand fits in the slots free at admission time.
-  - *How free capacity is measured:* the cluster's concurrent-task capacity (the number of tasks it
-    can run at once — what an all-at-once gang admission must check against, see §4) minus the tasks
-    currently running.
+- **Why free slots, not total** (the slot check itself is defined in §4). Free capacity subtracts
+  the tasks already running — for every running group and regular job — so a group is admitted only
+  if its full demand fits in what the others leave free. This is what makes co-residency safe:
+  comparing against *total* capacity would let two groups each pass the check yet fail to co-fit,
+  which is exactly the partial co-residency gang admission forbids. (This is why the §4 slot check
+  diverges from barrier's total-capacity check.)
 - **No waiting queue, no partial reservation.** A group that doesn't fit fails its admission; it never
   sits in a queue holding slots incrementally. This keeps the scheduler change minimal and cannot
-  deadlock (a group never occupies slots a sibling is blocked on), matching barrier, which also fails
-  its slot check rather than queuing.
+  deadlock (a group never occupies slots a sibling is blocked on).
 - **Two failure reasons, one path.** Demand > total capacity can never fit (a plan/sizing error);
   demand > free slots but ≤ total could fit later. Both fail admission.
 - **Retry is the caller's decision.** The scheduler does not automatically retry a failed admission —

From d43238c389effd3eb5e24960f4a822e3ece1da27 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Thu, 9 Jul 2026 06:30:20 +0000
Subject: [PATCH 05/10] [SPARK-XXXXX][DOC] Address review round 3 (mridulm):
 multi-output/multi-group PGs, fan-out ownership, commit path, invariants

Addresses @mridulm's review comments on the spec:

- PG outputs (S2.2, S5, S7): a group need not contain a ResultStage; it may
  have multiple materialized output edges feeding downstream groups, like a
  barrier stage embedded in a larger DAG.
- Multiple concurrent PGs per DAG (S3): disjoint pipelined-edge sets form
  distinct groups that may run concurrently, subject to S4.1 slot arbitration.
- Single ownership / 1:1 fan-out (S4, S9): enforced explicitly by never reusing
  a pipelined ShuffleDependency (bypass shuffleIdToMapStage), not by relying on
  streaming re-planning. Reuse must be prevented explicitly; from the
  scheduler's view a shuffle-map stage can be reused across jobs unless forbidden.
- Output-commit (S5): the commit path is unchanged from base Spark --
  OutputCommitCoordinator arbitration never contends inside a group (speculation
  rejected, single-stage resubmit disabled), so each partition has one attempt.
  The only integration point is clearing stale commit-coordinator state on group
  teardown.
- Failure refinement (S6): post-v1, use a consumer-driven FetchFailed signal so a
  post-consumption producer loss does not force a rerun; on a pipelined edge it
  routes to group failure, not single-stage resubmit.
- Internal regular shuffle (S7, S9): reclassified from fail-fast to an invariant
  that should hold by construction (S3) but is still checked defensively.
- Push-based merge (S9): clarified it is rejected only as the incremental
  (pipelined) shuffle; a PG's regular input/output edges may still use it.
- S9 scoping: all rejected idioms are scoped within a PG; a group's regular
  input/output edges are ordinary shuffles and unaffected.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 94 +++++++++++++++++++++-------
 1 file changed, 72 insertions(+), 22 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 78abbd0f8c18..8c5cf74497c3 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -60,6 +60,13 @@ stage DAG when only pipelined edges are considered.
 **External input of G:** a regular shuffle dependency whose consumer is in G and whose producer is
 not — i.e. a normal materialized parent of the group.
 
+**Outputs of G:** a pipelined group need not contain a `ResultStage`. Its outputs to the rest of the
+DAG are regular (materialized) shuffle edges from its members to consumers outside the group, and
+there may be **more than one** — e.g. two different members each feeding a distinct downstream
+stage. A `ResultStage` is one possible member, not a requirement; a PG may instead be an interior
+component whose materialized outputs feed downstream groups (§7). In this respect a PG behaves like
+a barrier stage embedded in a larger DAG: an interior unit with regular-shuffle edges in and out.
+
 ---
 
 ## 3. Group formation
@@ -72,6 +79,12 @@ not — i.e. a normal materialized parent of the group.
   pipelined edge are placed in the same group; the group is the transitive closure.
 - **Every stage belongs to exactly one group** (singletons included). Group membership is fixed at
   stage-creation time.
+- **A DAG may contain multiple pipelined groups.** Disjoint sets of pipelined edges form distinct
+  groups, and a group's materialized output may feed another group (§7). Independent groups (no
+  dependency path between them) may be admitted and run concurrently, subject to the cross-group
+  slot arbitration in §4.1; a group that consumes another group's materialized output waits for it,
+  by the normal materialize-before-read sequencing on that regular edge. This mirrors how a DAG may
+  contain multiple barrier stages.
 
 ---
 
@@ -109,8 +122,15 @@ not — i.e. a normal materialized parent of the group.
   per-profile accounting — checking each profile's demand against that profile's own capacity — is a
   follow-up, not needed for the streaming shapes whose members share the default profile.
 - **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
-- **Single ownership.** A pipelined group belongs to exactly one job; a pipelined producer is not
-  shared across jobs.
+- **Single ownership and 1:1 (v1), enforced by never reusing a pipelined dependency.** A pipelined
+  producer feeds exactly one consumer and belongs to exactly one job. The regular-shuffle 1:many
+  fan-out and cross-job stage reuse are both disallowed, and both are prevented by the same
+  mechanism: a pipelined `ShuffleDependency` is never reused or shared. At stage/group creation the
+  `shuffleIdToMapStage` reuse path is bypassed for a pipelined dependency (each gets a fresh
+  producer stage, never bound to an existing one), and a group is rejected (fail-fast, §9) if a
+  pipelined producer has more than one consuming stage or its stage carries more than one jobId.
+  Reuse must be prevented explicitly: from the scheduler's perspective a shuffle-map stage *can* be
+  reused across jobs unless something forbids it.
 
 ### 4.1 Cross-group admission (multiple groups / groups vs. regular jobs)
 
@@ -137,10 +157,11 @@ slots, and a group that does not fit is failed rather than queued.
 
 ## 5. Completion
 
-- **Group-observable completion.** A member stage is not job-observably finished until **all** member
-  stages of its group have completed successfully. A result stage's job therefore completes only when
-  its entire group has, and a member that finishes ahead of the others cannot advance job or stage
-  completion until the group does.
+- **Group-observable completion.** A member stage is not observably finished until **all** member
+  stages of its group have completed successfully — a member that finishes ahead of the others
+  cannot advance stage or job completion, nor make its output visible to a downstream consumer,
+  until the group does. Whether the group's output is a `ResultStage`'s job result or a materialized
+  shuffle feeding a downstream group (§2.2), that output becomes observable only at group completion.
 - **Defer the finish decision, not per-task work.** When a member finishes before its group, only its
   stage-finish / job-finish transition is deferred until group completion. Per-task side effects that
   must always run — accumulator updates, output-commit coordination, task-end listener events — run
@@ -149,12 +170,19 @@ slots, and a group that does not fit is failed rather than queued.
   that window is a group failure (§6): the deferred finish transitions are dropped and the group
   reruns.
 - **In-group result-stage side effects must be idempotent.** Per-task side effects run immediately
-  (above), including a result stage's output commit. If a result task commits and a sibling then
-  fails in the replay window, the group reruns and re-delivers that output. This is the standard
-  streaming model — a batch is re-delivered on recovery and the sink must absorb it — so v1 requires
-  an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink that cannot
-  absorb re-delivery (§9). Deferring the commit itself to group completion is a possible future
-  augmentation if a non-idempotent committer ever needs to be supported.
+  (above), including a result stage's output commit. The output-commit path itself is unchanged from
+  base Spark: `OutputCommitCoordinator` arbitration — deciding which of several racing attempts of a
+  partition may commit — never contends inside a group, because both sources of concurrent attempts
+  are absent (speculation is rejected, §9; single-stage resubmit is disabled for members, §6), so
+  each partition has exactly one attempt and is trivially authorized. If a result task commits and a
+  sibling then fails in the replay window, the group reruns and re-delivers that output. This is the
+  standard streaming model — a batch is re-delivered on recovery and the sink must absorb it — so v1
+  requires an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink
+  that cannot absorb re-delivery (§9). The one integration point the group adds is state cleanup: a
+  member whose tasks all succeeded holds populated commit-coordinator state that is never cleared
+  (the clear path runs only on the holder's own failure), so group teardown must clear each member's
+  output-commit state — via `stageEnd` / fresh stage ids on rerun — or a rerun's commits would be
+  denied against the dead attempt's holders.
 
 ### 5.1 Observable completion events (listener bus)
 
@@ -205,6 +233,15 @@ member as completed before the group has committed.
     availability would miss the consumer here; teardown is therefore keyed on group membership, so a
     producer failure always tears down its co-scheduled consumers regardless of the producer's
     availability at the failure instant.
+  - *(Refinement, post-v1.)* v1 fails the group on any member's executor loss — safe and hang-free,
+    but over-eager: a producer that finished and whose output was already fully consumed can lose its
+    executor with no impact on the DAG, yet still triggers a rerun. A more precise, consumer-driven
+    signal is a post-v1 optimization: have the streaming reader raise a `FetchFailed` only when it
+    actually fails to read its input, so a post-consumption producer loss yields no failure and no
+    teardown. This reuses Spark's existing `FetchFailed` channel as the notification, with one
+    adaptation — on a pipelined edge it must route to **group failure**, not the base single-stage
+    mapper-resubmit (disabled for members) — so detection is consumer-driven while recovery stays
+    group-atomic.
 - **Recovery.** Group failure tears down all member stages atomically and discards their deferred
   completions. Because the shuffle is transient it cannot be partially recovered; the job fails and
   the caller re-runs it.
@@ -218,11 +255,19 @@ member as completed before the group has committed.
   - A lost external durable input reruns the group rather than being recomputed in isolation.
 - **Regular shuffle out of a group — supported.** A regular dependency from a member to a stage
   outside the group produces a materialized output that a downstream group consumes with normal
-  sequencing; the downstream waits for the group to complete.
-- **Regular shuffle internal to a group — fail-fast / unsupported.** If a regular dependency's
-  producer and consumer both fall in the same group, the group would co-schedule them concurrently
-  while the regular edge demands the producer be materialized first — a contradiction. Reject at
-  stage creation.
+  sequencing; the downstream waits for the group to complete. A group may have more than one such
+  output edge (from one or several members), and need not contain a `ResultStage` at all (§2.2).
+  - A group's external input and output edges are ordinary regular shuffles and behave exactly as
+    today, including optimizations like push-based shuffle merge; the pipelined restrictions apply
+    only to the group's internal (incremental) edges.
+- **Regular shuffle internal to a group — should not occur; checked as an invariant.** Grouping is
+  the connected component over *pipelined* edges only (§3), so a regular shuffle is a group
+  *boundary*: its producer and consumer fall into different groups by construction, splitting a
+  pipelined region rather than sitting inside one. A regular shuffle therefore should never be
+  internal to a group. The scheduler still verifies this as an invariant at stage/group creation and
+  fails fast if violated — a regular edge whose endpoints landed in the same group would be a
+  contradiction (co-schedule them, yet materialize-before-read), signalling inconsistent pipelined
+  marking rather than a valid plan.
 - Pipelined edges are intra-group by construction, so they never cross a group boundary.
 
 ---
@@ -242,25 +287,30 @@ member as completed before the group has committed.
 
 A pipelined group that involves any of the following is rejected at stage/group creation with a clear
 error, rather than silently mis-scheduled. Each is a documented limitation of the first version.
+The scope throughout is **within a PG** — how each idiom interacts with the members and internal
+(pipelined) edges of a group. A group's regular input/output edges are ordinary shuffles and are
+unaffected (§7); the entries below do not restrict them.
 
 The rejected idioms fall into two kinds. Some are **moot** under the group failure model (§6):
 Spark's stage-level recompute/rollback mechanisms never fire inside a group, because any failure
 aborts the whole group and the caller reruns from scratch — so a mechanism whose only job is to
 recompute or roll back a stage after a partial failure is never reached. We reject these rather than
 carry dead, self-contradictory machinery. The rest are **incompatible** with concurrent execution
-itself and would corrupt or hang a group that never fails.
+itself and would corrupt or hang a group that never fails. One row is neither — an **invariant** that
+should hold by construction (§3) but is still checked defensively, and would signal an inconsistent
+plan if violated.
 
 | Rejected condition | Kind | Why rejected |
 |--------------------|------|--------------|
 | Barrier execution in a member stage | incompatible | Barrier exposes output only after a global sync, contradicting concurrent partial reads. |
 | Dynamic resource allocation | incompatible | Gang admission needs a stable slot set; reclaiming executors from a pinned-open group can deadlock it. |
 | Speculative execution | incompatible | A speculative producer copy races a consumer already reading partial output; no commit barrier protects the read. |
-| Push-based shuffle merge | incompatible | Push-based shuffle merges each partition's blocks into large pre-sealed files on merger services, and those files become readable only after a post-map-completion "finalize" step — the exact opposite of a pipelined shuffle, whose consumer reads before the producer finishes, so the two can never apply to the same edge. |
+| Push-based shuffle merge as a pipelined (incremental) shuffle | incompatible | Push-based merge exposes output only after a post-completion "finalize" step — the opposite of incremental reads — so it cannot serve as the incremental shuffle within a PG. (A PG's regular input/output edges may still use push-based merge, like any regular shuffle — §7.) |
 | Statically-indeterminate producer | moot | Its recovery is stage rollback-and-recompute, which a group never performs (§6: any failure aborts the group). The mechanism is never reached; rejected so the inapplicability is explicit rather than latent. (A producer whose output RDD is classified `INDETERMINATE` — a rerun can yield different data, not just a different order — determinable from the RDD graph at stage-creation time.) |
 | Checksum-mismatch full retry | moot | The runtime counterpart to static indeterminism: it checksums each map task's output and, on a cross-attempt mismatch, rolls back and re-runs the succeeding stages. A group never keeps succeeding stages across a retry (§6), so this never fires; rejected to keep the inapplicability explicit. |
-| Cached/persisted RDD in a member's within-stage chain | incompatible | Would capture partial output read mid-run and serve it as a complete result on reuse. Scoped to the member's within-stage (narrow-dependency) chain: a cached *complete* input reached via a broadcast variable or across a materialized shuffle — e.g. the static side of a stream-static join — is outside that chain and is unaffected. |
-| Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. |
-| Regular shuffle internal to a group | incompatible | Concurrent co-scheduling contradicts materialize-before-read (§7). |
+| Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. Enforced with the fan-out rule below — a pipelined dependency is never reused or shared (§4). |
+| Pipelined producer with more than one consumer (fan-out) | incompatible | Fan-out of a pipelined edge comes only from exchange/stage reuse, which needs a durable, independently-readable output; a transient incremental shuffle has none (it is a live, once-through stream). Well-defined only over a persistent/replayable shuffle (the deferred `persistentHint` axis, §2), so v1 enforces 1:1 by never reusing a pipelined dependency (§4). |
+| Regular shuffle internal to a group (should not occur) | invariant | Split by construction into separate groups (§3); checked and rejected as an inconsistency if it ever arises (§7). |
 | Members with differing resource profiles | incompatible | The gang slot check compares one demand against one capacity (§4); v1 requires a group to be single-profile. Per-profile accounting is a follow-up. |
 | Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
 

From a2d82097014544815a349e1f6e8a4fcda802716f Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Thu, 9 Jul 2026 06:31:49 +0000
Subject: [PATCH 06/10] [SPARK-XXXXX][DOC] Address review round 3 (mridulm):
 drop S5.1 open question on member StageCompleted timing

Remove the open question asking whether SparkListenerStageCompleted for a
member should be deferred to group commit or emitted in real time. The S5.1
table already specifies the behavior (deferred to group commit), so the
trailing open-question paragraph was redundant.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 8c5cf74497c3..8cc52c5b8da5 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -205,12 +205,6 @@ success), a group failure needs only to emit them as *failure* — there is no p
 to retract. Already-emitted task events stand as-is (they were true), consistent with how Spark
 treats task events from a stage attempt that is later discarded.
 
-Open question for reviewers: whether `SparkListenerStageCompleted` for a member should be deferred to
-group commit (above) or emitted in real time when the member's tasks finish. Deferring keeps the
-listener's notion of "completed" consistent with the atomic-commit model at the cost of an inflated
-stage duration on the timeline; real-time emission gives crisper per-stage timings but reports a
-member as completed before the group has committed.
-
 ---
 
 ## 6. Failure

From e36cba4558442b6d83e3ac58891bd88b9c49f678 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Thu, 9 Jul 2026 18:27:35 +0000
Subject: [PATCH 07/10] [SPARK-XXXXX][DOC] Rewrite spec for clarity and
 concision (semantics unchanged)

A readability pass over the pipelined-shuffle spec. No semantic changes --
verified by sharded adversarial review (162 normative claims checked against
the original; zero critical drift). Changes are wording, structure, and
de-duplication only:

- Unify the group abbreviation to PG throughout (was a mix of G and PG).
- Fix a dangling reference in the S9 fan-out row: it pointed at a
  persistentHint axis in S2 that S2 never defined; reworded to 'a deferred
  capability v1 does not provide' (kind and reason unchanged).
- Keep the free-vs-total slot-check rationale in one place (S4.1); drop the
  duplicated derivation from S4's 'How capacity is measured'.
- Split the dense S5 output-commit bullet into two (idempotent commit +
  teardown state cleanup) and shorten the sentences.
- Unify S5.1 terminology on 'group completion' (was 'group commit' /
  'atomic commit' for the same milestone).
- Regroup the S9 table rows to match the preamble's taxonomy (moot,
  incompatible, invariant); same rows, same kinds.
- Add a short worked example in S2.2 (Z -> A -> B -> C) illustrating grouping,
  external input, concurrency, and materialized output.
- Remove assorted restatements across S2.1, S2.2, S4, S4.1, S6, S7.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 164 +++++++++++++--------------
 1 file changed, 82 insertions(+), 82 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 8cc52c5b8da5..ac17a33bdf1e 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -31,9 +31,8 @@ output while the producer stage is still running.
   and carried into the `ShuffleDependency` the `DAGScheduler` reads at stage-creation time.
 - It is also the **per-dependency selector** for the shuffle implementation: the shuffle layer maps a
   pipelined dependency to an incremental `ShuffleManager` and everything else to the default, so one
-  job with both regular and pipelined groups uses the right implementation for each — selected
-  per-dependency, not per-job. The scheduler construct stays generic; the shuffle implementation
-  stays pluggable.
+  job with both regular and pipelined groups uses the right implementation for each. The scheduler
+  construct stays generic; the shuffle implementation stays pluggable.
   - For example, an additional conf (e.g. `spark.shuffle.manager.incremental`) can be introduced to
     specify the incremental shuffle implementation used for pipelined shuffle dependencies, alongside
     the existing `spark.shuffle.manager` for the default.
@@ -42,12 +41,11 @@ A **regular shuffle dependency (RSD)** is an ordinary shuffle dependency: its ou
 materialized before any consumer reads it.
 
 Note: the name *pipelined* is deliberately chosen over *streaming*. The property is that a consumer
-reads producer output as it is produced — software-pipelining of dependent stages — which is a
-general execution capability. Streaming / real-time mode is the first caller, but nothing about the
-primitive is streaming-specific, so the dependency and the group are named for the capability, not
-the caller.
+reads producer output as it is produced — software-pipelining of dependent stages — a general
+execution capability. Streaming / real-time mode is the first caller, but nothing about the primitive
+is streaming-specific.
 
-### 2.2 Pipelined group (G)
+### 2.2 Pipelined group (PG)
 
 The set of stages connected to one another through pipelined edges — the connected component of the
 stage DAG when only pipelined edges are considered.
@@ -57,15 +55,22 @@ stage DAG when only pipelined edges are considered.
 - The group — not the edge or the individual stage — is the unit of **admission**, **slot
   checking**, **completion**, and **failure**.
 
-**External input of G:** a regular shuffle dependency whose consumer is in G and whose producer is
-not — i.e. a normal materialized parent of the group.
+**External input of PG:** a regular shuffle dependency whose consumer is in the group and whose
+producer is not — i.e. a normal materialized parent of the group.
 
-**Outputs of G:** a pipelined group need not contain a `ResultStage`. Its outputs to the rest of the
+**Outputs of PG:** a pipelined group need not contain a `ResultStage`. Its outputs to the rest of the
 DAG are regular (materialized) shuffle edges from its members to consumers outside the group, and
-there may be **more than one** — e.g. two different members each feeding a distinct downstream
-stage. A `ResultStage` is one possible member, not a requirement; a PG may instead be an interior
-component whose materialized outputs feed downstream groups (§7). In this respect a PG behaves like
-a barrier stage embedded in a larger DAG: an interior unit with regular-shuffle edges in and out.
+there may be **more than one** — e.g. two members each feeding a distinct downstream stage. A PG may
+instead be an interior component whose materialized outputs feed downstream groups (§7); in that
+respect it behaves like a barrier stage embedded in a larger DAG — an interior unit with
+regular-shuffle edges in and out.
+
+**Example.** Take a job with stage DAG `Z -> A -> B -> C`, where `Z -> A` and `B -> C` are regular
+edges and `A -> B` is pipelined. Grouping over pipelined edges gives one non-singleton group
+`PG = {A, B}` (`Z` and `C` are singletons). `Z`'s output is the group's external input, which the
+group waits for; `A` and `B` run concurrently, `B` reading `A`'s output as `A` produces it; and
+`B`'s output to `C` is a materialized output of the group that `C` consumes by normal
+materialize-before-read sequencing.
 
 ---
 
@@ -92,13 +97,13 @@ a barrier stage embedded in a larger DAG: an interior unit with regular-shuffle
 
 - **A pipelined edge is non-sequencing.** The consumer of a pipelined dependency does not wait for
   its producer to materialize. (A regular-dependency consumer still waits — the default behavior.)
-- **Group readiness.** A group is ready to be admitted when every external input of the group (its
-  regular materialized parents) is available — the same precondition a normal stage has today, lifted
-  to the group. Pipelined parents inside the group impose no readiness precondition.
+- **Group readiness.** A group is ready to be admitted when every external input (its regular
+  materialized parents) is available — the same precondition a normal stage has today, lifted to the
+  group. Pipelined parents inside the group impose no readiness precondition.
 - **Gang admission (all-or-nothing).** A pipelined group is admitted only if the cluster can currently
   run all tasks of all member stages **concurrently**; admission then submits every member stage at
-  once. There is no partial admission — a pipelined group is never left with some members running
-  while others wait on slots the running members occupy.
+  once. A pipelined group is never left with some members running while others wait on slots the
+  running members occupy.
   - *A non-pipelined (singleton) group is unaffected:* it is admitted exactly as a normal stage is
     today — submitted when its parents are available, filling slots incrementally, with no all-at-once
     requirement. Gang admission applies only to a group of two or more stages connected by pipelined
@@ -111,9 +116,7 @@ a barrier stage embedded in a larger DAG: an interior unit with regular-shuffle
   - *How capacity is measured:* the total concurrent-task capacity is the value barrier's slot check
     uses (`sc.maxNumConcurrentTasks` — task slots summed over active executors), not
     `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run
-    concurrently). Note this diverges from barrier in one way: barrier compares demand against that
-    *total*, whereas a pipelined group compares against *free* capacity (total minus running tasks).
-    Computing free capacity is therefore a new computation, not barrier's check reused verbatim.
+    concurrently).
 - **Single resource profile per group (v1).** A resource profile is the executor/task resource
   requirement (cores, memory, GPUs, ...) a stage runs under; the number of concurrent slots is
   defined *per profile* (a cluster may run many concurrent CPU tasks but few GPU tasks). Comparing
@@ -123,26 +126,27 @@ a barrier stage embedded in a larger DAG: an interior unit with regular-shuffle
   follow-up, not needed for the streaming shapes whose members share the default profile.
 - **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
 - **Single ownership and 1:1 (v1), enforced by never reusing a pipelined dependency.** A pipelined
-  producer feeds exactly one consumer and belongs to exactly one job. The regular-shuffle 1:many
-  fan-out and cross-job stage reuse are both disallowed, and both are prevented by the same
-  mechanism: a pipelined `ShuffleDependency` is never reused or shared. At stage/group creation the
-  `shuffleIdToMapStage` reuse path is bypassed for a pipelined dependency (each gets a fresh
-  producer stage, never bound to an existing one), and a group is rejected (fail-fast, §9) if a
-  pipelined producer has more than one consuming stage or its stage carries more than one jobId.
-  Reuse must be prevented explicitly: from the scheduler's perspective a shuffle-map stage *can* be
+  producer feeds exactly one consumer and belongs to exactly one job. Both the regular-shuffle
+  1:many fan-out and cross-job stage reuse are disallowed, prevented by the same mechanism: a
+  pipelined `ShuffleDependency` is never reused or shared. Concretely, at stage/group creation the
+  `shuffleIdToMapStage` reuse path is bypassed for a pipelined dependency, so each gets a fresh
+  producer stage never bound to an existing one. Separately, a group is rejected (fail-fast, §9) if a
+  pipelined producer has more than one consuming stage, or its stage carries more than one jobId.
+  This must be enforced explicitly: from the scheduler's perspective a shuffle-map stage *can* be
   reused across jobs unless something forbids it.
 
 ### 4.1 Cross-group admission (multiple groups / groups vs. regular jobs)
 
 A pipelined group holds its slots for its whole run, so admission is decided against currently-free
-slots, and a group that does not fit is failed rather than queued.
+slots.
 
 - **Why free slots, not total** (the slot check itself is defined in §4). Free capacity subtracts
   the tasks already running — for every running group and regular job — so a group is admitted only
   if its full demand fits in what the others leave free. This is what makes co-residency safe:
   comparing against *total* capacity would let two groups each pass the check yet fail to co-fit,
-  which is exactly the partial co-residency gang admission forbids. (This is why the §4 slot check
-  diverges from barrier's total-capacity check.)
+  which is exactly the partial co-residency that gang admission forbids. This is where the check
+  diverges from barrier, which compares demand against total capacity; computing free capacity is a
+  new computation, not barrier's check reused verbatim.
 - **No waiting queue, no partial reservation.** A group that doesn't fit fails its admission; it never
   sits in a queue holding slots incrementally. This keeps the scheduler change minimal and cannot
   deadlock (a group never occupies slots a sibling is blocked on).
@@ -158,10 +162,10 @@ slots, and a group that does not fit is failed rather than queued.
 ## 5. Completion
 
 - **Group-observable completion.** A member stage is not observably finished until **all** member
-  stages of its group have completed successfully — a member that finishes ahead of the others
-  cannot advance stage or job completion, nor make its output visible to a downstream consumer,
-  until the group does. Whether the group's output is a `ResultStage`'s job result or a materialized
-  shuffle feeding a downstream group (§2.2), that output becomes observable only at group completion.
+  stages of its group have completed successfully. A member that finishes ahead of the others cannot
+  advance stage or job completion, nor make its output visible to a downstream consumer, until the
+  group does — whether that output is a `ResultStage`'s job result or a materialized shuffle feeding a
+  downstream group (§2.2).
 - **Defer the finish decision, not per-task work.** When a member finishes before its group, only its
   stage-finish / job-finish transition is deferred until group completion. Per-task side effects that
   must always run — accumulator updates, output-commit coordination, task-end listener events — run
@@ -170,47 +174,45 @@ slots, and a group that does not fit is failed rather than queued.
   that window is a group failure (§6): the deferred finish transitions are dropped and the group
   reruns.
 - **In-group result-stage side effects must be idempotent.** Per-task side effects run immediately
-  (above), including a result stage's output commit. The output-commit path itself is unchanged from
-  base Spark: `OutputCommitCoordinator` arbitration — deciding which of several racing attempts of a
-  partition may commit — never contends inside a group, because both sources of concurrent attempts
-  are absent (speculation is rejected, §9; single-stage resubmit is disabled for members, §6), so
-  each partition has exactly one attempt and is trivially authorized. If a result task commits and a
-  sibling then fails in the replay window, the group reruns and re-delivers that output. This is the
-  standard streaming model — a batch is re-delivered on recovery and the sink must absorb it — so v1
-  requires an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink
-  that cannot absorb re-delivery (§9). The one integration point the group adds is state cleanup: a
-  member whose tasks all succeeded holds populated commit-coordinator state that is never cleared
-  (the clear path runs only on the holder's own failure), so group teardown must clear each member's
-  output-commit state — via `stageEnd` / fresh stage ids on rerun — or a rerun's commits would be
-  denied against the dead attempt's holders.
+  (above), including a result stage's output commit, and the commit path itself is unchanged from
+  base Spark. `OutputCommitCoordinator` arbitration decides which of several racing attempts at a
+  partition may commit; inside a group it never contends, because both sources of concurrent attempts
+  are absent (speculation is rejected, §9; single-stage resubmit is disabled for members, §6). Each
+  partition therefore has exactly one attempt and is trivially authorized. If a result task commits
+  and a sibling then fails in the replay window, the group reruns and re-delivers that output — the
+  standard streaming model, where a batch is re-delivered on recovery and the sink must absorb it. So
+  v1 requires an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink
+  that cannot absorb re-delivery (§9).
+- **Group teardown clears members' output-commit state.** When a member's tasks all succeed, its
+  commit-coordinator state stays populated — the clear path runs only when the holder's own attempt
+  fails. So group teardown must clear each member's output-commit state (via `stageEnd`, or fresh
+  stage ids on rerun); otherwise a rerun's commits are denied against the dead attempt's holders.
 
 ### 5.1 Observable completion events (listener bus)
 
 The listener bus is an external contract that monitoring tools depend on, so it is worth stating
-exactly when each event is delivered. The rule follows directly from atomic commit: **task-level
-events flow in real time, but stage-completion and job-completion events are held until the group
-commits — so a listener never observes a member as *successfully completed* before the group as a
-whole has.**
+exactly when each event is delivered. The rule follows directly from group-atomic completion (the
+point at which the group's outputs become observable — distinct from the per-task output-commit
+above): **task-level events flow in real time, but stage-completion and job-completion events are
+held until the group completes — so a listener never observes a member as *successfully completed*
+before the group as a whole has.**
 
 | Event | Timing | Rationale |
 |-------|--------|-----------|
 | `SparkListenerTaskStart` / `SparkListenerTaskEnd` | Real time, as they occur | Per-task facts are true when they happen; a group's members genuinely run concurrently. Deferring these would freeze a member's live progress and metrics for the whole group's duration. Note a successful `TaskEnd` means "this task finished," not "its output is committed" — already true in Spark, since a stage attempt can later be discarded. |
 | `SparkListenerStageSubmitted` | Real time, at group admission | All member stages are submitted together (§4); a monitor should show them active simultaneously. |
-| `SparkListenerStageCompleted` | Deferred to group commit; on group failure, emitted with a failure reason | "Completed" should track commit, which is atomic at the group level. A member whose tasks finish early is reported as still running until the group commits — which matches the truth that its results are not usable until then. This avoids emitting a success-shaped completion for an attempt that a later group failure would discard. |
-| `SparkListenerJobEnd(JobSucceeded)` | At group commit only | Job completion delivers results to the caller and cancels sibling stages; emitting it before the group commits risks double/inconsistent result delivery if the group later fails. Non-negotiable. |
+| `SparkListenerStageCompleted` | Deferred to group completion; on group failure, emitted with a failure reason | "Completed" should track group completion, which is atomic at the group level. A member whose tasks finish early is reported as still running until the group completes — which matches the truth that its results are not usable until then. This avoids emitting a success-shaped completion for an attempt that a later group failure would discard. |
+| `SparkListenerJobEnd(JobSucceeded)` | At group completion only | Job completion delivers results to the caller and cancels sibling stages; emitting it before the group completes risks double/inconsistent result delivery if the group later fails. Non-negotiable. |
 | `SparkListenerJobEnd(JobFailed)` | On group failure | Group-atomic failure (§6): buffered success transitions are dropped, never replayed as success. |
 
-Failure-path consequence: because stage- and job-completion are deferred (never emitted early as
-success), a group failure needs only to emit them as *failure* — there is no premature success event
-to retract. Already-emitted task events stand as-is (they were true), consistent with how Spark
-treats task events from a stage attempt that is later discarded.
+Failure-path consequence: already-emitted task events stand as-is (they were true), consistent with
+how Spark treats task events from a stage attempt that is later discarded.
 
 ---
 
 ## 6. Failure
 
-- **Failure is group-atomic.** Any member task failure, for any reason, fails the whole group; there
-  is no independent single-stage failure within a group.
+- **Failure is group-atomic.** Any member task failure, for any reason, fails the whole group.
   - *Single-stage resubmit is disabled for group members.* That isolated-resubmit branch is turned
     off for a member of a pipelined group: the transient pipelined shuffle cannot be re-read and
     members are co-scheduled, so a lone-stage resubmit is never valid. With it disabled, every member
@@ -220,13 +222,13 @@ treats task events from a stage attempt that is later discarded.
       the failed stage in isolation and resuming.
   - *Teardown is by group membership, not producer availability.* At the end of a batch the producer
     finishes and registers all its map outputs — becoming "available" — while the consumer is still
-    draining the remainder, so a producer-available-while-consumer-still-running window is normal,
-    not rare. For a transient pipelined shuffle, "available" does not mean the consumer is safe: the
-    data is not durably stored and the consumer is still actively reading it, so a producer failure
-    in that window still strands the consumer. Failure propagation that keys off producer
-    availability would miss the consumer here; teardown is therefore keyed on group membership, so a
-    producer failure always tears down its co-scheduled consumers regardless of the producer's
-    availability at the failure instant.
+    draining the remainder, so it is normal, not rare, for the producer to be available while the
+    consumer is still running. For a transient pipelined shuffle, "available" does not mean the
+    consumer is safe: the data is not durably stored and the consumer is still actively reading it, so
+    a producer failure in that window still strands the consumer. Failure propagation that keys off
+    producer availability would miss the consumer here; teardown is therefore keyed on group
+    membership, so a producer failure always tears down its co-scheduled consumers regardless of the
+    producer's availability at the failure instant.
   - *(Refinement, post-v1.)* v1 fails the group on any member's executor loss — safe and hang-free,
     but over-eager: a producer that finished and whose output was already fully consumed can lose its
     executor with no impact on the DAG, yet still triggers a rerun. A more precise, consumer-driven
@@ -234,8 +236,7 @@ treats task events from a stage attempt that is later discarded.
     actually fails to read its input, so a post-consumption producer loss yields no failure and no
     teardown. This reuses Spark's existing `FetchFailed` channel as the notification, with one
     adaptation — on a pipelined edge it must route to **group failure**, not the base single-stage
-    mapper-resubmit (disabled for members) — so detection is consumer-driven while recovery stays
-    group-atomic.
+    mapper-resubmit — so detection is consumer-driven while recovery stays group-atomic.
 - **Recovery.** Group failure tears down all member stages atomically and discards their deferred
   completions. Because the shuffle is transient it cannot be partially recovered; the job fails and
   the caller re-runs it.
@@ -247,13 +248,12 @@ treats task events from a stage attempt that is later discarded.
 - **Regular shuffle into a group — supported.** A regular dependency from outside the group to a
   member is a normal materialized parent (an external input); the group waits for it.
   - A lost external durable input reruns the group rather than being recomputed in isolation.
-- **Regular shuffle out of a group — supported.** A regular dependency from a member to a stage
-  outside the group produces a materialized output that a downstream group consumes with normal
-  sequencing; the downstream waits for the group to complete. A group may have more than one such
-  output edge (from one or several members), and need not contain a `ResultStage` at all (§2.2).
-  - A group's external input and output edges are ordinary regular shuffles and behave exactly as
-    today, including optimizations like push-based shuffle merge; the pipelined restrictions apply
-    only to the group's internal (incremental) edges.
+- **Regular shuffle out of a group — supported.** A member may have a regular output edge to a stage
+  outside the group; the downstream consumer waits for the group to complete and reads it by normal
+  materialize-before-read sequencing. As in §2.2, a group may have more than one such output. These
+  external input and output edges are ordinary regular shuffles and behave exactly as today,
+  including optimizations like push-based shuffle merge; the pipelined restrictions apply only to the
+  group's internal (incremental) edges.
 - **Regular shuffle internal to a group — should not occur; checked as an invariant.** Grouping is
   the connected component over *pipelined* edges only (§3), so a regular shuffle is a group
   *boundary*: its producer and consumer fall into different groups by construction, splitting a
@@ -296,17 +296,17 @@ plan if violated.
 
 | Rejected condition | Kind | Why rejected |
 |--------------------|------|--------------|
+| Statically-indeterminate producer | moot | Its recovery is stage rollback-and-recompute, which a group never performs (§6: any failure aborts the group), so the mechanism is never reached. (A producer whose output RDD is classified `INDETERMINATE` — a rerun can yield different data, not just a different order — determinable from the RDD graph at stage-creation time.) |
+| Checksum-mismatch full retry | moot | The runtime counterpart to static indeterminism: it checksums each map task's output and, on a cross-attempt mismatch, rolls back and re-runs the succeeding stages. A group never keeps succeeding stages across a retry (§6), so this never fires. |
 | Barrier execution in a member stage | incompatible | Barrier exposes output only after a global sync, contradicting concurrent partial reads. |
 | Dynamic resource allocation | incompatible | Gang admission needs a stable slot set; reclaiming executors from a pinned-open group can deadlock it. |
 | Speculative execution | incompatible | A speculative producer copy races a consumer already reading partial output; no commit barrier protects the read. |
 | Push-based shuffle merge as a pipelined (incremental) shuffle | incompatible | Push-based merge exposes output only after a post-completion "finalize" step — the opposite of incremental reads — so it cannot serve as the incremental shuffle within a PG. (A PG's regular input/output edges may still use push-based merge, like any regular shuffle — §7.) |
-| Statically-indeterminate producer | moot | Its recovery is stage rollback-and-recompute, which a group never performs (§6: any failure aborts the group). The mechanism is never reached; rejected so the inapplicability is explicit rather than latent. (A producer whose output RDD is classified `INDETERMINATE` — a rerun can yield different data, not just a different order — determinable from the RDD graph at stage-creation time.) |
-| Checksum-mismatch full retry | moot | The runtime counterpart to static indeterminism: it checksums each map task's output and, on a cross-attempt mismatch, rolls back and re-runs the succeeding stages. A group never keeps succeeding stages across a retry (§6), so this never fires; rejected to keep the inapplicability explicit. |
 | Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. Enforced with the fan-out rule below — a pipelined dependency is never reused or shared (§4). |
-| Pipelined producer with more than one consumer (fan-out) | incompatible | Fan-out of a pipelined edge comes only from exchange/stage reuse, which needs a durable, independently-readable output; a transient incremental shuffle has none (it is a live, once-through stream). Well-defined only over a persistent/replayable shuffle (the deferred `persistentHint` axis, §2), so v1 enforces 1:1 by never reusing a pipelined dependency (§4). |
-| Regular shuffle internal to a group (should not occur) | invariant | Split by construction into separate groups (§3); checked and rejected as an inconsistency if it ever arises (§7). |
+| Pipelined producer with more than one consumer (fan-out) | incompatible | Fan-out of a pipelined edge comes only from exchange/stage reuse, which needs a durable, independently-readable output; a transient incremental shuffle has none (it is a live, once-through stream). It would be well-defined only over a persistent/replayable shuffle, a deferred capability v1 does not provide, so v1 enforces 1:1 by never reusing a pipelined dependency (§4). |
 | Members with differing resource profiles | incompatible | The gang slot check compares one demand against one capacity (§4); v1 requires a group to be single-profile. Per-profile accounting is a follow-up. |
 | Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
+| Regular shuffle internal to a group (should not occur) | invariant | Split by construction into separate groups (§3); checked and rejected as an inconsistency if it ever arises (§7). |
 
 The AQE row forbids AQE's *intra-batch, mid-stage* replanning — reading a running producer's statistics
 within one execution and reshaping its consumers — not statistics-driven adaptivity in general. A

From 12fe2a0d94e089ecbafee8e6e9328879bd70c8e4 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Sat, 11 Jul 2026 05:51:34 +0000
Subject: [PATCH 08/10] [SPARK-XXXXX][DOC] Address review round 4 (mridulm
 follow-up): fan-out deferred, reuse layers, checkpoint, admission retry,
 commit reset

Addresses @mridulm's follow-up comments on the spec:

- Fan-out / generality (S4, S5, S9): reframe single-ownership at the construct
  level (the pipelined dependency and PG are generic DAGScheduler constructs, not
  SQL/RTM-specific). Split fan-out into two cases: cross-time / cross-job reuse
  stays incompatible (a transient once-through edge has no retained output to
  reuse), but concurrent branching (1:N co-scheduled consumers) is reclassified
  from incompatible to DEFERRED -- a supported model not built in v1 (needs
  multicast + per-consumer backpressure). A producer may also tee a durable
  materialized edge for out-of-group consumers; S5 gains deferred output
  registration so that dual-write is consistency-clean.

- Cross-job reuse enforcement (S4): a pipelined shuffle is transient, so reuse
  across jobs is unsound and must be prevented at BOTH layers the scheduler would
  otherwise reuse a shuffle -- shuffleIdToMapStage stage reuse AND the
  MapOutputTracker-availability skip -- not just the former.

- Checkpoint (S9): reliable RDD.checkpoint() in a member's within-stage chain is
  incompatible (durable lineage-truncated snapshot -> cross-time reuse; and a
  post-success recompute of a now-vanished transient input). Fail fast on
  checkpointData being a ReliableRDDCheckpointData. Cache / persist / local
  checkpoint remain whole-partition/ephemeral and allowed.

- Admission retry (S4.1): v1 leaves retry to the caller, but that only suffices
  for a simple linear-chain PG run by a caller with its own restart loop. Name
  that limitation, and add scheduler-side PG-granular admission retry (mirroring
  barrier's transient-shortfall retry) as a post-v1 refinement.

- Output-commit (S5): a PG is an atomic unit with no per-task replay, whereas
  OutputCommitCoordinator permanently denies re-commit of a committed partition
  (sound today because only missing partitions are recomputed). A group-atomic
  rerun reruns already-committed result tasks, so the rerun must reset
  per-partition commit authorization -- via fresh stage ids or stageEnd on
  teardown.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 105 +++++++++++++++++++--------
 1 file changed, 73 insertions(+), 32 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index ac17a33bdf1e..231683c5e3ef 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -125,15 +125,26 @@ materialize-before-read sequencing.
   per-profile accounting — checking each profile's demand against that profile's own capacity — is a
   follow-up, not needed for the streaming shapes whose members share the default profile.
 - **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
-- **Single ownership and 1:1 (v1), enforced by never reusing a pipelined dependency.** A pipelined
-  producer feeds exactly one consumer and belongs to exactly one job. Both the regular-shuffle
-  1:many fan-out and cross-job stage reuse are disallowed, prevented by the same mechanism: a
-  pipelined `ShuffleDependency` is never reused or shared. Concretely, at stage/group creation the
-  `shuffleIdToMapStage` reuse path is bypassed for a pipelined dependency, so each gets a fresh
-  producer stage never bound to an existing one. Separately, a group is rejected (fail-fast, §9) if a
-  pipelined producer has more than one consuming stage, or its stage carries more than one jobId.
-  This must be enforced explicitly: from the scheduler's perspective a shuffle-map stage *can* be
-  reused across jobs unless something forbids it.
+- **Single ownership and 1:1 (v1).** In v1 a pipelined producer feeds exactly one consumer and
+  belongs to exactly one job. These are two separate restrictions, and they are stated at the level
+  of the construct itself, not any particular caller (the pipelined dependency and PG are generic
+  `DAGScheduler` constructs that may be depended on directly, not only from SQL/RTM):
+  - *No cross-job / cross-time reuse — intrinsic.* A pipelined shuffle is transient: a once-through
+    live stream with no retained, addressable output. Unlike a regular shuffle, there is nothing for
+    a second job to reuse, so reuse across jobs is unsound for it. This must be enforced explicitly —
+    from the scheduler's perspective a shuffle-map stage *can* be reused unless something forbids it —
+    and at **both** layers by which the scheduler would otherwise reuse a shuffle: the
+    `shuffleIdToMapStage` stage-object reuse is bypassed (each pipelined dependency gets a fresh
+    producer stage), and the `MapOutputTracker`-availability reuse is prevented (a pipelined shuffle's
+    tracker registration does not outlive its group, so `getMissingParentStages` cannot skip a
+    re-created producer as already-available). Bypassing only the former is insufficient. A pipelined
+    producer whose stage carries more than one jobId is rejected (fail-fast, §9). (A producer may
+    separately emit a durable *materialized* output edge — a distinct regular `ShuffleDependency` that
+    reuses normally; run-once binds only the transient edge.)
+  - *1:1 within the group (v1) — deferred, not intrinsic.* v1 rejects a pipelined producer with more
+    than one consuming stage, but 1:N fan-out is a supported model that v1 defers rather than an
+    incompatibility (§9): co-scheduled consumers would read the live stream via multicast, and
+    non-co-scheduled consumers read a materialized edge the producer also writes.
 
 ### 4.1 Cross-group admission (multiple groups / groups vs. regular jobs)
 
@@ -152,10 +163,23 @@ slots.
   deadlock (a group never occupies slots a sibling is blocked on).
 - **Two failure reasons, one path.** Demand > total capacity can never fit (a plan/sizing error);
   demand > free slots but ≤ total could fit later. Both fail admission.
-- **Retry is the caller's decision.** The scheduler does not automatically retry a failed admission —
-  it fails the job. Whether and how to retry is up to the caller. For a streaming query this is the
-  batch-execution loop restarting the batch (with its own backoff), so a transiently-busy cluster is
-  handled by the caller's restart policy, not a scheduler retry loop.
+- **Retry (v1: caller's decision; scheduler-side retry is a post-v1 refinement).** In v1 the
+  scheduler does not retry a failed admission — it fails the job, and retry is the caller's decision
+  (for a streaming query, the batch-execution loop restarting the batch with its own backoff). This
+  is adequate only for a simple `prefix* -> PG -> suffix*` shape run by a caller that has its own
+  restart loop: caller-retry's unit is the whole job, so with concurrent work (sibling stages, other
+  PGs) or an already-committed prefix it discards work it cannot selectively preserve, and a
+  directly-depended-on PG may have no retrying caller at all — so a transient slot shortfall either
+  kills the job or forces over-provisioning.
+  - *(Refinement, post-v1.)* The scheduler retries **admission of the PG specifically** — hold the
+    group, re-run the gang slot-check on a timer, admit when it fits, fail after a bounded number of
+    attempts — leaving any completed prefix and concurrent stages untouched. This mirrors barrier's
+    transient-shortfall retry (re-post on a scheduled interval bounded by a max-failure count, then
+    fail; `DAGScheduler` `BarrierJobSlotsNumberCheckFailed` path), the difference being that a
+    mid-DAG PG retries its own admission rather than re-posting the whole job (barrier can re-post
+    the job only because its check runs before any stage executes). Making transient-shortfall
+    tolerance a property of the construct, rather than a burden each caller re-implements, is why it
+    belongs in the scheduler.
 
 ---
 
@@ -170,23 +194,37 @@ slots.
   stage-finish / job-finish transition is deferred until group completion. Per-task side effects that
   must always run — accumulator updates, output-commit coordination, task-end listener events — run
   immediately.
+  - *Deferred output registration.* A member's materialized output edge (to a consumer outside the
+    group, §7) is written as its tasks run, but its map-output registration is **deferred to group
+    completion** — so an out-of-group consumer, which waits for the group anyway (§7), only ever sees
+    the single committed version. A failed intermediate attempt's blocks are simply orphaned (never
+    registered), exactly as for a resubmitted regular map stage today. This is what makes a producer
+    that writes both an incremental (in-group) and a materialized (out-of-group) output edge
+    consistency-clean: the group's internal replay never reaches the materialized consumer, because
+    it has not started.
 - **Replay window.** There is a window between a member finishing and group completion. A failure in
   that window is a group failure (§6): the deferred finish transitions are dropped and the group
   reruns.
 - **In-group result-stage side effects must be idempotent.** Per-task side effects run immediately
-  (above), including a result stage's output commit, and the commit path itself is unchanged from
-  base Spark. `OutputCommitCoordinator` arbitration decides which of several racing attempts at a
-  partition may commit; inside a group it never contends, because both sources of concurrent attempts
-  are absent (speculation is rejected, §9; single-stage resubmit is disabled for members, §6). Each
-  partition therefore has exactly one attempt and is trivially authorized. If a result task commits
-  and a sibling then fails in the replay window, the group reruns and re-delivers that output — the
-  standard streaming model, where a batch is re-delivered on recovery and the sink must absorb it. So
-  v1 requires an in-group result stage's side effects to be idempotent, and fail-fast rejects a sink
-  that cannot absorb re-delivery (§9).
-- **Group teardown clears members' output-commit state.** When a member's tasks all succeed, its
-  commit-coordinator state stays populated — the clear path runs only when the holder's own attempt
-  fails. So group teardown must clear each member's output-commit state (via `stageEnd`, or fresh
-  stage ids on rerun); otherwise a rerun's commits are denied against the dead attempt's holders.
+  (above), including a result stage's output commit. If a result task commits and a sibling then
+  fails in the replay window, the group reruns and re-delivers that output — the standard streaming
+  model, where a batch is re-delivered on recovery and the sink must absorb it. So v1 requires an
+  in-group result stage's side effects to be idempotent, and fail-fast rejects a sink that cannot
+  absorb re-delivery (§9).
+- **Group-atomic rerun must reset per-partition commit authorization.** A PG is an atomic
+  scheduling unit: there are no per-task or per-partition replays within it. `OutputCommitCoordinator`
+  is built on the opposite assumption — it authorizes exactly one attempt per partition and
+  permanently denies any later request for a partition that already committed. That is sound today
+  because a committed partition is never recomputed: a stage rerun (e.g. on fetch failure) recomputes
+  only its *missing* partitions, so a committed task never needs to commit again, and the permanent
+  deny is correct. A PG breaks that assumption — because the group is atomic, a failure anywhere
+  reruns the *whole* group, including a result stage whose tasks already succeeded and committed. Those
+  committed partitions are rerun and must be allowed to commit again, but the coordinator still holds
+  the previous attempt's authorized committers and would deny them (a `Success` clears nothing; only a
+  *failed* holder's slot is cleared today). So the group rerun must let the new tasks commit, by one
+  of: (a) rerunning members under **fresh stage ids** — a fresh coordinator `StageState` with no prior
+  committers; or (b) **resetting** the committed state for those stages in the `OutputCommitCoordinator`
+  (e.g. `stageEnd` on teardown) before the rerun.
 
 ### 5.1 Observable completion events (listener bus)
 
@@ -289,10 +327,12 @@ The rejected idioms fall into two kinds. Some are **moot** under the group failu
 Spark's stage-level recompute/rollback mechanisms never fire inside a group, because any failure
 aborts the whole group and the caller reruns from scratch — so a mechanism whose only job is to
 recompute or roll back a stage after a partial failure is never reached. We reject these rather than
-carry dead, self-contradictory machinery. The rest are **incompatible** with concurrent execution
-itself and would corrupt or hang a group that never fails. One row is neither — an **invariant** that
-should hold by construction (§3) but is still checked defensively, and would signal an inconsistent
-plan if violated.
+carry dead, self-contradictory machinery. Most of the rest are **incompatible** with concurrent
+execution itself and would corrupt or hang a group that never fails. One row is neither — an
+**invariant** that should hold by construction (§3) but is still checked defensively, and would
+signal an inconsistent plan if violated. One further row is **deferred** — a capability that is
+well-defined and compatible with the model but not built in v1, so v1 rejects it fail-fast rather
+than mis-scheduling it; it is expected to be supported later.
 
 | Rejected condition | Kind | Why rejected |
 |--------------------|------|--------------|
@@ -302,8 +342,9 @@ plan if violated.
 | Dynamic resource allocation | incompatible | Gang admission needs a stable slot set; reclaiming executors from a pinned-open group can deadlock it. |
 | Speculative execution | incompatible | A speculative producer copy races a consumer already reading partial output; no commit barrier protects the read. |
 | Push-based shuffle merge as a pipelined (incremental) shuffle | incompatible | Push-based merge exposes output only after a post-completion "finalize" step — the opposite of incremental reads — so it cannot serve as the incremental shuffle within a PG. (A PG's regular input/output edges may still use push-based merge, like any regular shuffle — §7.) |
-| Pipelined producer shared across jobs | incompatible | Breaks single ownership: a failure must map to exactly one job's group. Enforced with the fan-out rule below — a pipelined dependency is never reused or shared (§4). |
-| Pipelined producer with more than one consumer (fan-out) | incompatible | Fan-out of a pipelined edge comes only from exchange/stage reuse, which needs a durable, independently-readable output; a transient incremental shuffle has none (it is a live, once-through stream). It would be well-defined only over a persistent/replayable shuffle, a deferred capability v1 does not provide, so v1 enforces 1:1 by never reusing a pipelined dependency (§4). |
+| Pipelined producer shared across jobs (cross-time reuse) | incompatible | A transient incremental edge is a live, once-through stream with no retained, addressable output, so a consumer running at a different time has nothing to fetch — cross-time / cross-job reuse is intrinsically undefined for it. Enforced by never binding a pipelined producer's stage to more than one job, at both the `shuffleIdToMapStage` and `MapOutputTracker` reuse layers (§4). |
+| Pipelined producer with more than one co-scheduled consumer (fan-out / branching) | deferred | 1:N is a supported model, deferred in v1. A consumer's transport is picked by group membership: an in-group consumer reads the incremental edge (needs multicast — one producer pushing each record to N live reader channels — plus per-consumer backpressure, not yet built); an out-of-group consumer reads a materialized edge the producer also writes (a tee), sequenced after group commit like any regular shuffle out (§7). v1 rejects a pipelined producer with more than one consuming stage fail-fast until multicast lands. |
+| Reliable RDD checkpoint in a member's within-stage chain | incompatible | Reliable `RDD.checkpoint()` writes a durable, lineage-truncated snapshot, which both reintroduces cross-time reuse of a transient edge (§4) and requires a post-success recompute of the member's now-vanished transient input. Rejected at group creation by walking the within-stage chain and failing on any RDD whose `checkpointData` is a `ReliableRDDCheckpointData` (keyed on `checkpointData`, not `isCheckpointed`, since the write has not happened yet). Cache / `.persist()` / local checkpoint are whole-partition and ephemeral, and are not rejected. |
 | Members with differing resource profiles | incompatible | The gang slot check compares one demand against one capacity (§4); v1 requires a group to be single-profile. Per-profile accounting is a follow-up. |
 | Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
 | Regular shuffle internal to a group (should not occur) | invariant | Split by construction into separate groups (§3); checked and rejected as an inconsistency if it ever arises (§7). |

From def82a45f3cb27fbd97f9208b1073c6c47b0be24 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Mon, 13 Jul 2026 04:42:56 +0000
Subject: [PATCH 09/10] [SPARK-XXXXX][DOC] Fact-check pass: fix S1 concurrency
 claim, clarify dense sections

Verified every code-fact claim in the spec against the Spark codebase and did a
clarity pass. One factual correction; the rest is presentation (no semantics
changed).

- S1 (correctness): 'a multi-stage job runs one stage at a time' was wrong --
  independent stages already run concurrently today (DAGScheduler.submitStage
  submits all missing parents); only shuffle-connected stages serialize. Scoped
  the claim accordingly.
- S4 single-ownership 'no cross-job reuse' and S5 commit-reset: split the two
  densest paragraphs into followable sub-points; meaning unchanged.
- S4 opening, S4.1 retry, S9 checkpoint row, S9 fan-out row: broke up
  multi-clause sentences and glossed jargon (multicast, transport, backpressure).
- Minor: expand RTM on first use; define a 'slot' in the S4 capacity note
  (cores / cores-per-task); note that pipelined/incremental/transient name the
  same edge.

All other code claims across S5-S9 (SparkListener events, OutputCommitCoordinator
behavior, findMissingPartitions, FetchFailed->resubmit, and every S9 row's named
mechanism) were verified accurate and left unchanged.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 67 +++++++++++++++-------------
 1 file changed, 37 insertions(+), 30 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 231683c5e3ef..56281bbdb804 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -7,10 +7,12 @@ consumer reads incrementally.
 
 ## 1. Motivation
 
-Today a multi-stage job runs one stage at a time: each shuffle is fully materialized before the next
-stage starts. Some workloads need the stages of a single job to run **concurrently**, connected by a
-shuffle whose consumer reads the producer's output **as it is produced** rather than after the
-producer finishes. This spec introduces the scheduler primitives to express and run that.
+Today, when two stages of a job are connected by a shuffle, they run one after another: the consumer
+stage does not start until the producer's shuffle output is fully materialized. (Independent stages —
+ones not connected by a shuffle — already run concurrently.) Some workloads need even
+shuffle-connected stages to run **concurrently**, with the consumer reading the producer's output
+**as it is produced** rather than after the producer finishes. This spec introduces the scheduler
+primitives to express and run that.
 
 "Run these stages concurrently" and "the connecting shuffle is incremental" are the same decision
 seen from two sides: co-scheduling a producer and consumer is only useful if the edge is readable
@@ -26,7 +28,9 @@ A shuffle dependency declared **incrementally readable**: a consumer stage may b
 output while the producer stage is still running.
 
 - It is a shuffle dependency (has a `shuffleId`, partitioner, map/reduce sides); the *pipelined*
-  property is a binding part of the scheduler contract, not an advisory hint.
+  property is a binding part of the scheduler contract, not an advisory hint. ("Pipelined",
+  "incremental", and "transient" all describe this same edge in this doc — read as its output being
+  streamed to the consumer as produced, not stored.)
 - The property is set during **physical planning** (an execution concern, not a logical-plan one)
   and carried into the `ShuffleDependency` the `DAGScheduler` reads at stage-creation time.
 - It is also the **per-dependency selector** for the shuffle implementation: the shuffle layer maps a
@@ -42,8 +46,8 @@ materialized before any consumer reads it.
 
 Note: the name *pipelined* is deliberately chosen over *streaming*. The property is that a consumer
 reads producer output as it is produced — software-pipelining of dependent stages — a general
-execution capability. Streaming / real-time mode is the first caller, but nothing about the primitive
-is streaming-specific.
+execution capability. Streaming / real-time mode (RTM) is the first caller, but nothing about the
+primitive is streaming-specific.
 
 ### 2.2 Pipelined group (PG)
 
@@ -114,9 +118,9 @@ materialize-before-read sequencing.
   admission fails fast, since the group could never become fully co-resident. (Free rather than
   total is essential once more than one group can be admitted; §4.1 explains why.)
   - *How capacity is measured:* the total concurrent-task capacity is the value barrier's slot check
-    uses (`sc.maxNumConcurrentTasks` — task slots summed over active executors), not
-    `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run
-    concurrently).
+    uses (`sc.maxNumConcurrentTasks` — for the group's resource profile, the number of task slots
+    across active executors, i.e. cores divided by cores-per-task), not `spark.default.parallelism`
+    (a default partition count, unrelated to how many tasks can run concurrently).
 - **Single resource profile per group (v1).** A resource profile is the executor/task resource
   requirement (cores, memory, GPUs, ...) a stage runs under; the number of concurrent slots is
   defined *per profile* (a cluster may run many concurrent CPU tasks but few GPU tasks). Comparing
@@ -126,21 +130,24 @@ materialize-before-read sequencing.
   follow-up, not needed for the streaming shapes whose members share the default profile.
 - **Co-residency.** Once admitted, all member stages of a group are simultaneously running.
 - **Single ownership and 1:1 (v1).** In v1 a pipelined producer feeds exactly one consumer and
-  belongs to exactly one job. These are two separate restrictions, and they are stated at the level
-  of the construct itself, not any particular caller (the pipelined dependency and PG are generic
-  `DAGScheduler` constructs that may be depended on directly, not only from SQL/RTM):
+  belongs to exactly one job. These are two separate restrictions. They follow from what a pipelined
+  shuffle *is* (a transient edge), not from how any particular caller builds its DAG — the pipelined
+  dependency and PG are generic `DAGScheduler` constructs that code can depend on directly, not only
+  through SQL / real-time mode:
   - *No cross-job / cross-time reuse — intrinsic.* A pipelined shuffle is transient: a once-through
-    live stream with no retained, addressable output. Unlike a regular shuffle, there is nothing for
-    a second job to reuse, so reuse across jobs is unsound for it. This must be enforced explicitly —
-    from the scheduler's perspective a shuffle-map stage *can* be reused unless something forbids it —
-    and at **both** layers by which the scheduler would otherwise reuse a shuffle: the
-    `shuffleIdToMapStage` stage-object reuse is bypassed (each pipelined dependency gets a fresh
-    producer stage), and the `MapOutputTracker`-availability reuse is prevented (a pipelined shuffle's
-    tracker registration does not outlive its group, so `getMissingParentStages` cannot skip a
-    re-created producer as already-available). Bypassing only the former is insufficient. A pipelined
-    producer whose stage carries more than one jobId is rejected (fail-fast, §9). (A producer may
-    separately emit a durable *materialized* output edge — a distinct regular `ShuffleDependency` that
-    reuses normally; run-once binds only the transient edge.)
+    live stream with no retained, addressable output. So, unlike a regular shuffle, there is nothing
+    for a second job to reuse — reuse across jobs is unsound for it.
+    - This must be enforced, not assumed: from the scheduler's perspective a shuffle-map stage *can*
+      be reused unless something forbids it. Spark would otherwise reuse a shuffle in two ways, and
+      **both** must be blocked. (1) Stage-object reuse via `shuffleIdToMapStage`: bypass it, so each
+      pipelined dependency gets a fresh producer stage. (2) Output-availability reuse: a re-created
+      producer would be skipped as already-done because its outputs are still registered in
+      `MapOutputTracker` (`getMissingParentStages` treats a registered stage as available), so a
+      pipelined shuffle's tracker registration must not outlive its group. Blocking only (1) is not
+      enough.
+    - A pipelined producer whose stage carries more than one jobId is rejected (fail-fast, §9).
+    - (A producer may separately write a durable *materialized* output edge — a distinct regular
+      `ShuffleDependency` that reuses normally. Run-once binds only the transient edge.)
   - *1:1 within the group (v1) — deferred, not intrinsic.* v1 rejects a pipelined producer with more
     than one consuming stage, but 1:N fan-out is a supported model that v1 defers rather than an
     incompatibility (§9): co-scheduled consumers would read the live stream via multicast, and
@@ -167,10 +174,10 @@ slots.
   scheduler does not retry a failed admission — it fails the job, and retry is the caller's decision
   (for a streaming query, the batch-execution loop restarting the batch with its own backoff). This
   is adequate only for a simple `prefix* -> PG -> suffix*` shape run by a caller that has its own
-  restart loop: caller-retry's unit is the whole job, so with concurrent work (sibling stages, other
-  PGs) or an already-committed prefix it discards work it cannot selectively preserve, and a
-  directly-depended-on PG may have no retrying caller at all — so a transient slot shortfall either
-  kills the job or forces over-provisioning.
+  restart loop. Its unit is the whole job, so it has two weaknesses: with concurrent work (sibling
+  stages, other PGs) or an already-committed prefix, restarting the job discards work it cannot
+  selectively preserve; and a directly-depended-on PG may have no retrying caller at all. The result
+  is that a transient slot shortfall either kills the job or forces over-provisioning.
   - *(Refinement, post-v1.)* The scheduler retries **admission of the PG specifically** — hold the
     group, re-run the gang slot-check on a timer, admit when it fits, fail after a bounded number of
     attempts — leaving any completed prefix and concurrent stages untouched. This mirrors barrier's
@@ -343,8 +350,8 @@ than mis-scheduling it; it is expected to be supported later.
 | Speculative execution | incompatible | A speculative producer copy races a consumer already reading partial output; no commit barrier protects the read. |
 | Push-based shuffle merge as a pipelined (incremental) shuffle | incompatible | Push-based merge exposes output only after a post-completion "finalize" step — the opposite of incremental reads — so it cannot serve as the incremental shuffle within a PG. (A PG's regular input/output edges may still use push-based merge, like any regular shuffle — §7.) |
 | Pipelined producer shared across jobs (cross-time reuse) | incompatible | A transient incremental edge is a live, once-through stream with no retained, addressable output, so a consumer running at a different time has nothing to fetch — cross-time / cross-job reuse is intrinsically undefined for it. Enforced by never binding a pipelined producer's stage to more than one job, at both the `shuffleIdToMapStage` and `MapOutputTracker` reuse layers (§4). |
-| Pipelined producer with more than one co-scheduled consumer (fan-out / branching) | deferred | 1:N is a supported model, deferred in v1. A consumer's transport is picked by group membership: an in-group consumer reads the incremental edge (needs multicast — one producer pushing each record to N live reader channels — plus per-consumer backpressure, not yet built); an out-of-group consumer reads a materialized edge the producer also writes (a tee), sequenced after group commit like any regular shuffle out (§7). v1 rejects a pipelined producer with more than one consuming stage fail-fast until multicast lands. |
-| Reliable RDD checkpoint in a member's within-stage chain | incompatible | Reliable `RDD.checkpoint()` writes a durable, lineage-truncated snapshot, which both reintroduces cross-time reuse of a transient edge (§4) and requires a post-success recompute of the member's now-vanished transient input. Rejected at group creation by walking the within-stage chain and failing on any RDD whose `checkpointData` is a `ReliableRDDCheckpointData` (keyed on `checkpointData`, not `isCheckpointed`, since the write has not happened yet). Cache / `.persist()` / local checkpoint are whole-partition and ephemeral, and are not rejected. |
+| Pipelined producer with more than one co-scheduled consumer (fan-out / branching) | deferred | 1:N (one producer, many consumers) is a supported model, just not built in v1. How each consumer reads depends on whether it is in the group: an in-group consumer would read the live stream, which requires the producer to *multicast* — send each record to all N live readers at once — and to slow down for a reader that falls behind (backpressure); neither is built yet. An out-of-group consumer instead reads a normal materialized copy the producer also writes, which it consumes after the group finishes like any regular shuffle output (§7). Until multicast exists, v1 rejects a pipelined producer with more than one consuming stage. |
+| Reliable RDD checkpoint in a member's within-stage chain | incompatible | Reliable `RDD.checkpoint()` writes the RDD to durable storage and cuts its lineage. Two problems: (1) that durable snapshot can be read by a later job, reintroducing the cross-time reuse a transient edge forbids (§4); and (2) the checkpoint is written by a *separate job that runs after the main one succeeds and recomputes the RDD from scratch* — but a member's input is the transient pipelined shuffle, which is already gone by then, so the recompute cannot read it. Rejected at group creation by walking the within-stage chain and failing on any RDD whose `checkpointData` is a `ReliableRDDCheckpointData` (checked via `checkpointData`, not `isCheckpointed`, since the write has not happened yet). Cache / `.persist()` / local checkpoint keep whole partitions in ephemeral storage and are safe — not rejected. |
 | Members with differing resource profiles | incompatible | The gang slot check compares one demand against one capacity (§4); v1 requires a group to be single-profile. Per-profile accounting is a follow-up. |
 | Adaptive Query Execution over a pipelined exchange | incompatible | AQE reshapes exchanges from complete map-output statistics, which are unavailable while the shuffle is read incrementally. Enforced where exchanges are marked pipelined. |
 | Regular shuffle internal to a group (should not occur) | invariant | Split by construction into separate groups (§3); checked and rejected as an inconsistency if it ever arises (§7). |

From d3293af2e05660e7ce7c0eca62e3479c62f5cd27 Mon Sep 17 00:00:00 2001
From: Jerry Peng <jerry.peng@databricks.com>
Date: Mon, 13 Jul 2026 06:09:56 +0000
Subject: [PATCH 10/10] [SPARK-XXXXX][DOC] Generalize the
 MapOutputTracker/executor-loss failure note

The S6 bullet on why a pipelined shuffle's availability is not tracked in
MapOutputTracker: dropped the internal ticket reference and stated the failure
generically (an incremental-shuffle writer blocking on end-of-stream acks from
already-finished reducers). Also clarity edits from an adversarial review pass:
name the subject in the wrong/dangerous sentence, split it into its two reasons,
spell out 'losses while the group is still running', and cash out how this makes
the no-single-stage-resubmit rule robust.

All code references (ShuffleMapStage.isAvailable, getNumAvailableOutputs,
removeOutputsOnExecutor/removeOutputsOnHost, StreamingShuffleOutputTracker) were
verified against the codebase.

Co-authored-by: Isaac
---
 PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md | 27 ++++++++++++++++++++++++---
 1 file changed, 24 insertions(+), 3 deletions(-)

diff --git a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
index 56281bbdb804..d26f881640c0 100644
--- a/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
+++ b/PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md
@@ -142,9 +142,10 @@ materialize-before-read sequencing.
       **both** must be blocked. (1) Stage-object reuse via `shuffleIdToMapStage`: bypass it, so each
       pipelined dependency gets a fresh producer stage. (2) Output-availability reuse: a re-created
       producer would be skipped as already-done because its outputs are still registered in
-      `MapOutputTracker` (`getMissingParentStages` treats a registered stage as available), so a
-      pipelined shuffle's tracker registration must not outlive its group. Blocking only (1) is not
-      enough.
+      `MapOutputTracker` (`getMissingParentStages` treats a registered stage as available). Blocking
+      only (1) is not enough. The fix for (2) is the same one §6 requires for executor loss: a
+      pipelined shuffle is not tracked in `MapOutputTracker` at all — its availability is owned
+      elsewhere — so neither cross-job reuse nor executor-loss removal can act on it.
     - A pipelined producer whose stage carries more than one jobId is rejected (fail-fast, §9).
     - (A producer may separately write a durable *materialized* output edge — a distinct regular
       `ShuffleDependency` that reuses normally. Run-once binds only the transient edge.)
@@ -265,6 +266,26 @@ how Spark treats task events from a stage attempt that is later discarded.
     - *Note this differs from the base scheduler,* which handles a task failure by resubmitting just
       that one stage — the `FetchFailed` -> resubmit path, and retry paths generally — recomputing
       the failed stage in isolation and resuming.
+  - *A pipelined shuffle's availability is not tracked in `MapOutputTracker`.* This is the mechanism
+    that makes "no single-stage resubmit" robust, and it also closes an executor-loss resubmit path.
+    A regular shuffle records its map outputs in `MapOutputTracker`, and `ShuffleMapStage.isAvailable`
+    is derived from them (`getNumAvailableOutputs`); on executor or host loss the scheduler strips
+    those outputs (`removeOutputsOnExecutor` / `removeOutputsOnHost`), which flips `isAvailable` to
+    false and resubmits the producer. For a transient pipelined shuffle, resubmitting the producer is
+    both wrong (its output was never durably stored, so there is nothing to recompute into) and
+    dangerous (the resubmitted producer has no live consumers left to serve and can hang — for
+    example, an incremental-shuffle writer that blocks waiting for end-of-stream acknowledgements from
+    reducers that have already finished). So a pipelined shuffle is *not* registered with
+    `MapOutputTracker`; its map-stage availability is
+    tracked separately — on the `ShuffleMapStage` itself (a monotonic completed-partition set) or a
+    streaming-specific tracker (`StreamingShuffleOutputTracker`) — and `MapOutputTracker` stays
+    streaming-agnostic. Executor/host-loss removal then never touches a pipelined shuffle and never
+    flips its availability, so the producer is never resubmitted from that path; genuine losses that
+    happen while the group is still running are handled by group-atomic teardown above. This is also
+    what makes the "no single-stage resubmit" rule above robust: with no tracked availability to flip,
+    the availability-driven resubmit cannot fire in the first place. (Same
+    `MapOutputTracker`-must-not-govern-a-transient-shuffle point as the reuse layer in §4, seen from
+    the executor-loss side rather than the cross-job-reuse side.)
   - *Teardown is by group membership, not producer availability.* At the end of a batch the producer
     finishes and registers all its map outputs — becoming "available" — while the consumer is still
     draining the remainder, so it is normal, not rare, for the producer to be available while the