Re: [PR] [SPARK-60271][CORE] Reject persisted native pipelined shuffle consumers [spark]

["cloud-fan (via GitHub)" <gi...@apache.org>]

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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org