[WIP] Stack/pipelined shuffle pr6 failfast

common/utils/src/main/resources/error/error-conditions.json

@@ -1126,6 +1126,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT" : {
+    "message" : [
+      "Cannot run the pipelined stage group: it needs <numTasks> concurrent task slots to run all its stages together, but only <numSlots> are currently free. The stages of a pipelined group must run concurrently, so the cluster must have enough free slots for all of them at once. Provision more executors, or reduce the parallelism of the query."
+    ],
+    "sqlState" : "53000"
+  },
   "CONCURRENT_STREAM_LOG_UPDATE" : {
     "message" : [
       "Concurrent update to the log. Multiple streaming jobs detected for <batchId>.",
@@ -6313,6 +6319,18 @@
     ],
     "sqlState" : "42K03"
   },
+  "PIPELINED_SHUFFLE_CROSS_JOB_REUSE" : {
+    "message" : [
+      "The pipelined shuffle <shuffleId> is being reused across jobs. A pipelined (incrementally-readable) shuffle is transient and has no retained output for another job to read, so its producer stage cannot be shared across jobs."
+    ],
+    "sqlState" : "0A000"
+  },
+  "PIPELINED_SHUFFLE_UNSUPPORTED" : {
+    "message" : [
+      "A pipelined shuffle stage group cannot be scheduled because it uses an unsupported feature: <reason>. Pipelined (incrementally-readable) shuffles run their producer and consumer stages concurrently over a transient, once-through stream, which is incompatible with this feature in this version."
+    ],
+    "sqlState" : "0A000"
+  },
   "PIPELINE_DATASET_WITHOUT_FLOW" : {
     "message" : [
       "Pipeline dataset <identifier> does not have any defined flows. Please attach a query with the dataset's definition, or explicitly define at least one flow that writes to the dataset."

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -2006,6 +2006,21 @@ package object config {
       .checkValue(v => v > 0, "The max failures should be a positive value.")
       .createWithDefault(40)
 
+  private[spark] val PIPELINED_GROUP_SLOT_CHECK_ENABLED =
+    ConfigBuilder("spark.scheduler.pipelinedGroup.slotCheck.enabled")
+      .internal()
+      .doc("When true, before co-scheduling a pipelined-shuffle stage group the DAGScheduler " +
+        "checks that the group's total task demand fits in the currently free slots of its " +
+        "resource profile (total capacity minus the outstanding -- running plus enqueued -- " +
+        "tasks of other work), and fails the job with CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT " +
+        "rather than co-scheduling a group that cannot fit and deadlocking. Set to false for " +
+        "deployments that admit capacity out-of-band (e.g. a slot reservation), which then own " +
+        "admission. Only applies to jobs that use a pipelined shuffle dependency.")
+      .version("4.3.0")
+      .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+      .booleanConf
+      .createWithDefault(true)
+
   private[spark] val NUM_CANCELLED_JOB_GROUPS_TO_TRACK =
     ConfigBuilder("spark.scheduler.numCancelledJobGroupsToTrack")
       .doc("The maximum number of tracked job groups that are cancelled with " +

core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala

@@ -63,4 +63,26 @@ private[spark] class ActiveJob(
   val finished = Array.fill[Boolean](numPartitions)(false)
 
   var numFinished = 0
+
+  /**
+   * Whether this job's RDD graph uses a `PipelinedShuffleDependency` (set once at job submission).
+   * Every pipelined-group scheduling path -- co-scheduling, deferral, and the per-submit
+   * `TaskSet.isPipelined` tagging -- is inert for a job without one, so this flag lets those paths
+   * short-circuit the group-membership graph walk for the common regular job at no cost.
+   */
+  var hasPipelinedDependency: Boolean = false
+
+  /**
+   * True once this job's pipelined group has passed up-front gang admission in
+   * `handleJobSubmitted` (or the slot check was disabled by config). This is a distinct fact from
+   * `hasPipelinedDependency` (which only says the job uses pipelining): the co-schedule path in
+   * `DAGScheduler.submitStage` gang-schedules a group's members with NO slot check, and checks this
+   * flag so that trust in up-front admission is enforced rather than merely commented.
+   *
+   * The whole job is a single pipelined group, so admission is a job-level fact. If a future change
+   * allows multiple groups per job, this job-level flag must be replaced by per-group admission
+   * state, and the check in `submitStage` replaced by that per-group gate -- NOT simply deleted, or
+   * an unadmitted group could be gang-scheduled and deadlock.
+   */
+  var pipelinedGroupAdmitted: Boolean = false
 }