cloud-fan commented on code in PR #59523:
URL: https://github.com/apache/spark/pull/59523#discussion_r4290941111
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1489,6 +1489,28 @@ 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 ||
+
rdd.checkpointData.exists(_.isInstanceOf[LocalRDDCheckpointData[_]])) {
+ persisted += rdd
+ }
+ rdd.dependencies.foreach(dep => enqueue(dep.rdd))
+ }
+ // Cache locations cannot establish a stable cut: eviction can revive an
upstream producer,
+ // and a partial cache hit skips a reader whose producer still needs its
termination ACK.
+ // Check storage intent instead, including pending local checkpoints and
narrow descendants.
+ val visited = new HashSet[RDD[_]]
+ if (persisted.exists(rdd => rddChainReadsPipelinedShuffle(rdd, visited))) {
+ throw pipelinedUnsupportedError(
+ "cache/persist/localCheckpoint in a pipelined consumer's within-stage
chain; " +
+ "unpersist cached RDDs or rebuild locally checkpointed consumers
without " +
+ "localCheckpoint before submitting the job; persist producer inputs
instead")
Review Comment:
**Nit (P3):** The last clause, `persist producer inputs instead`, is only
true for a producer whose own stage reads no pipelined shuffle. Take an
all-pipelined chain `A -PSD1-> B -PSD2-> C`. Every RDD in stage B is an input
of the B->C producer, but it also reads PSD1 within its stage, so
`rddChainReadsPipelinedShuffle` flags it. The new test `reject a persisted
stage that consumes and produces shuffles` is exactly this case: `middle` is
PSD2's producer root and is still rejected. A user who persisted `middle` is
told to do what they already did, and the job fails the same way again.
Could the clause name the placement the check actually allows? For example:
only RDDs that do not read a pipelined shuffle within their stage (such as
inputs of the first pipelined producer) can be persisted. No existing assertion
pins this clause, so rewording it should not disturb the current tests.
A smaller point: the template renders this as `uses an unsupported feature:
<reason>. Pipelined ... shuffles ... incompatible with this feature in this
version.` So the two imperative clauses read a bit oddly next to the
noun-phrase reasons the sibling rejections use.
**Verification:**
- **Behavior:** Persisting an RDD in the middle stage of a pipelined chain
that both consumes and produces pipelined shuffles is still rejected before any
stage registration. The resulting reason describes the allowed placement as one
outside any pipelined consumer's within-stage chain.
- **Compatibility:** Existing rejection coverage for direct consumers,
narrow descendants, pending and unpersisted local checkpoints, and real cached
blocks still observes the same error condition, exception type, and unpersist
and rebuild guidance.
--
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]
