Add a downward-only shuffle partition pass sized from consumer-stage shuffle input

[parthosa]

Closes #2128 (partially — see Scope)

What this adds

A shuffle-input analysis and partition-sizing pass, running after the job-level and cluster-level recommendations so it sees the value those produced. It sums the uncompressed data size of every shuffle branch entering each consumer stage — a join's inputs add up, a reused exchange counts twice — divides by a target size, takes the worst stage, and rounds up to whole cluster waves.

It holds the existing recommendation on any of:

• incomplete evidence — an unsized exchange, an unresolvable consumer stage, or an unfinished SQL execution
• a failed stage or an OOM anywhere in the application
• spill or shuffle-read skew in a consumer stage
• any increase the normal passes made
• runtime-managed shuffle, or a property that cannot be written

Against the 4.08 TB shuffle in #2128 it computes 3840 — within 1% of the 3800 that succeeded, from the first run's metrics rather than after three attempts.

Scope: applies downward only

It applies only when the result lowers the current value. On that workload the existing rules undershoot, so it computes the right number and rejects it as not-downward — it did not apply on any run of that workload. Fixing that needs the two changes tracked in #2128, so this ships disabled. What lands here is the measurement layer: correct, tested against real event logs, prerequisite for either direction.

Key design decisions

• Consumer-stage totals, not per-exchange — sizing the largest exchange alone understates a join stage by roughly its smaller input.
• Each metric read once, from its producing stage — SQLAccumProfileResults.total sums across stages and inflates any application that retried one.
• Both partition properties written together or not at all — a preserved/enforced/excluded/limited property is silently ignored on write.
• Two supporting changes make the evidence trustworthy: SparkListenerSQLExecutionEnd.errorMessage is recorded so a completed SQL execution can be told from a failed one, and task accumulables carry their stage attempt so GPU spill isn't attributed across attempts.

Configuration

DOWNWARD_SHUFFLE_ENABLED=false, opted into per run via --tuning_configs; five knobs in tuningConfigs.yaml.

Testing

New DownwardShufflePartitionsSuite and ShuffleStageInputMetricsSuite (real event logs: multi-branch joins, exchange reuse, AQE-split reads, missing metrics, unfinished SQL, attempt selection), plus AutoTuner coverage for the applied path and each gate. CI mvn verify is green across the matrix — Java 8/11/17, Spark 321/357, Scala 2.12 and 2.13.

[parthosa]

@rishic3 Could you please help reviewing this PR?

[rishic3]

I think what we want here is "affected stages", where

affected_stages = [
    s for s in shuffle-consuming stages
    if observed_task_count[s] > downward_suggestion
]

E.g., suppose we are suggesting 400, and stage A had 4,000 tasks with no spill, while stage B had 200 (post-AQE) tasks and spilled. Only A would be affected by the recommendation and B should not block.

[parthosa]

B's 200 is post AQE. The candidate sets the partition count before AQE coalesces, so comparing it to a count from after mixes two different things. B's 200 tasks may have come from 8000 partitions. Drop it to 400 and AQE has coarser input to work with, so B is still affected.

With coalescing off, numTasks is the fan-out and your filter is exactly right. So rather than drop the idea, I'd make the affected-stage filter depend on whether coalescing is on. Does that get you what you want? Follow-up with its own tests, since it loosens a gate.

[rishic3]

That makes sense, but IMO I still think we should consider the filter as I described. Maybe the right name for it is "constrained stages", i.e. stages that are actually constrained below their previously observed post-AQE tasks. In my example we are still giving enough nominal partitions to stage B to reproduce 200 tasks, so it isn't constrained. The counterfactual is that if B wasn't spilling, a seemingly "safe" reduction from 8,000 to 400 could now make B spill if AQE now coalesces more, so there is always this coalescing risk.

I'm fine with deferring after we do more testing, or starting with the 'coalescing is on' dependence. I am just coming from the fact that the example I described was literally describing a real job we saw that improved after we applied the heuristic, despite the spill in stage B.

[parthosa]

I think we should be proceed with more experimentation on this. Added as a follow up in #2133.

[rishic3]

Looks good to me, thanks @parthosa. Leaving some questions that can be deferred. (I don't have permission to resolve comments, but everything else looks good).

[rishic3]

That makes sense, but IMO I still think we should consider the filter as I described. Maybe the right name for it is "constrained stages", i.e. stages that are actually constrained below their previously observed post-AQE tasks. In my example we are still giving enough nominal partitions to stage B to reproduce 200 tasks, so it isn't constrained. The counterfactual is that if B wasn't spilling, a seemingly "safe" reduction from 8,000 to 400 could now make B spill if AQE now coalesces more, so there is always this coalescing risk.

I'm fine with deferring after we do more testing, or starting with the 'coalescing is on' dependence. I am just coming from the fact that the example I described was literally describing a real job we saw that improved after we applied the heuristic, despite the spill in stage B.

[sayedbilalbari]

Thanks @parthosa , LGTM !