This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new a9b8163c36f3 [SPARK-57667][SQL][TEST] Make SQL tests be independent
from the default value of `maxShuffledHashJoinLocalMapThreshold`
a9b8163c36f3 is described below
commit a9b8163c36f34b01772e52494bcf7ca9f5661050
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Jun 24 13:47:29 2026 -0700
[SPARK-57667][SQL][TEST] Make SQL tests be independent from the default
value of `maxShuffledHashJoinLocalMapThreshold`
### What changes were proposed in this pull request?
This PR aims to make SQL tests be independent from the default value of
`maxShuffledHashJoinLocalMapThreshold` by pinning on the current default value.
https://github.com/apache/spark/blob/9dab83b728705157ba4511c6d3b5b8244089a6f5/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L1302-L1311
### Why are the changes needed?
These suites assume AQE does not prefer a shuffled hash join, which only
holds while this threshold
defaults to `0`. If the default is raised, AQE picks a shuffled hash join
over a sort merge join and
breaks them (e.g. SMJ-spill checks, extended explain output, plan-tree node
ids). A test that assumes a config value should set it explicitly.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56747 from dongjoon-hyun/SPARK-57667.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit d55264e3e7bf505652278ec9d380acd761998bc1)
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 3186e1e9bf3bb3e75d6b42a6d00488e97eec4c9b)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala | 1 +
sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala | 3 +++
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala | 7 +++++++
.../src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala | 2 ++
.../test/scala/org/apache/spark/sql/collation/CollationSuite.scala | 3 +++
.../scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala | 3 +++
.../spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala | 3 +++
7 files changed, 22 insertions(+)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
index 5192d5d66f24..3d6fd5c1bf5f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
@@ -2571,6 +2571,7 @@ class DataFrameSuite extends SharedSparkSession
test("SPARK-41048: Improve output partitioning and ordering with AQE cache")
{
withSQLConf(
SQLConf.CAN_CHANGE_CACHED_PLAN_OUTPUT_PARTITIONING.key -> "true",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key -> "0",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
val df1 = spark.range(10).selectExpr("cast(id as string) c1")
val df2 = spark.range(10).selectExpr("cast(id as string) c2")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
index af52204dbb7f..618f0ab675e1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
@@ -552,6 +552,9 @@ class ExplainSuite extends ExplainSuiteHelper with
DisableAdaptiveExecutionSuite
class ExplainSuiteAE extends ExplainSuiteHelper with
EnableAdaptiveExecutionSuite {
import testImplicits._
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
test("SPARK-35884: Explain Formatted") {
val df1 = Seq((1, 2), (2, 3)).toDF("k", "v1")
val df2 = Seq((2, 3), (1, 1)).toDF("k", "v2")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
index 3ea77446f268..7f695e90df88 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
@@ -46,6 +46,9 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
setupTestData()
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
def statisticSizeInByte(df: classic.DataFrame): BigInt = {
df.queryExecution.optimizedPlan.stats.sizeInBytes
}
@@ -1834,6 +1837,10 @@ class ThreadLeakInSortMergeJoinSuite
with AdaptiveSparkPlanHelper {
setupTestData()
+
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
override protected def createSparkSession: TestSparkSession = {
classic.SparkSession.cleanupAnyExistingSession()
new TestSparkSession(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
index 389d0d5a29d5..395cb67f4415 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
@@ -179,6 +179,8 @@ class SQLQueryTestSuite extends SharedSparkSession with
SQLHelper
// regex magic.
.set("spark.test.noSerdeInExplain", "true")
.set(SQLConf.SCHEMA_LEVEL_COLLATIONS_ENABLED, true)
+ // SPARK-57667: pin so AQE keeps SMJ regardless of the default
+ .set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD, 0L)
// SPARK-32106 Since we add SQL test 'transform.sql' will use `cat` command,
// here we need to ignore it.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
index a6f4de1c80e3..37684c7fce3b 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
@@ -49,6 +49,9 @@ class CollationSuite extends DatasourceV2SQLBase with
AdaptiveSparkPlanHelper {
private val fullyQualifiedPrefix =
s"${CollationFactory.CATALOG}.${CollationFactory.SCHEMA}."
private val collations = Seq("UTF8_BINARY", "UTF8_LCASE", "UNICODE",
"UNICODE_CI")
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
@inline
private def isSortMergeForced: Boolean = {
SQLConf.get.getConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD) == -1
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
index 877e6970368b..f7afdb5e6e53 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
@@ -52,6 +52,9 @@ case class QueryExecutionTestRecord(
class QueryExecutionSuite extends SharedSparkSession {
import testImplicits._
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
def checkDumpedPlans(path: String, expected: Int): Unit =
Utils.tryWithResource(
Source.fromFile(path)) { source =>
assert(source.getLines().toList
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
index 50322905f29f..381305abec6a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
@@ -69,6 +69,9 @@ class AdaptiveQueryExecSuite
setupTestData()
+ override protected def sparkConf =
+
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
+
private def runAdaptiveAndVerifyResult(query: String,
skipCheckAnswer: Boolean = false): (SparkPlan, SparkPlan) = {
var finalPlanCnt = 0
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
