Define effective executor basis for downward shuffle sizing with dynamic allocation

[parthosa]

Problem

The downward shuffle-partition pass calculates one execution wave as:

effective executor count × cores per executor

Today, the effective executor count comes from the final spark.executor.instances recommendation, falling back to recommendedClusterInfo.numExecutors.

With dynamic allocation, several counts may exist:

• spark.dynamicAllocation.minExecutors
• spark.dynamicAllocation.initialExecutors
• spark.dynamicAllocation.maxExecutors
• event-log peak executor count
• recommended spark.executor.instances

It is not documented which count should define an execution wave.

PR #2129 was validated using a fixed 12-executor cluster and did not decide initial versus maximum executor semantics for dynamic allocation.

Current edge case

The AutoTuner only coordinates spark.executor.instances with spark.dynamicAllocation.initialExecutors when the source application explicitly contains spark.dynamicAllocation.initialExecutors.

If the source omits that property but a target cluster enforces spark.dynamicAllocation.maxExecutors=12, the maximum may not cap the independently calculated spark.executor.instances. The downward pass can then size its wave using an executor count larger than the configured maximum.

The recommended cluster record is also not updated after dynamic-allocation recommendations rescale or clamp executor counts.

Desired behavior

Define one explicit effective executor-count policy for dynamic allocation.

A possible policy is:

effectiveExecutors = min(recommended/observed executor count, effective maxExecutors)

Handle absent or unlimited maxExecutors, minExecutors, absent initialExecutors, enforced and preserved properties, and platforms such as Databricks that exclude spark.executor.instances.

Do not automatically use maxExecutors: it can be a safety ceiling that the application never reaches.

Acceptance criteria

• Document the executor count used for shuffle-wave sizing.
• Dynamic allocation works when initialExecutors is absent.
• An explicit finite maxExecutors bounds the wave executor count.
• The slot count never uses Spark's unlimited Int.MaxValue default.
• Recommendations satisfy min <= initial <= max.
• Tests cover min/initial/max = 1/1/12 with observed peaks of 1, 6, and 12.
• Tests cover target-cluster enforced and preserved values.
• Tests cover Databricks's executor-instances exclusion fallback.
• Cluster information and the executor count used by downward tuning do not silently disagree.

Context

Follow-up to #2129. This is separate from the general evidence and safety-gate follow-ups in #2133 because it defines a dynamic-allocation correctness boundary.

[rishic3]

So it sounds like there are two separate issues. One is a bug that is independent of #2129: if initialExecutors is not set, enforceDynamicAllocationInvariants() -- which ensures the minExecutors <= initialExecutors == executor.instances <= maxExecutors invariant -- does not do anything. In such a case, we can end up with a source application that sets minExecutors/maxExecutors, but our recommendation for executor.instances could be outside of that range.

The second issue is about how to determine the "effective executors" when DA is enabled. That would require more thought, but it is not a correctness issue.