jerrypeng commented on code in PR #59318:
URL: https://github.com/apache/spark/pull/59318#discussion_r4277604415
##########
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:
Thanks, added in 21ce72d9f0a: pipelined preflight: an owned parent of a new
stage is rejected before registration. A new frontier stage reads the
foreign-owned pipelined stage and a fresh pipelined sibling. Construction
builds a missing ancestor before the frontier stage looks up its parents, so
the sibling is always built first and the test needs no ordering. Over two
submissions it asserts the cross-job error, that no stage is constructed or
registered with a tracker, that the shuffle-to-stage map is unchanged, and that
only the owner job is active. With enqueueParents(parents) removed from
checkNewStage, it fails on the unexpected createShuffleMapStage.
--
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]
