Skip to content

[SPARK-58084][SQL] Support converting sort merge join to shuffled hash join in AQE physical plan #57181

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-58084][SQL] Support converting sort merge join to shuffled hash join in AQE physical plan #57181

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
d7d0f0c
[SPARK-58084][SQL] Support converting sort merge join to shuffled has…
ulysses-you Jul 10, 2026
59291b6
[SPARK-58084][SQL] Address review: guards, cost refactor, config, cle…
ulysses-you Jul 13, 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
24 changes: 24 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 @@ -375,6 +375,30 @@ AQE converts sort-merge join to shuffled hash join when all post shuffle partiti
</td>
<td>3.2.0</td>
</tr>
<tr>
<td><code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code></td>

@viirya viirya Jul 12, 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.

Nit: this entry only mentions the per-partition size condition, but preferShuffledHashJoin also requires spark.sql.adaptive.advisoryPartitionSizeInBytes <= maxShuffledHashJoinLocalMapThreshold, which is worth mentioning like the existing 3.2.0 entry above does.

@ulysses-you ulysses-you Jul 13, 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.

Updated - the entry now notes that preferShuffledHashJoin also requires spark.sql.adaptive.advisoryPartitionSizeInBytes to not be larger than maxShuffledHashJoinLocalMapThreshold, mirroring the 3.2.0 entry. Also added the two new configs to the table.

<td>false</td>
<td>
When true, Spark converts a sort-merge join to a shuffled hash join during adaptive execution when the build side's materialized per-partition sizes are all within <code>spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold</code> (which additionally requires <code>spark.sql.adaptive.advisoryPartitionSizeInBytes</code> to not be larger than it), even when non-shuffle operators (such as aggregate, project, filter and window) sit between the join and its input shuffle.
</td>
<td>4.3.0</td>
</tr>
<tr>
<td><code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.minWideningFactor</code></td>
<td>1.0</td>
<td>
The lower bound applied to the row-widening factor used by <code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code> when bounding a build side's shuffled hash map size. The factor scales the input shuffle bytes by the estimated per-row size growth of the operators between the join and its shuffle; a larger lower bound is more conservative and makes the conversion less likely when statistics may under-estimate the build size. Must be positive.
</td>
<td>4.3.0</td>
</tr>
<tr>
<td><code>spark.sql.adaptive.costEvaluator.countLocalSort.enabled</code></td>
<td>(value of <code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code>)</td>
<td>
When true, the default AQE cost evaluator also counts the number of local sorts as a lower-priority tiebreaker below the number of shuffles, so that among plans with the same number of shuffles the one with fewer local sorts is preferred. For example, a sort-merge join is replaced by a shuffled hash join only when the conversion does not push extra sorts elsewhere in the plan. Defaults to the value of <code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code>, so it is enabled together with that conversion.
</td>
<td>4.3.0</td>
</tr>
</table>

### Optimizing Skew Join
Expand Down
2 changes: 1 addition & 1 deletion ...catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1288,7 +1288,7 @@ case class FindInSet(left: Expression, right: Expression) extends BinaryExpressi

trait String2TrimExpression extends Expression with ImplicitCastInputTypes {

protected def srcStr: Expression
private[sql] def srcStr: Expression
protected def trimStr: Option[Expression]
protected def direction: String

Expand Down
15 changes: 14 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,10 @@ package org.apache.spark.sql.catalyst.optimizer
import scala.annotation.tailrec
import scala.util.control.NonFatal

import org.apache.spark.MapOutputStatistics
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{HASH_JOIN_KEYS, JOIN_CONDITION}
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, ExtractFiltersAndInnerJoins, ExtractSingleColumnNullAwareAntiJoin}
Expand Down Expand Up @@ -287,7 +289,18 @@ case object BuildRight extends BuildSide

case object BuildLeft extends BuildSide

trait JoinSelectionHelper extends Logging {
trait JoinSelectionHelper extends SQLConfHelper with Logging {

def preferShuffledHashJoin(
mapStats: MapOutputStatistics,
sizeInBytesFactor: Double = 1.0): Boolean = {
val maxShuffledHashJoinLocalMapThreshold =
conf.getConf(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD)
val advisoryPartitionSize = conf.getConf(SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES)
advisoryPartitionSize <= maxShuffledHashJoinLocalMapThreshold &&
mapStats.bytesByPartitionId.forall(
_ * sizeInBytesFactor <= maxShuffledHashJoinLocalMapThreshold)
}

def getBroadcastBuildSide(
join: Join,
Expand Down
49 changes: 49 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 @@ -1337,6 +1337,46 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefault(0L)

val ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED =
buildConf("spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled")
.doc("When true, Spark converts a sort merge join to a shuffled hash join during adaptive " +
"execution when the build side's materialized per-partition sizes are all within " +
s"${ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key} (which additionally requires " +
s"${ADVISORY_PARTITION_SIZE_IN_BYTES.key} to not be larger than it), even when " +
"non-shuffle operators sit between the join and its input shuffle.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
.createWithDefault(false)

val ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_MIN_WIDENING_FACTOR =
buildConf("spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.minWideningFactor")
.doc("The lower bound applied to the row-widening factor used by " +
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key} when bounding a " +
"build side's shuffled hash map size. The factor scales the input shuffle bytes by the " +
"estimated per-row size growth of the operators between the join and its shuffle; a " +
"larger lower bound is more conservative and makes the conversion less likely when " +
"statistics may under-estimate the build size. Must be positive.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.doubleConf
.checkValue(_ > 0, "The minimum widening factor must be positive.")
.createWithDefault(1.0)

val ADAPTIVE_COST_EVALUATOR_COUNT_LOCAL_SORT_ENABLED =

@viirya viirya Jul 12, 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.

Enabling only convertSortMergeJoinToShuffledHashJoin.enabled without costEvaluator.countLocalSort.enabled means a conversion that pushes extra sorts elsewhere is still adopted (your own test shows the 5-sorts-vs-4 case). Is there a reason to keep them independent? If the cost-evaluator change is kept behind its own flag for safety, the docs should at least recommend enabling them together - or countLocalSort could arguably default to true, since it is a no-op for cost comparison when nothing else changes the sort count.

@ulysses-you ulysses-you Jul 13, 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.

Made spark.sql.adaptive.costEvaluator.countLocalSort.enabled a fallbackConf to ...convertSortMergeJoinToShuffledHashJoin.enabled, so by default it is enabled exactly together with the conversion. I kept it as a separate overridable flag rather than defaulting it to true globally, because flipping it on unconditionally is not a no-op - it broke the existing "Change broadcast join to merge join" test, where a tied BHJ-vs-SMJ re-optimization now prefers the sort-free BHJ. The fallback gives the "enabled together" behavior you suggested without that regression.

buildConf("spark.sql.adaptive.costEvaluator.countLocalSort.enabled")
.doc("When true, the default AQE cost evaluator also counts the number of local sorts as a " +
"lower-priority tiebreaker below the number of shuffles. This lets adaptive execution " +
"prefer a plan with fewer local sorts among plans with the same number of shuffles, for " +
s"example a shuffled hash join produced by " +
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key} over a " +
"sort merge join when the conversion does not push extra sorts elsewhere. Defaults to " +
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key} so that it is " +
"enabled together with that conversion.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.fallbackConf(ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED)

val ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED =
buildConf("spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled")
.doc(s"When true and '${ADAPTIVE_EXECUTION_ENABLED.key}' is true, Spark will optimize the " +
Expand Down Expand Up @@ -8090,6 +8130,15 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def nonEmptyPartitionRatioForBroadcastJoin: Double =
getConf(NON_EMPTY_PARTITION_RATIO_FOR_BROADCAST_JOIN)

def convertSortMergeJoinToShuffledHashJoinEnabled: Boolean =
getConf(ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED)

def convertSortMergeJoinToShuffledHashJoinMinWideningFactor: Double =
getConf(ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_MIN_WIDENING_FACTOR)

def costEvaluatorCountLocalSortEnabled: Boolean =
getConf(ADAPTIVE_COST_EVALUATOR_COUNT_LOCAL_SORT_ENABLED)

def coalesceShufflePartitionsEnabled: Boolean = getConf(COALESCE_PARTITIONS_ENABLED)

def minBatchesToRetain: Int = getConf(MIN_BATCHES_TO_RETAIN)
Expand Down
8 changes: 7 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,9 @@ case class AdaptiveSparkPlanExec(
conf.getConf(SQLConf.ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS) match {
case Some(className) =>
CostEvaluator.instantiate(className, context.session.sparkContext.getConf)
case _ => SimpleCostEvaluator(conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN))
case _ => SimpleCostEvaluator(
conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN),
conf.costEvaluatorCountLocalSortEnabled)
}

// A list of physical plan rules to be applied before creation of query stages. The physical
Expand All @@ -125,6 +127,10 @@ case class AdaptiveSparkPlanExec(
InsertSortForLimitAndOffset,
AdjustShuffleExchangePosition,
ValidateSparkPlan,
// Must run before `ReplaceHashWithSortAgg`: converting a sort merge join to a shuffled hash
// join drops its child ordering, which `ReplaceHashWithSortAgg` would otherwise rely on to
// turn a hash aggregate into a sort aggregate.
ConvertSortMergeJoinToShuffledHashJoin(ensureRequirements),
ReplaceHashWithSortAgg,
RemoveRedundantSorts,
RemoveRedundantWindowGroupLimits,
Expand Down
Loading