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 2 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
2 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
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
61 changes: 61 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,67 @@ 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 reduction ratio (the
number of distinct grouping keys divided by the number of processed rows) at runtime and, if the
partial aggregation is not reducing rows enough 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, 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.3.0</td>
</tr>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.sampleRows</code></td>
<td>100000</td>
<td>
The number of input rows to sample before evaluating the reduction ratio. When the ratio is
below the threshold, the next evaluation happens after twice as many rows, so low-cardinality
input is re-checked only rarely.
</td>
<td>4.3.0</td>
</tr>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.noSpillReductionRatioThreshold</code></td>
<td>0.9</td>
<td>
The reduction ratio threshold applied while the aggregation map is still fully in memory. If
the ratio is at least this value the partial aggregation is bypassed. A larger value is more
conservative (keeps partial aggregation in more cases).
</td>
<td>4.3.0</td>
</tr>
<tr>
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.spillReductionRatioThreshold</code></td>
<td>(value of <code>noSpillReductionRatioThreshold</code>)</td>
<td>
The reduction ratio threshold applied when the aggregation map is about to spill. Setting it
lower makes the bypass more likely once spilling is imminent, and 0 always bypasses instead of
spilling.
</td>
<td>4.3.0</td>
</tr>
</table>

## Optimizing the Join Strategy

### Automatically Broadcasting Joins
Expand Down
70 changes: 70 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,64 @@ 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.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
.createWithDefault(true)

val ADAPTIVE_PARTIAL_AGGREGATION_SAMPLE_ROWS =
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.sampleRows")
.doc("The number of input rows to sample before evaluating the reduction ratio for the " +
s"no-spill tier of adaptive partial aggregation (see " +
s"'${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). From this many rows on, if the ratio " +
"of distinct grouping keys to processed rows is at least " +
s"'spark.sql.execution.aggregate.adaptivePartialAggregation." +
"noSpillReductionRatioThreshold', partial aggregation is bypassed for the rest of the " +
"input. When the ratio is below the threshold, the next evaluation happens after twice " +
"as many rows, so low-cardinality input is re-checked only rarely.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.intConf
.checkValue(_ > 0, "The sample row count must be positive.")
.createWithDefault(100000)

val ADAPTIVE_PARTIAL_AGGREGATION_NO_SPILL_REDUCTION_RATIO_THRESHOLD =

@cloud-fan cloud-fan Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we start with one minimum-row setting and one reduction threshold, applying the same policy both periodically and immediately before spilling? The separate no-spill and spill thresholds plus exponential resampling add configuration and behavioral complexity without showing that these dimensions must be independently tunable. At the spill boundary, switch to pass-through when the common policy says aggregation is ineffective; otherwise spill normally. This also gives the code-generated and interpreted paths one invariant to implement and test.

@ulysses-you ulysses-you Aug 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding two configs is a trade-off, due to that it is difficult to accurately determine how much data reduction rate can offset the performance loss caused by sort and spill. There is a way to use the spill threshold, setting spill thresholds to 0, so it can always passing through directly when spill happens.

I set both of these configs to the same default value to make their behavior the same by default.

@cloud-fan cloud-fan Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, the shared default plus the concrete spill-cost rationale addresses my concern about introducing two policies by default.

buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation." +
"noSpillReductionRatioThreshold")
.doc("The reduction ratio threshold used by the no-spill tier of adaptive partial " +
s"aggregation (see '${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). The reduction ratio " +
"is the number of distinct grouping keys divided by the number of processed rows. After " +
s"sampling '${ADAPTIVE_PARTIAL_AGGREGATION_SAMPLE_ROWS.key}' rows without spilling, if " +
"the ratio is at least this value the partial aggregation is bypassed. A larger value " +
"is more conservative (keeps partial aggregation in more cases).")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.doubleConf
.checkValue(v => v >= 0.0 && v <= 1.0, "The reduction ratio threshold must be in [0.0, 1.0].")
.createWithDefault(0.9)

val ADAPTIVE_PARTIAL_AGGREGATION_SPILL_REDUCTION_RATIO_THRESHOLD =
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation." +
"spillReductionRatioThreshold")
.doc("The reduction ratio threshold used by the on-spill tier of adaptive partial " +
s"aggregation (see '${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). When the aggregation " +
"map is about to spill, if the ratio of distinct grouping keys to processed rows is at " +
"least this value the partial aggregation is bypassed for the rest of the input. It " +
s"falls back to '${ADAPTIVE_PARTIAL_AGGREGATION_NO_SPILL_REDUCTION_RATIO_THRESHOLD.key}' " +
"so that both tiers apply one policy by default. Setting it lower makes the bypass more " +
"likely once spilling is imminent, and 0 always bypasses instead of spilling.")
.version("4.3.0")

@uros-b uros-b Aug 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just leaving a note that branch-4.3 has already been cut, so please check whether these changes are intended to go there as well @ulysses-you

.withBindingPolicy(ConfigBindingPolicy.SESSION)
.fallbackConf(ADAPTIVE_PARTIAL_AGGREGATION_NO_SPILL_REDUCTION_RATIO_THRESHOLD)

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 +8961,18 @@ class SQLConf extends Serializable with Logging with SqlApiConf {

def bypassPartialAggregation: Boolean = getConf(BYPASS_PARTIAL_AGGREGATION)

def adaptivePartialAggregationEnabled: Boolean =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_ENABLED)

def adaptivePartialAggregationSampleRows: Int =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_SAMPLE_ROWS)

def adaptivePartialAggregationNoSpillReductionRatioThreshold: Double =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_NO_SPILL_REDUCTION_RATIO_THRESHOLD)

def adaptivePartialAggregationSpillReductionRatioThreshold: Double =
getConf(ADAPTIVE_PARTIAL_AGGREGATION_SPILL_REDUCTION_RATIO_THRESHOLD)

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


Loading