jerrypeng commented on code in PR #59523:
URL: https://github.com/apache/spark/pull/59523#discussion_r4408263656
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1516,6 +1515,29 @@ private[spark] class DAGScheduler(
checkPipelinedStageCreation(finalRDD, partitions, jobId)
}
+ /** Reject persistence that could bypass a native pipelined shuffle reader.
*/
+ private def checkPipelinedCacheSupported(finalRDD: RDD[_]): Unit = {
+ val persisted = new HashSet[RDD[_]]
+ traverseRDDGraph(finalRDD) { (rdd, enqueue) =>
+ if (rdd.getStorageLevel != StorageLevel.NONE ||
Review Comment:
Agreed—the fresh localCheckpoint(); count() sequence can succeed and
truncate the pipelined lineage. This PR intentionally excludes that success
path through its persistence-intent check, and I’ll document that restriction
explicitly. Supporting a cold-checkpoint exception would require additional
coordination around cache population and failed submissions. Completed
checkpoints that have already removed the native pipelined ancestry remain
unaffected.
I will update the PR description.
--
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]
