[WIP] Stack/pipelined shuffle pr7 rtm - #57613
Open
jerrypeng wants to merge 39 commits into
Open
Conversation
…ask failure A pipelined-group member reads/writes a transient shuffle that cannot be re-read in isolation, so a single member task failure must fail the whole group -- which fails the job, letting the caller (the streaming batch loop) rerun -- rather than be retried per-task. TaskSet gains an isPipelined flag, set by DAGScheduler.isPipelinedGroupMember (true if the stage produces a PipelinedShuffleDependency or reads one at a shuffle boundary). For a pipelined task set, TaskSetManager caps attempts at 1 (effectiveMaxTaskFailures) and counts failures that are normally excluded -- most importantly executor loss not caused by the app, which strands the transient output. It deliberately does NOT count the benign reasons TaskKilled (a losing duplicate attempt after another succeeded, or a deliberate kill) and TaskCommitDenied (a speculation commit race): counting those would abort the group spuriously even though the task effectively succeeded. Recovery itself is not handled here: the job aborts and the streaming layer reruns the batch from the last committed offset with fresh stage ids. Inert for non-pipelined task sets: effectiveMaxTaskFailures and the counting condition both reduce to the original expressions when isPipelined is false. Tests (TaskSetManagerSuite): a pipelined set aborts after one genuine failure; counts executor loss; does NOT abort on TaskKilled or TaskCommitDenied; a non-pipelined set is unchanged (executor loss uncounted, retries apply). (DAGSchedulerSuite): both producer and consumer task sets are marked isPipelined; regular task sets are not. Co-authored-by: Isaac
… producer at both layers A pipelined shuffle is transient: a once-through live stream with no retained, addressable output. Unlike a regular shuffle there is nothing for a later or concurrent job to re-read, so a pipelined producer stage must never be reused. Spark can reuse a shuffle-map stage at two independent layers, and both must be closed for a pipelined shuffle (spec S4): 1. shuffleIdToMapStage stage-object reuse: getOrCreateShuffleMapStage fails fast with PIPELINED_SHUFFLE_CROSS_JOB_REUSE if a cached pipelined producer stage would be bound to a second job. A sequential re-run is unaffected: the earlier job's cleanup drops the stage from shuffleIdToMapStage, so a later job mints a fresh producer bound only to itself (this is how an RTM micro-batch reruns). 2. MapOutputTracker-availability reuse: a pipelined shuffle is never registered with the MapOutputTracker as durable output. ShuffleMapStage tracks its completed partitions locally in a monotonic set; numAvailableOutputs and findMissingPartitions read that set for a pipelined stage. Because executor/ host loss only strips MapOutputTracker entries, it can no longer flip a completed pipelined producer to unavailable and trigger a resubmit. Together these fix SC-235532 (an RTM streaming-shuffle query hangs after executor loss: the mapper is resubmitted and its streaming writer blocks forever in awaitTerminationAcks). Two further resubmit routes for a pipelined producer are closed here: - TaskSetManager.executorLost re-enqueues an already-successful map task via the Resubmitted channel on executor decommission (when getMapOutputLocation returns None, which is always true for an unregistered pipelined shuffle). This bypasses handleFailedTask, so the group-atomic abort would never see it. Guarded with !taskSet.isPipelined; a genuine loss still routes through handleFailedTask and fails the group. - handleTaskCompletion records a pipelined success unconditionally, before the "possibly bogus epoch" check. A straggler success whose executor is already marked lost otherwise removed its partition from pendingPartitions without recording it, leaving the stage "done but not available" and driving a resubmit. (Push-based shuffle merge is disabled for a PipelinedShuffleDependency in the routing change that introduces the incremental manager, since that is where a pipelined dependency would otherwise reach the merge path; see that commit.) Inert for jobs without a pipelined dependency: every branch is gated on the dependency subtype or ShuffleMapStage.isPipelined, and the full existing DAGSchedulerSuite / TaskSetManagerSuite pass unchanged. Co-authored-by: Isaac
…ipelined group on a member FetchFailed (SC-233883) A FetchFailed uniquely bypasses the group-atomic maxTaskFailures=1 lever: the base TaskSetManager marks a FetchFailed task successful, zombies the set, and returns without counting the failure (FetchFailed.countTowardsTaskFailures is false), so the whole recovery happens in the DAGScheduler's FetchFailed handler, which resubmits the map stage in isolation and recomputes serially. For a pipelined group that is a deadlock (SC-233883): the transient pipelined shuffle cannot be re-read, and members are co-scheduled, so a lone-stage resubmit is never valid (spec S6, "single-stage resubmit is disabled for group members; any member failure -- including FetchFailed -- fails the whole group"). Route a FetchFailed whose failed stage or map stage is a pipelined group member to abortStage instead: aborting the failed stage tears down the job's co-scheduled member stages via cancelRunningIndependentStages and fails the job, and the caller (the streaming batch loop) reruns the batch from scratch. The teardown is complete because a v1 pipelined group is exactly one job: the producer is registered to the same job (updateJobIdStageIdMaps walks parents), so cancelRunningIndependentStages kills/finishes it whether it is still running or already finished. This coupling to "group == one job" is guaranteed by the cross-job-reuse fail-fast (PIPELINED_SHUFFLE_CROSS_JOB_REUSE); if fan-out or cross-job sharing is ever enabled, this branch must switch to tearing down pipelinedGroupOf(failedStage) explicitly rather than relying on job-atomicity. Inert for jobs without a pipelined dependency: isPipelinedGroupMember returns false for every regular stage (it is a pipelined producer only if its shuffleDep is a PipelinedShuffleDependency, and a consumer only if it reads through one), so the original FetchFailed handling runs unchanged. The full existing DAGSchedulerSuite (including all FetchFailed / shuffle-merger tests) passes. Tests: a member FetchFailed (consumer reading the pipelined edge) fails the job with no single-stage resubmit; a producer FetchFailed reading its regular input (spec S7) also fails the group. Co-authored-by: Isaac
…mit authorization Spec S5 requires that when a pipelined group reruns after a failure, a result stage whose tasks already committed can commit again -- OutputCommitCoordinator otherwise permanently denies re-commit for a committed partition (keyed by stage id; a Success clears nothing). This is already satisfied by M1's existing mechanisms, and this test locks that in: - Group teardown runs the committed result stage through markStageAsFinished -> outputCommitCoordinator.stageEnd, clearing its committer state, so no stale authorization survives the failed attempt (option (b) in S5). - The caller's rerun is a NEW job whose stages get fresh stage ids, so the coordinator has no prior committer for them regardless (option (a) in S5). The test commits a result task in a pipelined group (registering committer state), fails the group via its producer, asserts the coordinator's committer state is cleared on teardown, then reruns the batch as a new job and asserts the rerun's result stage gets a fresh stage id and completes -- re-committing its partitions. No production change: M1's group-atomic-failure + caller-reruns-the-batch design already conforms to S5. This adds the regression test that proves it. Co-authored-by: Isaac
…or-all-PG restriction The M1 all-regular-or-all-PG restriction (PR3) rejects any job with a regular shuffle in a pipelined job up front, including a regular-shuffle prefix feeding a pipelined producer (regularRoot --regular--> producer(pipelined) --pipelined--> consumer). Three tests exercised that now-rejected shape: - "a consumer is a group member even if a REGULAR dep precedes the pipelined one" - "a FetchFailed on a pipelined producer reading its regular input (SC-233883)" - "a group-aborting FetchFailed still unregisters the dead executor's outputs" Remove them: the shape is out of the RTM scope (scan-of-files --pipelined--> stateful, no upstream shuffle). SC-233883's core behavior -- a member FetchFailed fails the whole group rather than resubmitting a single stage -- remains covered by the pure-shape test "a FetchFailed on a group member fails the group, not a single-stage resubmit (SC-233883)". Add a rejection test documenting that a regular-shuffle prefix feeding a pipelined producer is rejected up front. Co-authored-by: Isaac
…test non-vacuous; fix a spec citation
From an adversarial review:
- The SC-235532 executor-loss regression test completed the producer with
makeMapStatus("hostA-exec"), which registers the output under executor id
"hostA-exec-exec" (makeBlockManagerId appends "-exec"), while the test loses
executor "hostA-exec" -- so removeOutputsOnExecutor never matched and the test
passed even with the ShuffleMapStage pipelined-availability fix reverted (it
did not guard its named regression). Use makeMapStatus("hostA") so the lost
executor id matches the registered output. Verified: the test now FAILS when
the fix is reverted.
- Fix a spec citation in the FetchFailed group-abort comment: "registers no map
outputs in the tracker" is a spec-S4 (transient / no durable output) fact,
cited as S4 everywhere else; it was mistakenly attributed to S6 (the
group-atomic failure section).
Co-authored-by: Isaac
… descriptive prose Comments and test names referenced internal Databricks Jira tickets (SC-235532, SC-233883) and internal milestone tags (M1.2 / M1.4 / M1.6 / M1.8), which are meaningless to the Apache community. Replace each with a short description of the behavior it denotes (e.g. the streaming-writer resubmit hang, the group-atomic abort), keeping the technical content and dropping the opaque identifiers. No behavior change; comment/test-name only. Co-authored-by: Isaac
…eap per-job flag isPipelinedGroupMember(stage), used to tag TaskSet.isPipelined on every submitMissingTasks, walks the stage's RDD graph -- pure overhead for a regular job that has no pipelined dependency. Record whether a job uses a PipelinedShuffleDependency once at submission (ActiveJob.hasPipelinedDependency, from the hasPipelined already computed in handleJobSubmitted) and short-circuit the walk when it is false. No behavior change for a pipelined job (the flag is true, so the membership check runs exactly as before); regular jobs now skip the walk entirely. Co-authored-by: Isaac
…er entry is registered but stays empty The availability comment said a pipelined shuffle is "NOT registered with the MapOutputTracker", which could read as contradicting createShuffleMapStage (it does call registerShuffle for every ShuffleDependency, pipelined included). Clarify the precise behavior: the empty shuffle-status entry is registered to keep containsShuffle / unregisterShuffle bookkeeping uniform, but no map output is ever populated into it (registerMapOutput runs only on the non-pipelined path), so availability is read from the stage's local monotonic set instead. Comment-only. Co-authored-by: Isaac
…rd; reconcile a ShuffleMapStage comment
- Regression test for the finishOnly TaskEnd guard (fixed on the scheduling PR): a deferred
consumer's TaskEnd fires inline, the group is then aborted via a member FetchFailed (which
removes the consumer stage), and a finishOnly replay is delivered onto the torn-down stage.
Asserts the replay posts no additional TaskEnd. Lives here because it needs the
group-atomic FetchFailed abort. Reverting the guard makes it fail (TaskEnd count +1).
- Reconcile numAvailableOutputs's inline comment ("not tracked in the MapOutputTracker")
with the field-level comment (which correctly says the entry is registered but never
populated) -- both now say "outputs never populated / entry stays empty".
Co-authored-by: Isaac
…red consumer deferral Deferral-interaction coverage. The deferral buffer (dependentStageMap) is the only mutable state this feature adds and must never outlive its job. This drives the explicit job-cancellation path (JobCancelled -> handleJobCancellation -> failJobAndIndependent- Stages), which was not directly exercised against a buffered deferral, and asserts the entry is fully cleaned up (map empty) with no buffered success applied as a result. Verified non-vacuous: neutering BOTH cleanup mechanisms -- the release-drop in cancelRunningIndependentStages (primary, fires while the producer is running) and the cleanupStateForJobAndIndependentStages sweep (backstop) -- makes the test fail with a leaked deferral entry. Co-authored-by: Isaac
…whole pipelined group Defense-in-depth for the group-atomic failure model at the DAGScheduler layer. The maxTaskFailures=1 mechanism is covered at the TaskSetManager layer, and producer-failure teardown is covered by the deferral-drop test, but there was no DAGScheduler-level test that a CONSUMER (result) task failing tears down a still-running co-scheduled producer. This adds one: it leaves the producer running, asserts it is running, fails the consumer task set (as a maxTaskFailures=1 abort surfaces), then asserts the running producer is torn down and the job fails. Revert-checked: neutering the cancelRunningIndependentStages teardown makes it fail (producer left running). Co-authored-by: Isaac
…ments Strip internal shorthand that a reader of merged Apache Spark cannot resolve -- version/milestone tags, design-doc spec section numbers, "RTM"/"PG" abbreviations, and fine-grained/coarse model terminology -- from the pipelined-shuffle failure-path comments and test comments, rewording to plain behavioral language. Comment and assertion-string text only; no behavioral change. Co-authored-by: Isaac
…n-stage test The scheduling PR now defers a pipelined consumer's whole completion event (coarse model) instead of running its per-task effects inline and replaying a finishOnly-flagged event. The finishOnly flag and its torn-down-stage guard are gone, so the regression test that exercised "a finishOnly replay landing on a torn-down stage does not re-post TaskEnd" no longer describes a reachable path: there is no inline TaskEnd and no separate finishOnly replay to land on a torn-down stage. A buffered completion is either fully replayed (producer succeeded) or dropped with its TaskEnd flushed (producer failed) -- both already covered by "deferred consumer completions are dropped when the producer fails". Remove the obsolete test. The unrelated ShuffleMapStage numAvailableOutputs comment fix from the same original commit is preserved. Co-authored-by: Isaac
Reject, with a clear PIPELINED_SHUFFLE_UNSUPPORTED error, the pipelined-shuffle idioms a group cannot support in v1 (spec S9), so a misuse fails fast rather than silently mis-scheduling, hanging, or corrupting a group. Producer-side checks, in checkPipelinedProducerSupported (called from createShuffleMapStage when the shuffle is a PipelinedShuffleDependency), so a rejection throws during stage creation and leaves no partial scheduler state: - Barrier execution in a member stage (exposes output only after a global sync). - Dynamic resource allocation (gang admission needs a stable slot set). - Statically-indeterminate producer (its stage rollback-and-recompute recovery is moot under group-atomic failure -- a group never recomputes a single stage). - Checksum-mismatch full retry (the runtime counterpart to static indeterminism; also moot). Defensive: a PipelinedShuffleDependency does not enable it. - Push-based shuffle merge as the pipelined shuffle (finalize-then-read is the opposite of incremental reads). Defensive backstop: the dependency disables merge in its constructor. - A reliable RDD checkpoint in the member's within-stage chain (durable, lineage- truncated snapshot -> reintroduces cross-time reuse and needs a post-success recompute of the vanished transient input). Keyed on checkpointData being a ReliableRDDCheckpointData, not isCheckpointed (the write has not happened yet). Group-level check, in checkPipelinedGroupsSupportedInRDDGraph (called from handleJobSubmitted before any stage is created, alongside the speculation check, so a rejection fails the job up front with no partial state): - Fan-out: a pipelined producer feeding more than one consumer. 1:N is a supported model deferred to a later version (it needs multicast to N live readers); v1 rejects it. This also closes a group-demand undercount in the slot check when a sibling consumer stage is not yet created. Speculation and cross-job reuse (also S9) are rejected by earlier commits. The remaining S9 group-level rows -- mixed resource profiles and a regular shuffle internal to a group -- are structural invariants that do not arise for the single-profile, prefix*->PG->suffix* shapes v1 targets (a group is single-profile by construction, and groups split at regular-shuffle boundaries); they are deferred to the milestone that makes group formation first-class. Tests: barrier, indeterminate, reliable-checkpoint, and fan-out producers/groups are each rejected; regular (non-pipelined) shuffles using those same idioms are NOT rejected (inertness). Co-authored-by: Isaac
…R chain; widen group-check catch Follow-up from the M1.7 adversarial review: 1. (MEDIUM) The reliable-RDD-checkpoint rejection was scoped to a pipelined PRODUCER's within-stage chain only, but spec S9 covers any group MEMBER's chain -- including a CONSUMER, whose transient input IS the pipelined shuffle, so a reliable checkpoint there would re-read the vanished stream on recompute. Moved the reliable-checkpoint detection into checkPipelinedGroupsSupportedInRDDGraph, which runs at job submission over the whole RDD graph before any stage is created: it now walks both the producer chain (rooted at the produced RDD) and each consumer chain (rooted at a reading RDD). This also removes a partial-state leak the previous stage-creation-time consumer check would have caused (a consumer-stage throw after the producer stage had already registered). 2. (LOW) The group-check was in its own try with a SparkException-only catch, so an incidental non-Spark exception from the RDD-graph walk could escape handleJobSubmitted's job-level handler (with speculation disabled the walk is the first dependency traversal). Moved the call inside the existing createResultStage try, so any exception is handled by the same case e: Exception => jobFailed path. Not changed: fan-out keyed on shuffleId (a third, LOW review note) is correct as-is -- two distinct PipelinedShuffleDependency instances on one producer RDD are two independent 1:1 transient streams, not the multicast fan-out hazard; the check matches the spec's "more than one consumer for the same pipelined shuffle". Tests: a reliable checkpoint in a CONSUMER's chain is now rejected (new test); the producer-chain and all other rejection/inertness tests still pass. Co-authored-by: Isaac
…esses when a producer fails Regression coverage for a review concern: a pipelined consumer with more than one pipelined producer (a join) must DROP its buffered successes when any producer fails, never replay them via a still-pending sibling. On the full stack a member failure is group-atomic (abortStage tears down all members and marks each producer failed), so releaseDeferredPipelinedConsumers sees producerFailed=true for every producer and drops -- the "sibling succeeds -> replay" trigger cannot occur. This test pins that behavior (a producer feeding two consumers would be fan-out and is rejected; two producers feeding one consumer is a supported all-PG join). Co-authored-by: Isaac
…ed doc on where group idioms are checked The scaladoc said group-level idioms (fan-out, mixed resource profiles, internal regular shuffle) are "checked separately where the group is co-scheduled". That is inaccurate: fan-out is rejected up front at job submission by checkPipelinedGroupsSupportedInRDDGraph (before any stage is created), not at co-scheduling time; and mixed resource profiles / an internal regular shuffle are structural invariants that do not arise for v1's single-profile all-pipelined shape and are not separately checked at all. Correct the doc to point at the right place and stop implying checks that do not exist. Co-authored-by: Isaac
… so a group abort cannot leak the stage Follow-up to the fine-grained deferral change (which lives on the scheduling PR): update the abort-path test that previously guarded the coarse model's re-emit-on-teardown behavior. Under the fine-grained model a deferred consumer's TaskEnd is posted in real time when the task finishes -- only the stage/job-completion bookkeeping is deferred -- so the "leak the stage as perpetually running" failure the coarse model had to patch at teardown cannot arise. This test exercises the group-atomic FetchFailed->abort path (which this PR introduces), so it belongs here rather than on the scheduling PR. The test now asserts the consumer's TaskEnd is delivered inline (before the producer's fate is known), and that on the subsequent group abort the deferred completion is dropped (its result is never applied) with no duplicate TaskEnd re-emitted. Co-authored-by: Isaac
… reject mixed resource profiles (S9) Two review items on the pipelined-group fail-fast path: - Typed exception (was: brittle string match). handleJobSubmitted distinguished an up-front idiom rejection from a stage-creation failure by matching on e.getCondition == "PIPELINED_SHUFFLE_UNSUPPORTED" -- a rename or a wrapped cause would silently fall through to the generic "Creating new stage failed" branch and mis-report. Introduce a marker type PipelinedShuffleUnsupportedException (carrying the same error class, so the user-facing message is unchanged) and match on the TYPE. - Mixed-resource-profile fail-fast (spec S9). The gang slot check compares one demand against one profile's capacity (maxNumConcurrentTasks is per profile), so v1 requires a single-profile group. This was asserted in comments as a "structural invariant" but nothing enforced it. checkPipelinedGroupsSupportedInRDDGraph now collects the distinct explicit resource profiles across the group's RDDs in its existing single graph walk and rejects a group spanning more than one. Per-profile accounting remains a follow-up. Updated the two scaladocs that claimed mixed profiles "do not arise / are not checked". Also drops a stray internal milestone reference from a comment. Tests: a new test submits a producer/consumer group with two distinct resource profiles and asserts the up-front rejection; neutering the check makes it fail (non-vacuous). The existing idiom-rejection tests (barrier, indeterminate, checkpoint, fan-out) still pass through the typed-exception catch. Full DAGSchedulerSuite green (193). Co-authored-by: Isaac
…lined group; gate FetchFailed group check
Round-3 review follow-ups:
- Resource-profile admission gap (real). The mixed-profile check collected only distinct
EXPLICIT profiles (getResourceProfile() is null when unset) and rejected on count > 1,
while the slot check measures capacity against the DEFAULT profile. So a group running
uniformly on one non-default profile, or a non-default member mixed with default members,
passed the check yet was admitted against the wrong (default) capacity pool -- the exact
partial-residency deadlock gang admission prevents. Reject if ANY member carries an
explicit non-default profile: v1 requires the whole group on the default profile (spec
S9); per-profile accounting remains a follow-up. Added tests for the two previously-missed
shapes (uniform non-default; non-default + default), alongside the two-distinct case.
- FetchFailed group-membership check now gated on ActiveJob.hasPipelinedDependency, matching
the submitMissingTasks site, so a regular job's FetchFailed no longer walks the stage RDD
graph.
- Dropped an internal-review artifact ("Isaac-review probe") from a test comment.
Co-authored-by: Isaac
…dup a within-stage walk; warn when slot check is off Architecture-review follow-ups (no behavior change on the happy path): - Make the "job == one pipelined group, pre-admitted" invariant explicit and enforced. The submitStage co-schedule branch gang-schedules a group's members with no slot check, trusting that handleJobSubmitted already gang-admitted the whole job -- previously encoded only as a comment. Record ActiveJob.pipelinedGroupAdmitted on the admitted (or check-disabled) path and assert it at the co-schedule branch. This is a tripwire for a future version that relaxes job==group (multiple groups per job): it must move gang admission to a per-group check here and replace the assert, not delete it, or an unadmitted group could be gang-scheduled and deadlock. Verified the assert fires if admission is not recorded, and holds across all existing pipelined tests. - Collapse a duplicated within-stage walk: isPipelinedGroupMember's consumer walk was byte-identical to rddChainReadsPipelinedShuffle (differing only in the root RDD). Delegate to the latter as the single source of truth so a change to "what ends a within-stage chain" need not be made in two places. Behavior-preserving. - Warn (once) when spark.scheduler.pipelinedGroup.slotCheck.enabled=false, since disabling the only gang-admission deadlock check is safe only with out-of-band capacity reservation. Once-per-scheduler (event-loop-confined var) so it does not spam per submitted batch. Co-authored-by: Isaac
…ed, not just an assert The co-schedule branch verified the up-front gang-admission invariant with a bare `assert`, which is elided under -da (JVM assertions off in production). In v1 the invariant always holds, so this is only a forward tripwire -- but if a later version relaxes job==group and forgets to move gang admission to a per-group check here, a production build would silently gang-schedule an unadmitted group and could deadlock. Replace the assert with a fail-closed abortStage (reusing the same terminal jobFailed path as the sibling no-active-job case), so the invariant is enforced even under -da: a violation becomes a clean, rerunnable job failure instead of a silent deadlock. Verified: the guard never fires in normal v1 operation (all pipelined tests green); neutering admission recording makes it abort the group rather than co-schedule. Co-authored-by: Isaac
…ments Strip internal shorthand that a reader of merged Apache Spark cannot resolve -- version/milestone tags, design-doc spec section numbers, "all-PG" abbreviation, and fine-grained/coarse model terminology -- from the pipelined-shuffle fail-fast comments and test comments, rewording to plain behavioral language. Comment and assertion-string text only; no behavioral change. Co-authored-by: Isaac
… group-check version An earlier PR in this stack rejects a pipelined job whose member uses a non-default resource profile from rejectUnadmittablePipelinedGroup (a plain SparkException). This PR already performs the same rejection in checkPipelinedGroupsSupportedInRDDGraph, as a typed PipelinedShuffleUnsupported error alongside the fan-out and reliable-checkpoint group checks, and with broader coverage (uniform-non-default, mixed, and multi-profile shapes). Remove the now-redundant earlier check and its test so the resource-profile rejection lives in exactly one place with the typed error. Co-authored-by: Isaac
…arse completion model The scheduling PR now buffers a pipelined consumer's whole completion event (coarse model) rather than running its TaskEnd inline and deferring only the finish bookkeeping. Update the group-abort test that asserted inline TaskEnd timing: - Before: "a deferred consumer TaskEnd fires inline, so an abort cannot leak the stage" -- asserted the buffered success's TaskEnd was emitted immediately, before the abort. - After: "a buffered consumer success is dropped on group abort, and its TaskEnd is flushed so the stage is not leaked as running" -- asserts no TaskEnd while the producer runs (the whole event is buffered), then that the FetchFailed group-abort's drop path flushes the buffered success's TaskEnd (so active-task-tracking listeners still see the task finish) while its result is dropped. The scenario (buffered success + sibling FetchFailed group-abort) and its no-leak guarantee are unchanged; only the event timing differs. Reverting the drop-path flush (events.foreach(postTaskEnd)) makes the test fail on the flush assertion. Co-authored-by: Isaac
…ned so it cannot reject a regular job checkPipelinedGroupsSupportedInRDDGraph walks the whole RDD graph and rejects three group-level idioms: fan-out, a member on a non-default resource profile, and a reliable checkpoint in a member stage. The fan-out and checkpoint checks are keyed on a PipelinedShuffleDependency being present (consumersByShuffleId / producerRoots), so they are inert for a job with no pipelined dependency. The resource-profile check is NOT so keyed -- it fires for ANY RDD whose explicit profile is non-default. But handleJobSubmitted called checkPipelinedGroupsSupportedInRDDGraph unconditionally (the hasPipelined gates guarded only the mixed-shuffle and slot-admission checks). So an ordinary job with no pipelined dependency that merely attaches a non-default profile via RDD.withResources -- a GA stage-level-scheduling feature -- was rejected up front with PIPELINED_SHUFFLE_UNSUPPORTED "a pipelined group member with a non-default resource profile". That job runs fine on master; this was a regression wherever stage-level scheduling is used (K8s / YARN / Standalone with dynamic allocation). Fix: gate the whole checkPipelinedGroupsSupportedInRDDGraph call on hasPipelined, matching the two admission checks just above it. Every idiom it rejects concerns a pipelined group, so it must not run for a job that has no pipelined dependency. This changes behavior only for the previously mis-firing resource-profile branch (the other two were already inert); the three pipelined-group rejection tests still reject. Test: a regular job on a non-default resource profile (RDD.withResources, no pipelined dependency) must NOT be rejected and must run to completion. Reverting the gate makes it fail with PIPELINED_SHUFFLE_UNSUPPORTED. Full DAGSchedulerSuite green. Co-authored-by: Isaac
…st name The test name "regular job on a non-default resource profile is NOT rejected (RP check is pipelined-only)" was 102 chars, over the 100-char scalastyle limit for test sources (core/Test/scalastyle). Shorten the parenthetical. Co-authored-by: Isaac
…rt the rejection this stack produces The diamond and wide-fan-out tests were written on the scheduling PR to prove the admission demand counts a reused pipelined producer once per shuffle id, not once per consumer edge. On that PR's base a fan-out group is admittable, so they asserted admission. This PR rejects fan-out outright (checkPipelinedGroupsSupportedInRDDGraph), so the admission assertions no longer hold once the two are stacked. Rework them to assert the outcome this stack actually produces while still proving the demand dedup. The up-front slot admission (rejectUnadmittablePipelinedGroup) runs before the fan-out idiom check, so at capacity 4 a correctly-deduped diamond (producer 2 + result 2 = 4) passes the slot check and is then rejected for fan-out (PIPELINED_SHUFFLE_UNSUPPORTED); a per-edge-counted group (6, or 8 for the wide fan-out) would be rejected earlier for CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT. The tests now assert the fan-out rejection AND that it is not a slot rejection, so a dedup regression flips the error class and fails them (revert-checked against per-edge counting). The separate "two distinct shuffles over one producer RDD" test is unchanged (it guards shuffle-id-vs-RDD dedup keying, not fan-out). Co-authored-by: Isaac
…utput tracker A pipelined shuffle streams records from a live producer stage to a co-scheduled consumer stage, so the consumer must look up its producer tasks' locations while both stages run. That routing lives in the StreamingShuffleOutputTracker, but createShuffleMapStage only registered the shuffle with the MapOutputTracker (which, for a transient pipelined shuffle, tracks an empty never-materialized output). As a result a consumer task querying for writer locations found the shuffle unregistered and failed. Register a PipelinedShuffleDependency's shuffle with the StreamingShuffleOutputTracker inside the same once-per-shuffleId guard as the MapOutputTracker registration. Inert for a regular shuffle (and when no streaming tracker is configured). Note: this is a latent gap in the co-scheduling layer independent of RTM; it belongs with the pipelined-scheduling change earlier in the stack and should be re-homed there (with a real-transport regression test) when the stack is next reorganized. Co-authored-by: Isaac
…dependency Add a `pipelined` flag to ShuffleExchangeExec. When set, prepareShuffleDependency constructs a PipelinedShuffleDependency instead of a regular ShuffleDependency, so the DAGScheduler co-schedules the exchange's producer and consumer stages and routes the shuffle to the streaming shuffle manager. The flag defaults to false, so every existing shuffle is unchanged. Modeling this as a node field (rather than out-of-band metadata) keeps the pipelined/non-pipelined distinction part of the exchange's identity, so it participates in exchange-reuse and canonicalization -- correct for a transient shuffle that must not be reused as a durable one. Adding the field shifts ShuffleExchangeExec's positional arity, so update the pattern matches that destructure it positionally (EnsureRequirements, AQEUtils, InsertSortForLimitAndOffset and several test suites) to account for the new field; EnsureRequirements' repartition rewrite now uses copy() so the flag is preserved. Co-authored-by: Isaac
…d shuffle Enable a stateful streaming query (starting with deduplication) to run in Real-Time Mode by pipelining its repartition shuffle: - IncrementalExecution adds a Real-Time Mode preparation rule that marks every streaming shuffle exchange pipelined (ShuffleExchangeExec.pipelined = true). Real-Time Mode is detected structurally by a RealTimeStreamScanExec leaf. The producer (source scan) and consumer (stateful operator) stages are then co-scheduled and stream records through a transient shuffle instead of the consumer waiting for the producer to fully materialize. Inert for a non-RTM batch, so the ordinary microbatch path is unchanged. (AQE is already disabled for RTM batches upstream, so the pipelined shuffle is only ever submitted as a result job, never an AQE map-stage job.) - RealTimeModeAllowlist admits the operators a streaming dropDuplicates plans into (StreamingDeduplicateExec, StateStoreRestoreExec, StateStoreSaveExec) plus ShuffleExchangeExec, so the query is no longer rejected before it runs. Co-authored-by: Isaac
…e over a pipelined shuffle Add an end-to-end test: a streaming dropDuplicates query in Real-Time Mode whose repartition shuffle is pipelined. It asserts correct dedup output across two batches, that every shuffle exchange in the executed plan is pipelined, and that the producer and consumer stages are co-scheduled (>= 2 stages running at once). The Real-Time Mode operator allowlist check is left enabled, so the test also covers admission of the dedup plan's operators. Co-authored-by: Isaac
… over a pipelined shuffle Add a fault-tolerance case to the pipelined-shuffle Real-Time Mode suite: a stateful dropDuplicates query whose repartition shuffle is pipelined, that hits a task failure mid-batch and then restarts from checkpoint. It asserts the query fails on the task error (Real-Time Mode does not retry tasks; a member failure aborts the whole co-scheduled group), and that on restart the deduplication state recovered from the last committed batch -- already-seen keys are not re-emitted. Switches the suite to the manual-clock base so batch boundaries are deterministic across the failure and restart, and uses CheckAnswerWithTimeout (which polls the sink in real time) with advanceRealTimeClock to drive Real-Time Mode batches. Co-authored-by: Isaac
… in Real-Time Mode Add a second fault-tolerance case: a dedup query over a pipelined shuffle whose commit log write is fault-injected (via FailureInjectionCheckpointFileManager) so a batch cannot commit and the query fails. On restart from the same checkpoint, the committed batch's deduplication state survives and the uncommitted batch is re-run, so its data is still emitted. The injected close() failure surfaces as the underlying IOException in this path (the COMMIT_LOG_WRITE_FAILURE error-class wrapping is not open source), so the test expects that. Co-authored-by: Isaac
… output tracker on cleanup A pipelined shuffle is registered with the driver-only StreamingShuffleOutputTracker in DAGScheduler.createShuffleMapStage, alongside its MapOutputTracker registration, but nothing ever removed it. Over a long-running Real-Time Mode query -- which allocates a fresh shuffle id per micro-batch -- the tracker's shuffleInfos map grows without bound, and a re-registration of a reused id would hit the "registered twice" guard. Unregister it in ContextCleaner.doCleanupShuffle, right beside the existing mapOutputTrackerMaster.unregisterShuffle. The streaming tracker's state is driver-only (the worker tracker caches nothing and just RPCs the master), so a direct driver-side call is the correct counterpart -- unlike the MapOutputTracker, there is no per-executor cache to invalidate, so this does not belong in the RemoveShuffle RPC handler. The call is guarded by a new containsShuffle predicate so regular (non-pipelined) shuffles, which are never registered here, are skipped without emitting a spurious "not registered" warning on every shuffle cleanup. Also correct the StreamingShuffleManager.unregisterShuffle comment, which previously claimed the tracker was unregistered in BlockManagerStorageEndpoint's RemoveShuffle handler -- untrue in OSS (that path routes to this no-op manager method). Tests: a containsShuffle unit test, plus two ContextCleanerSuite integration tests (a pipelined shuffle is unregistered on cleanup; a regular shuffle is left untouched and quiet). Both integration tests were revert-checked -- each fails when its respective production change (the unregister call, or the containsShuffle guard) is removed. Co-authored-by: Isaac
…putTracker A pipelined shuffle produces no durable, addressable map output: its producer/consumer task-location routing lives entirely in the StreamingShuffleOutputTracker, its availability is tracked on the stage (pipelinedCompletedPartitions), and its reader locates writers through the streaming tracker. Yet createShuffleMapStage still registered an empty ShuffleStatus for it in the MapOutputTracker, purely to keep containsShuffle / unregisterShuffle bookkeeping uniform. That coupled the streaming tracker's lifecycle to MapOutputTracker state: registration and cleanup of the streaming entry were both nested inside MapOutputTracker.containsShuffle guards, so a future divergence between the two trackers could double-register (throwing "registered twice") or leak the streaming entry. Split the two trackers cleanly by dependency type, with no overlap: - createShuffleMapStage: a PipelinedShuffleDependency registers ONLY with the StreamingShuffleOutputTracker (self-guarded on that tracker's own containsShuffle); a regular shuffle registers ONLY with the MapOutputTracker, exactly as before. No empty MapOutputTracker shell is created for a pipelined shuffle. - ContextCleaner.doCleanupShuffle: two independent branches, each keyed on its own tracker's membership -- MapOutputTracker cleanup for a regular shuffle, a direct driver-side StreamingShuffleOutputTracker unregister for a pipelined one (no RemoveShuffle RPC: a streaming shuffle has no block-manager-served files). Neither branch depends on the other tracker's state. This is safe because every MapOutputTracker path a pipelined shuffle could reach is either already type-gated out (output registration, availability, FetchFailed invalidation which routes to a group-atomic abort, push-merge, indeterminate/checksum rollback, map-stage-job stats) or tolerant of the entry's absence (executor/host-loss strips iterate shuffleStatuses and simply skip it). Consumer task locality never reads the tracker for a shuffle dependency (getPreferredLocsInternal recurses only through narrow deps). Tests (ContextCleanerSuite): a pipelined shuffle registered ONLY in the streaming tracker is still cleaned up by doCleanupShuffle even though it is absent from the MapOutputTracker (revert-checked: fails without the streaming cleanup branch); a regular shuffle takes the MapOutputTracker branch and never touches the streaming tracker. Co-authored-by: Isaac
… (before any state mutation) with no streaming tracker, and abort the group on a base-path pipelined FetchFailed Three hardening fixes to "Fully separate pipelined shuffles from the MapOutputTracker", all found by review of that change: 1. createShuffleMapStage registered a pipelined dependency only inside a streamingShuffleOutputTracker.foreach, with the MapOutputTracker registration in the else-if branch. If the streaming tracker were absent (a non-streaming custom incremental manager), the shuffle was registered in NEITHER tracker -- a consumer would then fail with an opaque NullPointerException in the shuffle writer. Fail loud instead: a PipelinedShuffleDependency requires a StreamingShuffleOutputTracker. Crucially, resolve the tracker UP FRONT -- before createShuffleMapStage mutates stageIdToStage / shuffleIdToMapStage / updateJobIdStageIdMaps -- so the fail-loud throw leaves no partial scheduler state. (handleJobSubmitted's catch does no rollback, so a throw after those mutations would leak a half-created stage; a re-submission would then reuse the stale cached stage and fail with a misleading cross-job-reuse error instead of re-throwing the real misconfiguration.) 2. A pipelined-group FetchFailed is normally intercepted by the group-atomic abort branch, whose guard is keyed on the failed stage's active job carrying hasPipelinedDependency. If a pipelined-shuffle FetchFailed reaches the base (non-group-atomic) path instead, that path is invalid for a pipelined shuffle in two ways: (a) its MapOutputTracker invalidation would throw ShuffleStatusNotFoundException (a pipelined shuffle is no longer registered there); and (b) its resubmit branch would enqueue a lone-stage resubmit of the transient producer, which cannot be re-read and would deadlock the group. Guard the whole base-path tail on mapStage.isPipelined and abort the group instead -- matching the group-atomic branch's outcome. Tests (revert-checked): - PipelinedShuffleRoutingSuite: a job with a pipelined dependency and no streaming tracker fails loud with the IllegalStateException, and a re-submission re-throws the SAME error -- proving no partial stage leaked (revert-checked against the throw-after-mutation placement, which instead leaks and yields a cross-job-reuse error on re-submit). - DAGSchedulerSuite: a FetchFailed for a pipelined shuffle reaching the base path (group-atomic guard forced to miss) aborts the group -- the job fails, the throwing MapOutputTracker calls are never invoked, and the transient producer is not resubmitted (revert-checked against both the no-guard and the invalidation-only-guard variants). Co-authored-by: Isaac
…gle-shuffle assumption Two small follow-ups from a review pass: - PipelinedShuffleRoutingSuite: reflow a 101-char comment line (dev/scalastyle flagged it as exceeding 100 characters). Comment-only, no behavior change. - IncrementalExecution.MarkPipelinedShuffleForRealTimeMode: document why marking every ShuffleExchangeExec pipelined is safe today (an RTM plan has exactly one shuffle -- a second is rejected by RealTimeModeAllowlist, see the "repartition not allowed" test / SPARK-54237), and record the latent hazard for when RTM is widened to multi-shuffle plans: transformUp cannot reach a ReusedExchangeExec's wrapped child (a leaf field, not a tree child), so a reused shuffle would stay regular while its twin becomes pipelined -- a mixed regular+pipelined job the scheduler rejects. Note the fix options (mark inside ReusedExchangeExec, run before exchange reuse, or disable reuse for RTM) for that future change. Co-authored-by: Isaac
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Why are the changes needed?
Does this PR introduce any user-facing change?
How was this patch tested?
Was this patch authored or co-authored using generative AI tooling?