Skip to content

[SPARK-58511][SQL] Bypass ineffective pre-shuffle partial aggregation at runtime #57742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ulysses-you wants to merge 5 commits into
base: master
Choose a base branch
from
Open

[SPARK-58511][SQL] Bypass ineffective pre-shuffle partial aggregation at runtime #57742

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a62ca30
[SPARK-58511][SQL] Bypass ineffective pre-shuffle partial aggregation…
ulysses-you Aug 4, 2026
ce6e4dc
Address review comments
ulysses-you Aug 4, 2026
e1bbdd7
Simplify the adaptive partial aggregation policy
ulysses-you Aug 5, 2026
d95d6c5
Count fast-map rows in the compaction ratio, allow minRows = 0
ulysses-you Aug 5, 2026
f35a80c
Guard the adaptive codegen fragments and merge the per-row block
ulysses-you Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions docs/sql-performance-tuning.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,59 @@ Missing or inaccurate statistics will hinder Spark's ability to select an optima
- **Query plan estimates**: You can inspect Spark's cost estimates in the optimized query plan via [`EXPLAIN COST`](sql-ref-syntax-qry-explain.html) or `DataFrame.explain(mode="cost")`.
- **Runtime statistics**: You can inspect these statistics in the [SQL UI](web-ui.html#sql-tab) under the "Details" section as a query is running. Look for `Statistics(..., isRuntime=true)` in the plan.

## Optimizing the Aggregate

### Adaptive Partial Aggregation

A grouping aggregation normally runs in two phases: a partial aggregation before the shuffle and a
final aggregation after it. The partial aggregation is only worthwhile when it actually reduces the
number of rows; when the grouping keys are close to unique it maintains -- and possibly spills -- an
aggregation map roughly as large as its input while emitting almost as many rows as it consumed.

When adaptive partial aggregation is enabled, hash aggregation measures the compaction ratio (the
number of processed rows divided by the number of keys held in its aggregation maps) at runtime
and, if the partial aggregation is not collapsing enough rows to be worthwhile, stops populating
the aggregation map and passes the remaining rows through as single-row partial aggregation buffers
for the final aggregation to merge. Query results are unchanged. The ratio is evaluated
periodically, and again right before the aggregation map would spill -- in which case the spill is
skipped entirely -- so a query that only becomes ineffective later in its input is still caught.

<table class="spark-config">
<thead><tr><th>Property Name</th><th>Default</th><th>Meaning</th><th>Since Version</th></tr></thead>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.enabled</code></td>
<td>true</td>
<td>
When true, hash aggregation adaptively bypasses the pre-shuffle partial aggregation at runtime
when it observes that the partial aggregation is not reducing the number of rows enough to be
worthwhile. This applies only to hash aggregation with grouping keys.
</td>
<td>4.4.0</td>
</tr>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.minRows</code></td>
<td>100000</td>
<td>
The number of rows to process before the compaction ratio is evaluated, so a decision is never
made on too few rows. The ratio is re-evaluated every time this many further rows have been
processed. Setting this to <code>0</code> disables the periodic evaluation entirely, leaving
only the check made when the aggregation map is about to spill.
</td>
<td>4.4.0</td>
</tr>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.minCompaction</code></td>
<td>1.1</td>
<td>
The minimum compaction ratio required to keep the partial aggregation. A ratio of 10 means the
partial aggregation collapses ten rows into one key; when the observed ratio is below this
value the partial aggregation is bypassed for the rest of the input. A larger value bypasses
more aggressively.
</td>
<td>4.4.0</td>
</tr>
</table>

## Optimizing the Join Strategy

### Automatically Broadcasting Joins
Expand Down
51 changes: 51 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4156,6 +4156,48 @@ object SQLConf {
.booleanConf
.createWithDefault(false)

val ADAPTIVE_PARTIAL_AGGREGATION_ENABLED =
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.enabled")
.doc("When true, hash aggregation adaptively bypasses the pre-shuffle partial aggregation " +
"at runtime when it observes that the partial aggregation is not reducing the number of " +
"rows enough to be worthwhile. Once bypassed, the remaining input rows are passed " +
"through as single-row partial aggregation buffers for the final aggregation to merge, " +
"which avoids the cost of maintaining and spilling a large aggregation map with little " +
"reduction benefit. This applies only to hash aggregation with grouping keys.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
.createWithDefault(true)

val ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS =
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minRows")
.doc("The number of rows to process before adaptive partial aggregation (see " +
s"'${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}') evaluates the compaction ratio. The " +
"ratio is evaluated once this many rows have been processed since the previous " +
"evaluation, so a decision is never made on too few rows. A value of 0 disables the " +
"periodic evaluation entirely, leaving only the check made when the aggregation map is " +
"about to spill.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.longConf
.checkValue(_ >= 0, "The minimum row count must not be negative.")
.createWithDefault(100000)

val ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION =
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minCompaction")
.doc("The minimum compaction ratio required to keep the pre-shuffle partial aggregation " +
s"(see '${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). The compaction ratio is the " +
"number of processed rows divided by the number of keys held in the aggregation maps, " +
"so a ratio of 10 means the partial aggregation collapses ten rows into one. When the " +
s"ratio is below this value after '${ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key}' rows, " +
"or when the aggregation map is about to spill, the partial aggregation is bypassed for " +
"the rest of the input. A larger value bypasses more aggressively.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.doubleConf
.checkValue(_ >= 1.0, "The minimum compaction ratio must be at least 1.0.")
.createWithDefault(1.1)

val JSON_GENERATOR_IGNORE_NULL_FIELDS =
buildConf("spark.sql.jsonGenerator.ignoreNullFields")
.doc("Whether to ignore null fields when generating JSON objects in JSON data source and " +
Expand Down Expand Up @@ -8903,6 +8945,15 @@ class SQLConf extends Serializable with Logging with SqlApiConf {

def bypassPartialAggregation: Boolean = getConf(BYPASS_PARTIAL_AGGREGATION)

def adaptivePartialAggregationEnabled: Boolean =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_ENABLED)

def adaptivePartialAggregationMinRows: Long =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS)

def adaptivePartialAggregationMinCompaction: Double =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION)

def objectAggSortBasedFallbackThreshold: Int = getConf(OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD)

def variableSubstituteEnabled: Boolean = getConf(VARIABLE_SUBSTITUTE_ENABLED)
Expand Down
56 changes: 56 additions & 0 deletions sql/core/benchmarks/AdaptivePartialAggregationBenchmark-jdk21-results.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
================================================================================================
high-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 4094 4126 45 2.0 488.0 1.0X
codegen = true, adaptive = T 2402 2424 32 3.5 286.3 1.7X
codegen = false, adaptive = F 4989 4994 6 1.7 594.8 0.8X
codegen = false, adaptive = T 3183 3193 13 2.6 379.5 1.3X


================================================================================================
low-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 287 299 11 58.4 17.1 1.0X
codegen = true, adaptive = T 282 290 5 59.4 16.8 1.0X
codegen = false, adaptive = F 1328 1329 2 12.6 79.2 0.2X
codegen = false, adaptive = T 1342 1350 12 12.5 80.0 0.2X


================================================================================================
high-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 8850 8911 87 0.9 1055.0 1.0X
codegen = true, adaptive = T 4448 4570 173 1.9 530.3 2.0X
codegen = false, adaptive = F 9261 9357 136 0.9 1104.0 1.0X
codegen = false, adaptive = T 5276 5355 112 1.6 629.0 1.7X


================================================================================================
low-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 789 806 16 21.3 47.0 1.0X
codegen = true, adaptive = T 813 835 28 20.6 48.5 1.0X
codegen = false, adaptive = F 1351 1425 104 12.4 80.5 0.6X
codegen = false, adaptive = T 1346 1350 6 12.5 80.2 0.6X


56 changes: 56 additions & 0 deletions sql/core/benchmarks/AdaptivePartialAggregationBenchmark-jdk25-results.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
================================================================================================
high-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 4083 4087 6 2.1 486.7 1.0X
codegen = true, adaptive = T 2431 2443 17 3.5 289.8 1.7X
codegen = false, adaptive = F 4913 4924 14 1.7 585.7 0.8X
codegen = false, adaptive = T 3214 3220 9 2.6 383.1 1.3X


================================================================================================
low-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 257 267 8 65.3 15.3 1.0X
codegen = true, adaptive = T 282 292 7 59.5 16.8 0.9X
codegen = false, adaptive = F 1290 1290 0 13.0 76.9 0.2X
codegen = false, adaptive = T 1298 1301 4 12.9 77.4 0.2X


================================================================================================
high-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 7932 7986 77 1.1 945.5 1.0X
codegen = true, adaptive = T 4246 4354 152 2.0 506.2 1.9X
codegen = false, adaptive = F 9290 9386 136 0.9 1107.4 0.9X
codegen = false, adaptive = T 5244 5298 76 1.6 625.2 1.5X


================================================================================================
low-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 764 774 17 22.0 45.5 1.0X
codegen = true, adaptive = T 786 794 8 21.3 46.9 1.0X
codegen = false, adaptive = F 1362 1362 0 12.3 81.2 0.6X
codegen = false, adaptive = T 1363 1368 7 12.3 81.2 0.6X


56 changes: 56 additions & 0 deletions sql/core/benchmarks/AdaptivePartialAggregationBenchmark-results.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
================================================================================================
high-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 3975 4060 120 2.1 473.8 1.0X
codegen = true, adaptive = T 2381 2404 32 3.5 283.8 1.7X
codegen = false, adaptive = F 4850 4854 5 1.7 578.2 0.8X
codegen = false, adaptive = T 3081 3086 7 2.7 367.3 1.3X


================================================================================================
low-cardinality input, no-spill pass-through (Tier 1)
================================================================================================

OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, no spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 287 318 24 58.4 17.1 1.0X
codegen = true, adaptive = T 312 320 7 53.8 18.6 0.9X
codegen = false, adaptive = F 1261 1263 3 13.3 75.1 0.2X
codegen = false, adaptive = T 1301 1303 3 12.9 77.6 0.2X


================================================================================================
high-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, high card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 8319 8410 128 1.0 991.7 1.0X
codegen = true, adaptive = T 4421 4464 61 1.9 527.0 1.9X
codegen = false, adaptive = F 9254 9314 85 0.9 1103.2 0.9X
codegen = false, adaptive = T 5251 5257 8 1.6 626.0 1.6X


================================================================================================
low-cardinality input, on-spill pass-through (Tier 2)
================================================================================================

OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
AMD EPYC 7763 64-Core Processor
adaptive partial agg, low card, spill: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
codegen = true, adaptive = F 808 816 11 20.8 48.1 1.0X
codegen = true, adaptive = T 826 842 19 20.3 49.3 1.0X
codegen = false, adaptive = F 1332 1343 15 12.6 79.4 0.6X
codegen = false, adaptive = T 1303 1304 1 12.9 77.7 0.6X


Loading