[SPARK-58263][CORE] Concurrently schedule pipelined-shuffle stage groups in the DAGScheduler

### What changes were proposed in this pull request?

This PR adds native `DAGScheduler` support for **concurrently scheduling stages connected by a
`PipelinedShuffleDependency`** (added earlier in the stack), together with the admission and
completion semantics that make co-scheduling correct.

A pipelined shuffle is incrementally readable: a consumer stage may begin reading the producer's
output while the producer is still running. The stock scheduler runs a consumer only after its
producer has fully materialized; this PR teaches the scheduler to co-schedule a producer and its
pipelined consumer as a **pipelined group** instead. Every new path is gated on a job actually
using a pipelined dependency, so behavior is unchanged for all existing jobs (the existing
`DAGSchedulerSuite` is unaffected).

This PR supports a job that is **all-regular or all-pipelined** (a job mixing the two is rejected
up front). So an all-pipelined job's whole stage graph is one pipelined group, which lets admission
be decided once, up front.

- **Up-front gang admission (`handleJobSubmitted`).** All members of a pipelined group must run
  concurrently, so a group that cannot fit would deadlock (a consumer holds slots waiting for
  producer output while the producer cannot get slots to produce). Before any stage is created, the
  group's total task demand (computed from the RDD graph) is compared against the currently **free**
  slots of its resource profile -- total capacity (`maxNumConcurrentTasks`) minus the *outstanding*
  (running **plus enqueued**) task demand of other work in the same profile. Counting enqueued, not
  just running, demand prevents two groups from each passing the check yet failing to co-fit. If the
  group does not fit, the job fails fast with `CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT` before any
  member runs -- true all-or-nothing admission that leaves no partial scheduler state (like the
  barrier slot check). There is no scheduler-side retry: a transient shortfall is the caller's to
  retry (a streaming query's batch loop reruns the batch). The check can be disabled with
  `spark.scheduler.pipelinedGroup.slotCheck.enabled=false` for deployments that admit capacity
  out-of-band (e.g. a slot reservation).

- **Co-scheduling (`submitStage`).** A stage's missing parents are classified by shuffle dependency
  type; a parent reached through a `PipelinedShuffleDependency` is a "pipelined parent". The stage is
  co-scheduled with its producers (tasks submitted immediately) only if every missing parent is
  pipelined **and** each is already running; otherwise it parks in `waitingStages` exactly as before.
  `submitWaitingPipelinedChildStages` is the "producer started running" analog of the existing
  "producer completed" hook: when a pipelined producer starts, its waiting consumers are reconsidered
  immediately, cascading transitively down a chain.

- **Deferred completion for a co-scheduled consumer (`handleTaskCompletion` /
  `markStageAsFinished`).** A consumer co-scheduled with a still-running producer can finish first.
  Advancing its stage/job completion early would end the job and cancel the still-running producer,
  or make the consumer's output observable before the producer's. So the consumer's completion is
  deferred until the producer's outcome is final: its whole `CompletionEvent` is buffered and
  returns before any side effect runs (accumulator updates, `SparkListenerTaskEnd`, stage/job
  completion), which makes those side effects run exactly once, at replay. The buffered event is
  replayed on genuine producer success (applied normally), or dropped on producer failure -- on the
  drop the buffered tasks' `TaskEnd` events are still emitted (the tasks did finish, so active-task
  listeners must see them end) but no stage/job success is applied, since the group reruns. Deferrals
  are cleaned up on job end/abort so none outlive their job, flushing any still-buffered `TaskEnd`
  events on the way out.

- **Other guards.** A job using a pipelined dependency is rejected up front when speculation is
  enabled (a speculative producer copy would race a consumer already reading partial output, with no
  commit barrier), when dynamic allocation is enabled (gang admission needs a stable slot set), or
  when a group member carries a **non-default resource profile** (gang admission measures capacity
  against the default profile, so the whole group must run on it -- per-profile admission is a
  follow-up); and a `PipelinedShuffleDependency` cannot be submitted as a map-stage job (no durable
  map output to compute statistics from).

Main changes:

- `DAGScheduler.scala` -- job classification, up-front gang admission, co-scheduling, and
  completion deferral. A one-pass RDD-graph walk (`rddGraphHasPipelinedDependency` /
  `classifyJobShuffleKinds`) keeps every new path inert for a job with no pipelined dependency.
- `TaskSchedulerImpl.outstandingTasksForOtherWorkInProfile` -- a resource-profile-scoped
  outstanding-task (running + enqueued) count for the admission check (`private[scheduler]`).
- `spark.scheduler.pipelinedGroup.slotCheck.enabled` -- a new `internal()` config (default `true`).
- `CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT` -- a new error condition.

This is a follow-up to `PipelinedShuffleDependency` and the dependency-typed shuffle-manager
routing (SPARK-58185, already in `master`); it is the first PR that makes the scheduler behave
differently for a pipelined dependency. Group-atomic failure/rerun and additional fail-fast checks
for unsupported idioms follow in later PRs of the stack.

### Why are the changes needed?

`PipelinedShuffleDependency` and its incremental shuffle routing let a consumer read a producer's
output as it is produced, but nothing takes advantage of that until the scheduler co-schedules the
two stages -- otherwise the consumer still waits for the producer to fully materialize and the
pipelining is never realized. Co-scheduling in turn requires admission control (a group that cannot
co-fit must fail fast, not deadlock) and completion control (a fast-finishing consumer must not end
the job or cancel its producer). This PR provides both.

### Does this PR introduce _any_ user-facing change?

No. All new behavior is gated on a job using a `PipelinedShuffleDependency`, which nothing constructs
yet, so for every existing job the scheduler behaves exactly as before. The new
`spark.scheduler.pipelinedGroup.slotCheck.enabled` config is `internal()` and defaults to `true`,
and `CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT` is a new error condition that can only surface for a job
that uses a pipelined dependency.

### How was this patch tested?

New unit tests in `DAGSchedulerSuite` cover:

- concurrent submission of a pipelined producer/consumer; inertness for a regular shuffle; a job
  mixing pipelined + regular shuffles rejected up front; a deep all-pipelined chain co-scheduled;
  transitive cascade when a producer starts; no double-submission on producer completion;
- speculation, dynamic-allocation, and non-default-resource-profile rejection for a pipelined job
  (and that the corresponding regular jobs are not rejected -- including a regular job that merely
  attaches a non-default profile via `RDD.withResources`); a pipelined dependency submitted as a
  map-stage job rejected;
- up-front admission: a group too large to co-fit failing fast with
  `CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT`; a group that fits total capacity but not free slots
  failing fast; a group whose demand exactly equals free capacity admitted; other work's outstanding
  demand charged against admission; the slot check disabled admitting an over-capacity group;
- deferred completion: an early-finishing consumer not ending the job or cancelling its running
  producer; its completion buffered until the producer finishes, then applied exactly once at replay
  (no buffer+replay `TaskEnd` duplication); normal producer-then-consumer ordering; the deferral
  dropped on producer failure; the deferral released only when the producer is genuinely available;
  an explicit job cancellation cleaning up the buffered deferral; and job teardown flushing a
  buffered consumer's `TaskEnd` events even when the release path never drained it (so active-task
  listeners do not leak the tasks as running).

`TaskSchedulerImplSuite` covers `outstandingTasksForOtherWorkInProfile` counting running + enqueued
tasks, excluding given stages, being resource-profile-scoped, and not double-counting a
zombie + live attempt.

### Was this patch authored or co-authored using generative AI tooling?

Co-authored with: Claude Code (Opus 4.8)

Closes #57341 from jerrypeng/stack/pipelined-shuffle-pr3-scheduling.

Authored-by: Boyang Jerry Peng <jerry.peng@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>