jerrypeng commented on code in PR #59318:
URL: https://github.com/apache/spark/pull/59318#discussion_r4277606573
##########
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:
Done in a00c49bf607 with your wording. The L1519 comment was rewritten in
7822b2822f0 and no longer uses the word. The tests' "native" names now say
pipelined too
--
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]
