[SPARK-59670][CORE] Avoid redundant waiting-stage scans and job-preflight traversals

[jerrypeng]

What changes were proposed in this pull request?

Reduce redundant DAGScheduler work introduced by pipelined-shuffle scheduling:

1. Return from submitWaitingPipelinedChildStages before scanning waiting stages when the
   running stage is not a pipelined producer. Pipelined-producer wake-up behavior is unchanged.
2. Reuse the negative pipelined-dependency check when speculation or dynamic allocation is
   enabled, instead of traversing an ordinary job's RDD graph again for shuffle-shape
   classification. Preserve short-circuit rejection and validation precedence.
3. Count pipelined producer task demand and enqueue dependencies in one dependency loop,
   preserving shuffle-ID deduplication and graph traversal order.

The changes are submission-local: no cross-job cache, new configuration, or scheduling policy.

JIRA: SPARK-59670

Why are the changes needed?

The pipelined-child helper currently scans waiting stages even when the running stage cannot
have any pipelined children to wake. Repeated scans add avoidable event-loop work, especially
for large ordinary stage graphs.

Ordinary jobs with speculation or dynamic allocation enabled also pay for two negative
preflight graph traversals. The first has already established that no pipelined dependency
exists, so the second cannot add classification information. Pipelined task-demand calculation
separately enumerates each RDD's dependencies twice when one loop suffices.

This removes those redundant operations, not all scheduler graph traversals. With speculation
and dynamic allocation both disabled, ordinary-job preflight still uses its existing classifier.

Does this PR introduce any user-facing change?

No. Scheduling, admission decisions, error messages, and validation precedence are unchanged.
The changes reduce scheduler overhead.

How was this patch tested?

Added 17 regression cases to DAGSchedulerSuite covering:

• No waiting-stage parent inspection when starting ordinary result or shuffle-map stages.
• A single ordinary-job preflight traversal across speculation/dynamic-allocation and
  shuffle/no-shuffle combinations.
• Short-circuit feature rejection and precedence over mixed-shape rejection.
• Single dependency enumeration during pipelined task-demand calculation, with unchanged demand.
• Successful job completion and scheduler-state cleanup on the relevant paths.

Local validation with Java 17:

build/sbt -batch \
  'core/testOnly org.apache.spark.scheduler.DAGSchedulerSuite' \
  'core/Compile/scalastyle' \
  'core/Test/scalastyle'

• All 256 DAGSchedulerSuite tests passed.
• Both production and test scalastyle checks passed with zero errors or warnings.
• Mutation checks temporarily restored the redundant operations and confirmed that the
  corresponding traversal-count tests fail. The optimized code was restored and the full
  suite rerun successfully.
• git diff --check passed.

An exploratory local helper benchmark used 1,000- and 10,000-RDD chains with a shuffle every
16 edges. Graph setup was outside timing. Separate counters verified that ordinary preflight
(with speculation enabled) and pipelined task-demand calculation each reduced dependency
enumeration from 2 * (N - 1) to N - 1, with identical classification/demand results.
Best warmed ordinary-preflight timings were 59.7 to 25.7 microseconds for 1,000 RDDs and
1.096 to 0.513 milliseconds for 10,000 RDDs. These are helper-level results from one local run,
not end-to-end query speedups. No waiting-stage-scan timing benchmark was run, and the temporary
benchmark harness is not included in this PR.

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

Co-authored with OpenAI Codex CLI 0.144.1

[cloud-fan]

Review summary

I did not find any PR-caused correctness, concurrency, compatibility, or test-coverage issues in the pinned change. The optimized paths preserve the existing event-loop ownership and shuffle-shape decisions, and the new mock and counter assertions are regression-sensitive to the redundant work being removed. Two scanner-raised wording concerns were verified in the merge target with unchanged scope and consequence, so they are not findings for this PR.

Findings

0 total: 0 P0, 0 P1, 0 P2, 0 P3.

No findings.

[cloud-fan]

The failed SQLAppStatusListenerWithRocksDBBackendSuite is unrelated (other variants passed), merging to master/4.x/4.3, thanks!

[cloud-fan]

Merge Summary:

• merged into master 00d52e2
• merged into branch-4.x e0785b3
• merged into branch-4.3 8f80dc1

Posted by merge_spark_pr.py