This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new bbd7c46919a4 [SPARK-57261][SQL] Allow to disable HashAggregateExec by
config
bbd7c46919a4 is described below
commit bbd7c46919a4546e4921acb6e64495a7d966bfb2
Author: Cheng Pan <[email protected]>
AuthorDate: Wed Jun 10 16:13:35 2026 +0800
[SPARK-57261][SQL] Allow to disable HashAggregateExec by config
### What changes were proposed in this pull request?
Currently, Spark always prefers to use `HashAggregateExec` over
`SortAggregateExec` if possible, this PR adds a config
`spark.sql.execution.useHashAggregateExec` to allow users to disable
`HashAggregateExec` explicitly.
### Why are the changes needed?
We found some jobs fail with `HashAggregateExec` due to OOM (auto fallback
logic does not work well), and it runs well with `SortAggregateExec`
```
26/06/04 18:47:30 ERROR [SIGTERM handler] CoarseGrainedExecutorBackend:
RECEIVED SIGNAL TERM
26/06/04 18:47:30 WARN [Executor task launch worker for task 9749.0 in
stage 14.0 (TID 61758)] TaskMemoryManager: Failed to allocate a page
(2147483648 bytes) for 0 times, try again.
java.lang.OutOfMemoryError: Java heap space
at
org.apache.spark.unsafe.memory.HeapMemoryAllocator.allocate(HeapMemoryAllocator.java:72)
at
org.apache.spark.memory.TaskMemoryManager.allocatePage(TaskMemoryManager.java:398)
at
org.apache.spark.memory.TaskMemoryManager.allocatePage(TaskMemoryManager.java:359)
at
org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:96)
at
org.apache.spark.unsafe.map.BytesToBytesMap.allocate(BytesToBytesMap.java:868)
at
org.apache.spark.unsafe.map.BytesToBytesMap.growAndRehash(BytesToBytesMap.java:991)
at
org.apache.spark.unsafe.map.BytesToBytesMap$Location.append(BytesToBytesMap.java:817)
at
org.apache.spark.sql.execution.UnsafeFixedWidthAggregationMap.getAggregationBufferFromUnsafeRow(UnsafeFixedWidthAggregationMap.java:135)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage11.hashAgg_doConsume_0$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage11.hashAgg_doAggregateWithKeys_0$(Unknown
Source)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage11.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:44)
at
org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:50)
at scala.collection.Iterator$$anon$9.hasNext(Iterator.scala:593)
at
org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:195)
at
org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:57)
at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:111)
at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
at
org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:206)
at org.apache.spark.scheduler.Task.run(Task.scala:147)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:900)
at
org.apache.spark.executor.Executor$TaskRunner$$Lambda$709/0x00007f84474fd558.apply(Unknown
Source)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:86)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:83)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:99)
at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:903)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
UT is tuned.
Also verified with a production case, `HashAggregateExec` vs
`SortAggregateExec`
<img width="1764" height="346" alt="Xnip2026-06-04_21-23-43"
src="https://github.com/user-attachments/assets/401d3ec2-2bdc-4be3-bc1c-9ab2c2543e1f";
/>
<img width="1756" height="167" alt="Xnip2026-06-04_21-22-38"
src="https://github.com/user-attachments/assets/84146634-1e71-4887-ad4b-e374df112441";
/>
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #56323 from pan3793/SPARK-57261.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../scala/org/apache/spark/sql/internal/SQLConf.scala | 12 +++++++++++-
.../apache/spark/sql/execution/aggregate/AggUtils.scala | 16 +++-------------
.../org/apache/spark/sql/DataFrameAggregateSuite.scala | 7 +++++--
.../spark/sql/execution/WholeStageCodegenSuite.scala | 2 +-
.../spark/sql/execution/metric/SQLMetricsSuite.scala | 2 +-
.../configs-without-binding-policy-exceptions | 1 +
6 files changed, 22 insertions(+), 18 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index eb542bf8c883..8a54fc55ba7a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3808,11 +3808,19 @@ object SQLConf {
val USE_OBJECT_HASH_AGG =
buildConf("spark.sql.execution.useObjectHashAggregateExec")
.internal()
- .doc("Decides if we use ObjectHashAggregateExec")
+ .doc("Decides if we use ObjectHashAggregateExec when possible.")
.version("2.2.0")
.booleanConf
.createWithDefault(true)
+ val USE_HASH_AGG = buildConf("spark.sql.execution.useHashAggregateExec")
+ .internal()
+ .doc("Decides if we use HashAggregateExec when possible, the rule takes
precedence " +
+ s"over ${USE_OBJECT_HASH_AGG.key}.")
+ .version("4.3.0")
+ .booleanConf
+ .createWithDefault(true)
+
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 " +
@@ -8199,6 +8207,8 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def useObjectHashAggregation: Boolean = getConf(USE_OBJECT_HASH_AGG)
+ def useHashAggregation: Boolean = getConf(USE_HASH_AGG)
+
def objectAggSortBasedFallbackThreshold: Int =
getConf(OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD)
def variableSubstituteEnabled: Boolean = getConf(VARIABLE_SUBSTITUTE_ENABLED)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
index 58055fa6129a..c2a7024367af 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
@@ -75,13 +75,12 @@ object AggUtils {
initialInputBufferOffset: Int = 0,
resultExpressions: Seq[NamedExpression] = Nil,
child: SparkPlan): SparkPlan = {
- val useHash = Aggregate.supportsHashAggregate(
+ val useHash = child.conf.useHashAggregation &&
Aggregate.supportsHashAggregate(
aggregateExpressions.flatMap(_.aggregateFunction.aggBufferAttributes),
groupingExpressions)
val forceObjHashAggregate = forceApplyObjectHashAggregate(child.conf)
- val forceSortAggregate = forceApplySortAggregate(child.conf)
- if (useHash && !forceSortAggregate && !forceObjHashAggregate) {
+ if (useHash && !forceObjHashAggregate) {
HashAggregateExec(
requiredChildDistributionExpressions =
requiredChildDistributionExpressions,
isStreaming = isStreaming,
@@ -97,7 +96,7 @@ object AggUtils {
val useObjectHash = Aggregate.supportsObjectHashAggregate(
aggregateExpressions, groupingExpressions)
- if (forceObjHashAggregate || (objectHashEnabled && useObjectHash &&
!forceSortAggregate)) {
+ if (forceObjHashAggregate || (objectHashEnabled && useObjectHash)) {
ObjectHashAggregateExec(
requiredChildDistributionExpressions =
requiredChildDistributionExpressions,
isStreaming = isStreaming,
@@ -584,15 +583,6 @@ object AggUtils {
}
}
- /**
- * Returns whether a sort aggregate should be force applied.
- * The config key is hard-coded because it's testing only and should not be
exposed.
- */
- private def forceApplySortAggregate(conf: SQLConf): Boolean = {
- Utils.isTesting &&
- conf.getConfString("spark.sql.test.forceApplySortAggregate", "false") ==
"true"
- }
-
/**
* Returns whether a object hash aggregate should be force applied.
* The config key is hard-coded because it's testing only and should not be
exposed.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 9d78e4c6c1a5..31428ab7108a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -3360,14 +3360,17 @@ class DataFrameAggregateSuite extends SharedSparkSession
df: => DataFrame,
expected: Int): Unit = {
val configurations = Seq(
- Seq.empty[(String, String)], // hash aggregate is used by default
+ Seq(SQLConf.USE_HASH_AGG.key -> "true"),
Seq(SQLConf.CODEGEN_FACTORY_MODE.key -> "NO_CODEGEN",
"spark.sql.TungstenAggregate.testFallbackStartsAt" -> "1, 10"),
Seq("spark.sql.test.forceApplyObjectHashAggregate" -> "true"),
Seq(
"spark.sql.test.forceApplyObjectHashAggregate" -> "true",
SQLConf.OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD.key -> "1"),
- Seq("spark.sql.test.forceApplySortAggregate" -> "true")
+ Seq(SQLConf.USE_HASH_AGG.key -> "false"),
+ Seq(
+ SQLConf.USE_HASH_AGG.key -> "false",
+ SQLConf.USE_OBJECT_HASH_AGG.key -> "false")
)
// Make tests faster
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
index a83d5c99bb5d..886df9184aca 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
@@ -59,7 +59,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
test("SortAggregate should be included in WholeStageCodegen") {
val df = spark.range(10).agg(max(col("id")), avg(col("id")))
- withSQLConf("spark.sql.test.forceApplySortAggregate" -> "true") {
+ withSQLConf(SQLConf.USE_HASH_AGG.key -> "false") {
val plan = df.queryExecution.executedPlan
assert(plan.exists(p =>
p.isInstanceOf[WholeStageCodegenExec] &&
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
index e2eb37e7a433..6f06a7348080 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
@@ -329,7 +329,7 @@ class SQLMetricsSuite extends SharedSparkSession with
SQLMetricsTestUtils
test("SortAggregate metrics") {
// Force use SortAggregateExec instead of HashAggregateExec
- withSQLConf("spark.sql.test.forceApplySortAggregate" -> "true") {
+ withSQLConf(SQLConf.USE_HASH_AGG.key -> "false") {
// Assume the execution plan is
// -> SortAggregate(nodeId = 0)
// -> Sort(nodeId = 1)
diff --git
a/sql/hive/src/test/resources/conf/binding-policy-exceptions/configs-without-binding-policy-exceptions
b/sql/hive/src/test/resources/conf/binding-policy-exceptions/configs-without-binding-policy-exceptions
index de94d0127931..26f29ba24d10 100644
---
a/sql/hive/src/test/resources/conf/binding-policy-exceptions/configs-without-binding-policy-exceptions
+++
b/sql/hive/src/test/resources/conf/binding-policy-exceptions/configs-without-binding-policy-exceptions
@@ -603,6 +603,7 @@ spark.sql.execution.replaceHashWithSortAgg
spark.sql.execution.reuseSubquery
spark.sql.execution.sortBeforeRepartition
spark.sql.execution.topKSortFallbackThreshold
+spark.sql.execution.useHashAggregateExec
spark.sql.execution.useObjectHashAggregateExec
spark.sql.execution.usePartitionEvaluator
spark.sql.extendedExplainProviders
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
