HyukjinKwon commented on code in PR #59318:
URL: https://github.com/apache/spark/pull/59318#discussion_r4237623908
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1396,10 +1406,14 @@ private[spark] class DAGScheduler(
* (The remaining group-level case -- a regular shuffle internal to a group
-- is a structural
* invariant that does not arise for the prefix -> pipelined-group -> suffix
shapes targeted here:
* groups are split at regular-shuffle boundaries. The producer-side idioms
-- barrier, DRA,
- * indeterminate, checksum, push-merge -- are rejected in
checkPipelinedProducerSupported at stage
- * creation, where a producer-only throw leaves no partial state.)
+ * indeterminate, checksum, push-merge -- and existing-stage ownership are
checked after these
+ * group-level restrictions, without registering any stages.)
+ *
+ * @throws PipelinedShuffleUnsupportedException for an unsupported pipelined
idiom
+ * @throws SparkException for cross-job reuse or an invalid barrier stage
*/
- private def checkPipelinedGroupsSupportedInRDDGraph(finalRDD: RDD[_]): Unit
= {
+ private def checkPipelinedJobSupported(
Review Comment:
**Nit (P3):** Renaming this method leaves one stale reference outside the
diff. The `MarkPipelinedShuffleForRealTimeMode` scaladoc in
`sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/IncrementalExecution.scala:791`
still cites `checkPipelinedGroupsSupportedInRDDGraph` as the DAGScheduler
check that rejects pipelined fan-out, and `git grep` finds no other occurrence.
Could you update it to `checkPipelinedJobSupported` so the cross-module pointer
stays valid?
**Verification:**
- **Inspection:** A repository-wide search for the old method name finds no
match, and the updated scaladoc names a method that exists in DAGScheduler.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1457,25 +1471,93 @@ private[spark] class DAGScheduler(
// Rooting the consumer check at each checkpointed RDD (rather than at the
PSD-reading RDD) is
// what makes it cover a checkpoint anywhere DOWNSTREAM in the consumer
stage, not just on the
// reading RDD itself.
- def chainHasReliableCheckpoint(root: RDD[_]): Boolean =
- !traverseParentRDDsWithinStage(root, (r: RDD[_]) =>
- !r.checkpointData.exists(_.isInstanceOf[ReliableRDDCheckpointData[_]]))
- val offending =
- // CONSUMER side: a checkpointed RDD whose own within-stage chain reads
a pipelined shuffle is
- // inside a consumer member stage (covers a checkpoint anywhere in that
stage, not just on the
- // reading RDD). PRODUCER side: a producer root's within-stage chain
carries a checkpoint.
- reliablyCheckpointed.exists(rddChainReadsPipelinedShuffle) ||
- producerRoots.exists(chainHasReliableCheckpoint)
+ val offending = reliablyCheckpointed.nonEmpty && {
+ // Shared ancestry is visited once per search, not once per root. The
two searches need
+ // separate visited sets because they have different predicates, even
when roots overlap.
+ val consumerVisited = new HashSet[RDD[_]]
+ val producerVisited = new HashSet[RDD[_]]
+ reliablyCheckpointed.exists(rdd => rddChainReadsPipelinedShuffle(rdd,
consumerVisited)) ||
+ producerRoots.exists { root =>
+ !traverseParentRDDsWithinStage(
+ root, (r: RDD[_]) => !reliablyCheckpointed.contains(r),
producerVisited)
+ }
+ }
if (offending) {
throw pipelinedUnsupportedError(
"a reliable RDD checkpoint in a pipelined-group member's within-stage
chain")
}
+ checkPipelinedStageCreation(finalRDD, partitions, jobId)
+ }
+
+ /** Validate the stages a native result job would create, without
registering them. */
Review Comment:
**Nit (P3):** "native" doesn't appear anywhere else in DAGScheduler; the
file contrasts *pipelined* with *regular*, and in Spark "native" usually
suggests native code or engines. Since this validates every result job with a
pipelined dependency, "pipelined result job" would be clearer. The warm-up
comment at L1519 ("so native determinism does not recurse ...") has the same
issue; it refers to the pipelined producer's determinism check.
```suggestion
/** Validate the stages a pipelined result job would create, without
registering them. */
```
**Verification:**
- **Inspection:** A whole-word search for 'native' in DAGScheduler.scala
finds no remaining use referring to pipelined jobs or producers, and the
reworded comments stay within the 100-character line limit.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1457,25 +1471,93 @@ private[spark] class DAGScheduler(
// Rooting the consumer check at each checkpointed RDD (rather than at the
PSD-reading RDD) is
// what makes it cover a checkpoint anywhere DOWNSTREAM in the consumer
stage, not just on the
// reading RDD itself.
- def chainHasReliableCheckpoint(root: RDD[_]): Boolean =
- !traverseParentRDDsWithinStage(root, (r: RDD[_]) =>
- !r.checkpointData.exists(_.isInstanceOf[ReliableRDDCheckpointData[_]]))
- val offending =
- // CONSUMER side: a checkpointed RDD whose own within-stage chain reads
a pipelined shuffle is
- // inside a consumer member stage (covers a checkpoint anywhere in that
stage, not just on the
- // reading RDD). PRODUCER side: a producer root's within-stage chain
carries a checkpoint.
- reliablyCheckpointed.exists(rddChainReadsPipelinedShuffle) ||
- producerRoots.exists(chainHasReliableCheckpoint)
+ val offending = reliablyCheckpointed.nonEmpty && {
+ // Shared ancestry is visited once per search, not once per root. The
two searches need
+ // separate visited sets because they have different predicates, even
when roots overlap.
+ val consumerVisited = new HashSet[RDD[_]]
+ val producerVisited = new HashSet[RDD[_]]
+ reliablyCheckpointed.exists(rdd => rddChainReadsPipelinedShuffle(rdd,
consumerVisited)) ||
+ producerRoots.exists { root =>
+ !traverseParentRDDsWithinStage(
+ root, (r: RDD[_]) => !reliablyCheckpointed.contains(r),
producerVisited)
+ }
+ }
if (offending) {
throw pipelinedUnsupportedError(
"a reliable RDD checkpoint in a pipelined-group member's within-stage
chain")
}
+ checkPipelinedStageCreation(finalRDD, partitions, jobId)
+ }
+
+ /** Validate the stages a native result job would create, without
registering them. */
+ private def checkPipelinedStageCreation(
+ finalRDD: RDD[_], partitions: Array[Int], jobId: Int): Unit = {
+ val checkedShuffleIds = new HashSet[Int]
+ val pending = new ListBuffer[() => Unit]
+
+ def isRegistered(shuffleId: Int): Boolean =
+ checkedShuffleIds.contains(shuffleId) ||
shuffleIdToMapStage.contains(shuffleId)
+
+ def checkStage(rdd: RDD[_], numTasks: => Int):
HashSet[ShuffleDependency[_, _, _]] = {
+ val (parents, profiles) = getShuffleDependenciesAndResourceProfiles(rdd)
+ val profile = mergeResourceProfilesForStage(profiles)
+ checkBarrierStage(rdd, profile, numTasks)
+ parents
+ }
+
+ def enqueueParents(parents: HashSet[ShuffleDependency[_, _, _]]): Unit = {
+ parents.toSeq.reverseIterator.foreach { dep =>
+ pending.prepend(() => checkShuffle(dep))
+ }
+ }
+
+ def checkNewStage(dep: ShuffleDependency[_, _, _]): Unit = {
+ val parents = checkStage(dep.rdd, dep.rdd.getNumPartitions)
+ checkPipelinedProducerSupported(dep)
+ outputTrackerMaster(dep)
+ pending.prepend(() => {
+ // Warm regular ancestors too, so native determinism does not recurse
through a cold
+ // materialized prefix. This must happen after the parents have been
checked.
+ dep.rdd.outputDeterministicLevel
+ checkedShuffleIds += dep.shuffleId
+ ()
+ })
+ enqueueParents(parents)
Review Comment:
**Nit (P3):** This `enqueueParents(parents)` is what pre-checks ownership of
a registered pipelined parent that sits *below* a newly planned stage, but no
job-submission test exercises that path. The new cross-job tests all place the
foreign-owned stage directly under the result stage (the lookup frontier).
`direct stage construction retains parent ownership validation` calls
`createShuffleMapStage` directly, so it never runs the preflight. If this
branch stopped checking registered parents, construction would still throw
`PIPELINED_SHUFFLE_CROSS_JOB_REUSE`, just after registering sibling stages, so
every assertion would still pass and the partial-registration leak this PR
fixes would come back unnoticed.
A small case would pin it down. The result reads a fresh pipelined A and a
fresh pipelined B, and B's producer reads a pipelined C owned by another active
job, with A ordered first. Assert the error, check
`assertNoPipelinedStageRegistration`, and confirm the owner job's stages are
unchanged.
**Verification:**
- **Regression:** The new test passes at the PR head. It fails when the
below-frontier ownership pre-check in checkPipelinedStageCreation is disabled,
because stage registration then occurs before the construction-time rejection.
- **Behavior:** The existing pipelined preflight and cross-job reuse tests
in the scheduler suite still pass.
--
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]
