dongjoon-hyun opened a new pull request, #59737:
URL: https://github.com/apache/spark/pull/59737
### What changes were proposed in this pull request?
This PR aims to limit the resource profile check of
`DAGScheduler.checkPipelinedJobSupported` to the members of a pipelined group.
Currently, the check runs inside the whole-graph walk of the method. Since
the walk also follows regular shuffle dependencies, it checks the RDDs of a
fully-materialized regular prefix too.
```scala
traverseRDDGraph(finalRDD) { (rdd, enqueue) =>
...
val rp = rdd.getResourceProfile()
if (rp != null && rp.id != ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID) {
hasNonDefaultResourceProfile = true
}
rdd.dependencies.foreach {
...
case sd: ShuffleDependency[_, _, _] =>
shuffleRoots.getOrElseUpdate(sd.shuffleId, sd.rdd)
enqueue(sd.rdd) // Walks into the materialized regular prefix too
...
}
}
```
This PR moves the check after the fan-out check and walks only the
within-stage chains of the group members, which are the result stage and each
pipelined producer. Like the producer side of the reliable checkpoint check, it
uses `traverseParentRDDsWithinStage` with a visited set shared across the
roots. The error message and the order of the checks are unchanged.
### Why are the changes needed?
SPARK-57399 (apache/spark#58097) allows a pipelined job whose regular
shuffles are a fully-materialized prefix below the pipelined shuffles (the
materialized-prefix mixed shape of `JobShuffleShape`). The RDDs in such a
prefix are not members of the pipelined group.
- The prefix stages are skipped because all of their map outputs are already
registered.
- A member stage's resource profile comes only from its own within-stage
chain because `getShuffleDependenciesAndResourceProfiles` stops at shuffle
boundaries.
- The gang admission counts only the result stage and the pipelined
producers (`pipelinedJobConcurrentTaskDemand`).
However, the resource profile check was added by SPARK-58398
(apache/spark#57361) when only all-pipelined jobs were accepted, so the whole
RDD graph was the group at that time. It was not narrowed when the
materialized-prefix shape was added. As a result, a job whose pipelined group
runs entirely on the default profile fails up front only because its
materialized prefix was computed with `RDD.withResources(...)`.
```
[PIPELINED_SHUFFLE_UNSUPPORTED] A pipelined shuffle stage group cannot be
scheduled because it uses an unsupported feature: a pipelined group member with
a non-default resource profile (the whole group must run on the default
profile). ...
```
The other member checks of the same method, the reliable checkpoint check
and the repeated-read check, already stop at regular shuffle boundaries.
### Does this PR introduce _any_ user-facing change?
No for the released Apache Spark versions because the pipelined shuffle
scheduling exists only in the `master` branch.
In the `master` branch, a pipelined job is no longer rejected when only its
materialized regular prefix uses a non-default resource profile. A job is still
rejected when any group member uses one. This case happens only through the RDD
API (`PipelinedShuffleDependency` is a `DeveloperApi`) because SQL does not
produce the materialized-prefix shape. It also requires a non-local deployment
with dynamic allocation disabled, since local mode rejects non-default resource
profiles outside tests and a pipelined job rejects dynamic allocation.
### How was this patch tested?
Pass the CIs with the newly added test cases.
Without this fix, the following test case fails with the above error.
- `SPARK-60496: pipelined shuffle: a non-default resource profile on a
materialized regular prefix does not reject the pipelined suffix`
The other new test case, `SPARK-60496: pipelined shuffle: a non-default
resource profile on an RDD shared by a materialized prefix and a member stage
is rejected`, makes sure that an RDD is still checked when it is both the
map-side RDD of the prefix and a narrow parent of a pipelined producer.
In addition, this PR increases the expected edge visits of `pipelined
checkpoint preflight: shared ancestry is visited once` by one because the new
check walks the members' within-stage chains once. The number of visits still
does not grow with the number of roots.
```
$ build/sbt 'core/testOnly org.apache.spark.scheduler.DAGSchedulerSuite --
-z "pipelined" -z "profile"'
...
[info] Tests: succeeded 140, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
