[jira] [Created] (SPARK-57688) Add config to allow bypassing pre-shuffle partial aggregation

["James Xu (Jira)" <ji...@apache.org>]

James Xu created SPARK-57688:
--------------------------------

             Summary: Add config to allow bypassing pre-shuffle partial aggregation
                 Key: SPARK-57688
                 URL: https://issues.apache.org/jira/browse/SPARK-57688
             Project: Spark
          Issue Type: Improvement
          Components: Optimizer
    Affects Versions: 4.3.0
            Reporter: James Xu


Problem:

The standard two-phase aggregation plan (Partial → shuffle → Final) assumes that pre-shuffle partial aggregation meaningfully reduces data volume. This assumption breaks down in two scenarios.

Scenario 1: High group cardinality.
When group cardinality is high relative to partition size, every input row maps to a distinct key, so the partial aggregation produces one output row per input row and adds CPU and memory overhead with zero shuffle benefit.

SELECT user_id, SUM(amount), COUNT(order_id), AVG(price)
FROM orders
GROUP BY user_id   -- high-cardinality key: millions of distinct users

On a table with 500M rows and 200M distinct user_id values, the pre-shuffle HashAggregateExec in Partial mode churns through the full dataset, spills when the hash map overflows, and still emits ~200M rows into the shuffle. The partial phase wastes wall-clock time and memory without reducing shuffle write volume.

Scenario 2: Skewed input data.
Even when partial aggregation can reduce data volume on average, skewed input partitions can make it harmful. If one partition contains a disproportionate share of rows for a small number of keys, the partial HashAggregateExec on that partition must hold a large hash map in memory, triggering spills. The skewed partition becomes the bottleneck and dominates wall-clock time — worse than if the data had been shuffled first and aggregated on already-partitioned, evenly distributed data.

SELECT country_code, SUM(revenue)
FROM orders
GROUP BY country_code   -- a few dominant countries hold 80% of rows

---
Solution:

Add a new boolean SQL config spark.sql.execution.bypassPartialAggregation (default false). When set to true, the pre-shuffle partial aggregation phase is skipped and a single Complete-mode aggregation runs after the shuffle instead. Users who have identified either of the above patterns in their workload can opt in to bypass the partial phase.

The config has no effect on queries containing DISTINCT aggregate functions or session_window grouping keys, where the partial aggregation phases are required for correctness and are always applied.

---
Expected Impact:

For aggregations over high-cardinality keys where the pre-shuffle reduction ratio is near 1.0:
- Eliminates one full scan of the partition data by the partial HashAggregateExec
- Eliminates hash map construction and potential spill in the partial phase
- Wall-clock improvement is most significant in spill-heavy cases

For aggregations over skewed input:
- Moves aggregation to post-shuffle, where data is evenly distributed by grouping key
- Eliminates the partial-phase bottleneck on the skewed partition

The config is opt-in (false by default) so existing query behavior is unchanged.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org