Re: [PR] [WIP] Define PipelinedShuffleDependency and route shuffles to a ShuffleManager by dependency type [spark]

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

cloud-fan commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3597592245


##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -421,8 +417,12 @@ class BlockManagerMasterEndpoint(
           mapStatuses.filter(_ != null).foreach { mapStatus =>
             // Check if the executor has been deallocated
             if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
-              val blocksToDel =
-                shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
+              // Only a BlockingShuffleManager serves block-manager blocks; a pipelined shuffle is
+              // served out-of-band and is not in the MapOutputTracker this loop iterates, so this
+              // resolves only regular shuffles (None when the default manager is not blocking).
+              val blocksToDel = SparkEnv.get.shuffleBlockResolver

Review Comment:
   This answers your own question at the `_shuffleManager` param above: with this call the param is now unused (its only consumer, the old `shuffleManager` lazy val, is gone). It can be removed from the constructor — that also touches `SparkEnv.scala:656` (`_shuffleManager = null`) and the two `BlockManagerReplicationSuite` call sites. Non-blocking.



##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -80,12 +80,88 @@ class SparkEnv (
   // user jars to define custom ShuffleManagers.
   @volatile private var _shuffleManager: ShuffleManager = _
 
+  // The manager for pipelined (incrementally-readable) shuffle dependencies, configured by
+  // spark.shuffle.manager.incremental. Present only when that config is set. It is a peer of
+  // _shuffleManager, not a wrapper around it: a shuffle is routed to one or the other by its
+  // dependency type via `shuffleManagerFor`, so neither is ever installed "behind" the other.
+  @volatile private var _incrementalShuffleManager: Option[ShuffleManager] = None
+
   // Latch to signal when the ShuffleManager has been initialized.
   // Used to allow callers to wait for initialization.
   private val shuffleManagerInitLatch = new CountDownLatch(1)
 
+  /**
+   * The default [[ShuffleManager]] (configured by spark.shuffle.manager), which serves all regular
+   * shuffle dependencies. Use this only when the intent is specifically the default manager -- e.g.
+   * inspecting its concrete type, or resolving a shuffle block by id (a pipelined shuffle is served
+   * out-of-band by the incremental manager and produces no block-manager blocks). To serve a
+   * specific shuffle's reads/writes, use `shuffleManagerFor`, which routes by dependency type.
+   */
+  def defaultShuffleManager: ShuffleManager = _shuffleManager

Review Comment:
   There isn't really a "default" manager to name here: block-resolution paths always want the blocking manager, pipelined reads always want the pipelined manager, and routing picks by dependency type. Modeling one slot as the default (and sourcing the resolver from it at `shuffleBlockResolver`, line 112) hides exactly the blocking-vs-pipelined distinction the design relies on. Suggest holding both a blocking and a pipelined manager instance in `SparkEnv` and having each call site pick explicitly — always-blocking, always-pipelined, or by-dependency-type. That also removes the sharp edge where `spark.shuffle.manager.incremental` is set to a blocking manager (the routing suite sets it to `"sort"`) whose blocks are then unreachable through this accessor.



##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleManager.scala:
##########
@@ -28,14 +28,19 @@ import org.apache.spark.util.Utils
  * and on each executor, based on the spark.shuffle.manager setting. The driver registers shuffles
  * with it, and executors (or tasks running locally in the driver) can ask to read and write data.
  *
+ * This trait is sealed and is not meant to be implemented directly. A shuffle manager declares its
+ * kind by extending one of its two subtypes: [[BlockingShuffleManager]] (output is materialized as
+ * block-manager-addressed blocks served through a [[ShuffleBlockResolver]]) or
+ * [[PipelinedShuffleManager]] (output is read incrementally and served out-of-band).
+ *
  * NOTE:
  * 1. This will be instantiated by SparkEnv so its constructor can take a SparkConf and
  * boolean isDriver as parameters.
- * 2. This contains a method ShuffleBlockResolver which interacts with External Shuffle Service
- * when it is enabled. Need to pay attention to that, if implementing a custom ShuffleManager, to
- * make sure the custom ShuffleManager could co-exist with External Shuffle Service.
+ * 2. A [[BlockingShuffleManager]] exposes a ShuffleBlockResolver which interacts with the External
+ * Shuffle Service when it is enabled. Need to pay attention to that, if implementing a custom
+ * shuffle manager, to make sure it could co-exist with the External Shuffle Service.
  */
-private[spark] trait ShuffleManager {
+private[spark] sealed trait ShuffleManager {

Review Comment:
   Please don't `seal` this. Breaking changes to the interface are fine — moving `shuffleBlockResolver` onto `BlockingShuffleManager` is a good change — but `sealed` fully stops third-party `ShuffleManager` implementations, which SPARK-45762 deliberately enabled. We should keep that extension point. (Consequence: the SPARK-45762 test in `SparkSubmitSuite` should be updated to implement `BlockingShuffleManager` rather than removed — see review summary.)



-- 
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