This is an automated email from the ASF dual-hosted git repository.
ulysses-you pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new e5d87887d825 [SPARK-58084][SQL] Support converting sort merge join to
shuffled hash join in AQE physical plan
e5d87887d825 is described below
commit e5d87887d8251a43b4fc823d21dd7828ea3bcd8e
Author: Xiduo You <[email protected]>
AuthorDate: Wed Jul 15 14:48:39 2026 +0800
[SPARK-58084][SQL] Support converting sort merge join to shuffled hash join
in AQE physical plan
### What changes were proposed in this pull request?
This PR lets adaptive query execution (AQE) convert a `SortMergeJoinExec`
into a `ShuffledHashJoinExec` on the **physical plan**, so the conversion can
see through non-shuffle operators (aggregate, project, filter, window,
left-existence join) that sit between the join and its input shuffle.
1. **New rule `ConvertSortMergeJoinToShuffledHashJoin`** (in
`queryStagePreparationRules`, **before** `ReplaceHashWithSortAgg` so the
conversion does not destroy an ordering that rule relies on). Once the join's
input shuffles have materialized, it reaches each side's
`ShuffleQueryStageExec` through a set of operators above it and, if a build
side fits `spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold`, rewrites
the join to a shuffled hash join. Guards:
- **binary-stable keys** - skipped unless `hashJoinSupported(leftKeys,
rightKeys)`, so collated / non-binary-stable string keys stay a sort merge join
(matching `JoinSelection`);
- **explicit strategy hint** - skipped when the join carries any
left/right strategy hint (e.g. `MERGE`), so a user hint is never overridden;
- **non-spillable build-size bound** - the traversal only looks through
an operator whose outputs are all size-bounded (`isSizeBoundedExpr`:
attributes, fixed-width values, `cast`, and length-non-increasing string
transforms; not `repeat`/`concat`), and the build-side estimate is scaled by a
per-row widening factor floored at
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.minWideningFactor`
(default `1.0`). This prevents a row-widening operator from turning a spillable
[...]
The swap is shuffle-free (both are `ShuffledJoin` with the same
distribution/partitioning); only the child sorts become unnecessary. Because a
shuffled hash join loses the sort merge join's output ordering,
`EnsureRequirements` is re-run so any ordering an ancestor still needs is
re-established. Gated by
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default
`true`).
2. **The logical SMJ->SHJ preference is removed in favor of the new
physical rule.** `DynamicJoinSelection` is renamed to `DemoteBroadcastHashJoin`
and now only does what its name says - demotes a broadcast to a shuffle join
(`NO_BROADCAST_HASH`) for a child with many empty partitions. Its old
`PREFER_SHUFFLE_HASH` / `SHUFFLE_HASH` selection is dropped, and with it:
- the internal `PREFER_SHUFFLE_HASH` join-strategy hint (`hints.scala`)
and the `hintToPreferShuffleHashJoin*` helpers in `JoinSelectionHelper`, plus
their use in `SparkStrategies.checkHintNonEquiJoin`;
- the AQE optimizer batch is renamed accordingly (`Demote Broadcast Hash
Join`).
This consolidates all SMJ->SHJ selection into
`ConvertSortMergeJoinToShuffledHashJoin` on the physical plan. Note for tests /
callers that previously pinned a plan by excluding `DynamicJoinSelection` from
`spark.sql.adaptive.optimizer.excludedRules`: that no longer disables the SHJ
conversion - disable
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` instead.
3. **New helper `Literal.valueSizeInBytes`** returns a safe upper bound
(never an under-estimate) of a literal value's size in bytes, or `None` when it
cannot be determined reliably: fixed-length and null values report
`defaultSize`; string / binary report their real payload; arrays / maps /
structs sum their measurable elements; anything else (e.g. variant) returns
`None`. `isSizeBoundedExpr` uses it so a literal wider than its type's
`defaultSize` (and any unmeasurable literal) is t [...]
4. **`SimpleCost` now holds `(numSkewJoins, numShuffles, numLocalSorts)`
compared lexicographically** (replacing the packed-`Long` bit tricks).
Local-sort counting is controlled by
`spark.sql.adaptive.costEvaluator.countLocalSort.enabled`, which **falls back
to** the `lookThroughOperators.enabled` config, so it is enabled together with
it and is a no-op otherwise.
Class hierarchy note: `SortMergeJoinExec` and `ShuffledHashJoinExec` both
extend `ShuffledJoin`, which is why the swap needs no new shuffle.
`ShuffledHashJoinExec` (via `HashJoin.outputOrdering`) only preserves the
streamed side's ordering, whereas an inner `SortMergeJoinExec` keeps both
sides'; the local-sort cost handles cases where that difference forces a new
sort upstream.
### Why are the changes needed?
Previously `DynamicJoinSelection` (now `DemoteBroadcastHashJoin`) preferred
a shuffled hash join over a sort merge join, but it worked on the **logical**
plan and only fired when the join child was a shuffle stage directly. When
non-inflating operators (an aggregate, or a filter from a `HAVING`) sit between
the join and its input shuffle, the hint was never added and the join stayed a
sort merge join even though a shuffled hash join would be cheaper. Doing the
selection on the physica [...]
### Does this PR introduce _any_ user-facing change?
Yes, four new configurations, all documented in
`docs/sql-performance-tuning.md`:
- `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled`
(default `true`)
-
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled`
(default `false`)
-
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.minWideningFactor`
(default `1.0`)
- `spark.sql.adaptive.costEvaluator.countLocalSort.enabled` (defaults to
the value of the lookThroughOperators.enabled config)
All default to off / no-op, the direct-shuffle conversion reproduces the
old default behavior (inert unless maxShuffledHashJoinLocalMapThreshold is
raised) so there is no behavior change unless a user opts in.
### How was this patch tested?
New unit tests in `AdaptiveQueryExecSuite`:
- Converting a sort merge join to a shuffled hash join when an aggregate /
`HAVING` filter sits above the shuffle.
- The converted plan stays valid when an ancestor needs the join's ordering
(`EnsureRequirements` re-adds the sort).
- Conversion is looked through size-bounded operators
(`max`/`cast`/`substring`) but stopped by a row-widening operator
(`repeat(max(c2), n)`).
- `minWideningFactor` makes the size bound more conservative and rejects an
otherwise-eligible conversion.
- Collated (non-binary-stable) join keys and an explicit `MERGE` hint keep
the join as a sort merge join with the config enabled.
- `SimpleCostEvaluator` orders plans by skew joins, then shuffles, then
local sorts; a conversion that would add a local sort elsewhere is rejected
when local-sort counting is on.
New unit tests in `LiteralExpressionSuite` cover `Literal.valueSizeInBytes`
across fixed-length, null, string / binary (including multi-byte UTF-8), array,
map, struct, and unmeasurable (variant) values.
Full `AdaptiveQueryExecSuite` passes (124 tests).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57181 from ulysses-you/SPARK-convert-smj-to-shj-aqe.
Authored-by: Xiduo You <[email protected]>
Signed-off-by: Xiduo You <[email protected]>
(cherry picked from commit 3f123cc7620867165cff0356eadf6ba07be46975)
Signed-off-by: Xiduo You <[email protected]>
---
docs/sql-migration-guide.md | 1 +
docs/sql-performance-tuning.md | 32 ++
.../spark/sql/catalyst/expressions/literals.scala | 60 +++
.../catalyst/expressions/stringExpressions.scala | 2 +-
.../spark/sql/catalyst/optimizer/joins.scala | 24 +-
.../spark/sql/catalyst/plans/logical/hints.scala | 8 -
.../org/apache/spark/sql/internal/SQLConf.scala | 71 ++++
.../expressions/LiteralExpressionSuite.scala | 50 ++-
.../spark/sql/execution/SparkStrategies.scala | 3 +-
.../sql/execution/adaptive/AQEOptimizer.scala | 2 +-
.../execution/adaptive/AdaptiveSparkPlanExec.scala | 8 +-
.../ConvertSortMergeJoinToShuffledHashJoin.scala | 260 ++++++++++++
...lection.scala => DemoteBroadcastHashJoin.scala} | 31 +-
.../sql/execution/adaptive/simpleCosting.scala | 56 ++-
.../QueryPlanningTrackerEndToEndSuite.scala | 6 +-
.../adaptive/AdaptiveQueryExecSuite.scala | 462 ++++++++++++++++++++-
16 files changed, 990 insertions(+), 86 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 91fb7a52f670..9b119a2b8c83 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -27,6 +27,7 @@ license: |
- Since Spark 4.3, zero-length files are skipped during Parquet schema
inference instead of failing with a `FAILED_READ_FILE.CANNOT_READ_FILE_FOOTER`
error.
- Since Spark 4.3, the configuration key
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` has
been renamed to
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` to
reflect that it now applies to storage-partitioned joins, aggregates, and
windows. The old key continues to work as an alias.
- Since Spark 4.3, the Spark Thrift Server rejects setting JVM system
properties through the `set:system:` session configuration overlay (for
example, in a JDBC connection string). To restore the previous behavior, set
`spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties` to `true`.
+- Since Spark 4.3, the adaptive execution rule
`org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` has been renamed
to `DemoteBroadcastHashJoin`, which now only demotes broadcast hash joins
(emitting `NO_BROADCAST_HASH`). Its selection of shuffled hash join over sort
merge join has moved to a new physical rule gated by
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default
`true`). If you previously disabled the shuffled-hash-join preference by
listing `o [...]
## Upgrading from Spark SQL 4.1 to 4.2
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index 0294f05641d9..c86570ff34a1 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -380,6 +380,38 @@ 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>
+ <td>true</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). This is the master switch for the conversion. By default it only
looks through the join's own required sort to reach a direct input s [...]
+ </td>
+ <td>4.3.0</td>
+ </tr>
+ <tr>
+
<td><code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled</code></td>
+ <td>false</td>
+ <td>
+ When true, the sort-merge join to shuffled hash join conversion
additionally looks through non-shuffle operators (such as aggregate, project,
filter and window) sitting between the join and its input shuffle, instead of
only the join's own required sort. Has no effect unless
<code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code>
is true.
+ </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 when the
sort-merge-join to shuffled-hash-join conversion bounds 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.lookThroughOperators.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.lookThroughOpe
[...]
+ </td>
+ <td>4.3.0</td>
+ </tr>
</table>
### Optimizing Skew Join
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
index cea16999c437..546d82546d03 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/literals.scala
@@ -300,6 +300,60 @@ object Literal {
s"but class ${Utils.getSimpleName(value.getClass)} found.")
}
+ /**
+ * The size in bytes of a literal `value` of the given `dataType`, or `None`
when it cannot be
+ * determined reliably. The result is a safe upper bound, never an
under-estimate, so callers can
+ * use it to bound memory:
+ * - a fixed-length type reports its `defaultSize` (exact);
+ * - a null `value` reports the type's `defaultSize`;
+ * - a variable-length type whose real payload size is known (string /
binary, or an array /
+ * map / struct all of whose elements are measurable) reports that real
size;
+ * - any other type (e.g. variant) returns `None`.
+ */
+ private[expressions] def valueSizeInBytes(value: Any, dataType: DataType):
Option[Int] = {
+ if (value == null || UnsafeRow.isFixedLength(dataType)) {
+ return Some(dataType.defaultSize)
+ }
+ PhysicalDataType(dataType) match {
+ case _: PhysicalCalendarIntervalType =>
Some(CalendarIntervalType.defaultSize)
+ case _: PhysicalStringType =>
Some(value.asInstanceOf[UTF8String].numBytes())
+ case PhysicalBinaryType => Some(value.asInstanceOf[Array[Byte]].length)
+ case _: PhysicalBinaryViewType =>
Some(value.asInstanceOf[BinaryView].numBytes())
+ case PhysicalArrayType(et, _) =>
+ val array = value.asInstanceOf[ArrayData]
+ var size = 0
+ var i = 0
+ while (i < array.numElements()) {
+ valueSizeInBytes(array.get(i, et), et) match {
+ case Some(elementSize) => size += elementSize
+ case None => return None
+ }
+ i += 1
+ }
+ Some(size)
+ case PhysicalMapType(kt, vt, _) =>
+ val map = value.asInstanceOf[MapData]
+ for {
+ keySize <- valueSizeInBytes(map.keyArray(), ArrayType(kt))
+ valueSize <- valueSizeInBytes(map.valueArray(), ArrayType(vt))
+ } yield keySize + valueSize
+ case st: PhysicalStructType =>
+ val row = value.asInstanceOf[InternalRow]
+ var size = 0
+ var i = 0
+ while (i < st.fields.length) {
+ val fieldType = st.fields(i).dataType
+ valueSizeInBytes(if (row.isNullAt(i)) null else row.get(i,
fieldType), fieldType) match {
+ case Some(fieldSize) => size += fieldSize
+ case None => return None
+ }
+ i += 1
+ }
+ Some(size)
+ case _ => None
+ }
+ }
+
/**
* Inverse of [[Literal.sql]]
*/
@@ -450,6 +504,12 @@ case class Literal (value: Any, dataType: DataType)
extends LeafExpression {
override def nullable: Boolean = value == null
+ /**
+ * The size in bytes of this literal's value as a safe upper bound, or
`None` when it cannot be
+ * determined reliably. See [[Literal.valueSizeInBytes]] for the exact
contract.
+ */
+ lazy val valueSizeInBytes: Option[Int] = Literal.valueSizeInBytes(value,
dataType)
+
private def timeZoneId =
DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone)
override lazy val treePatternBits: BitSet = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
index 66c4a39ce823..71f30ca49d86 100755
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
@@ -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
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
index 13e3cb76805d..edd63829a711 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
@@ -323,20 +323,16 @@ trait JoinSelectionHelper extends Logging {
if (hintOnly) {
hintToShuffleHashJoinLeft(join.hint)
} else {
- hintToPreferShuffleHashJoinLeft(join.hint) ||
- (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(join.left,
conf) &&
- muchSmaller(join.left, join.right, conf)) ||
- forceApplyShuffledHashJoin(conf)
+ (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(join.left,
conf) &&
+ muchSmaller(join.left, join.right, conf)) ||
forceApplyShuffledHashJoin(conf)
}
}
def shouldBuildRight(): Boolean = {
if (hintOnly) {
hintToShuffleHashJoinRight(join.hint)
} else {
- hintToPreferShuffleHashJoinRight(join.hint) ||
- (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(join.right,
conf) &&
- muchSmaller(join.right, join.left, conf)) ||
- forceApplyShuffledHashJoin(conf)
+ (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(join.right,
conf) &&
+ muchSmaller(join.right, join.left, conf)) ||
forceApplyShuffledHashJoin(conf)
}
}
getBuildSide(
@@ -473,18 +469,6 @@ trait JoinSelectionHelper extends Logging {
hint.rightHint.exists(_.strategy.contains(SHUFFLE_HASH))
}
- def hintToPreferShuffleHashJoinLeft(hint: JoinHint): Boolean = {
- hint.leftHint.exists(_.strategy.contains(PREFER_SHUFFLE_HASH))
- }
-
- def hintToPreferShuffleHashJoinRight(hint: JoinHint): Boolean = {
- hint.rightHint.exists(_.strategy.contains(PREFER_SHUFFLE_HASH))
- }
-
- def hintToPreferShuffleHashJoin(hint: JoinHint): Boolean = {
- hintToPreferShuffleHashJoinLeft(hint) ||
hintToPreferShuffleHashJoinRight(hint)
- }
-
def hintToShuffleHashJoin(hint: JoinHint): Boolean = {
hintToShuffleHashJoinLeft(hint) || hintToShuffleHashJoinRight(hint)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
index b7cd36f82db6..bb316f5683d8 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
@@ -188,14 +188,6 @@ case object NO_BROADCAST_HASH extends JoinStrategyHint {
override def hintAliases: Set[String] = Set.empty
}
-/**
- * An internal hint to encourage shuffle hash join, used by adaptive query
execution.
- */
-case object PREFER_SHUFFLE_HASH extends JoinStrategyHint {
- override def displayName: String = "prefer_shuffle_hash"
- override def hintAliases: Set[String] = Set.empty
-}
-
/**
* An internal hint to prohibit broadcasting and replicating one side of a
join. This hint is used
* by some rules where broadcasting or replicating a particular side of the
join is not permitted,
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 aca325a43e85..132d38faac7c 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
@@ -1361,6 +1361,65 @@ 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).
This is the master " +
+ "switch for the conversion. By default it only looks through the
join's own required " +
+ "sort to reach a direct input shuffle; set " +
+
"spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled
" +
+ "to true to also look through non-shuffle operators (aggregate,
project, filter, " +
+ "window, etc.) sitting between the join and its input shuffle.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
+ val
ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_LOOK_THROUGH_OPERATORS_ENABLED
=
+ buildConf(
+
"spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled")
+ .doc("When true, the sort merge join to shuffled hash join conversion
(see " +
+
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key})
additionally " +
+ "looks through non-shuffle operators (aggregate, project, filter,
window, etc.) sitting " +
+ "between the join and its input shuffle, instead of only the join's
own required sort. " +
+ s"Has no effect unless " +
+
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key} is
true.")
+ .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 when the
sort-merge-join to " +
+ "shuffled-hash-join conversion bounds 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 =
+ 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 " +
+ "example a shuffled hash join that " +
+
s"${ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key}
produced from a " +
+ "sort merge join when the conversion does not push extra sorts
elsewhere. Defaults to " +
+
"spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled
" +
+ "so that it is enabled together with the look-through conversion.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .fallbackConf(
+
ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_LOOK_THROUGH_OPERATORS_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 " +
@@ -8238,6 +8297,18 @@ 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 convertSortMergeJoinToShuffledHashJoinLookThroughOperatorsEnabled:
Boolean =
+
getConf(ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_LOOK_THROUGH_OPERATORS_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)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala
index 58fff9bad849..85784579ad6d 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.types.DayTimeIntervalType._
import org.apache.spark.sql.types.YearMonthIntervalType._
-import org.apache.spark.unsafe.types.{BinaryView, CalendarInterval,
TimestampNanosVal, UTF8String}
+import org.apache.spark.unsafe.types.{BinaryView, CalendarInterval,
TimestampNanosVal, UTF8String, VariantVal}
class LiteralExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
@@ -855,4 +855,52 @@ class LiteralExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
assert(Literal(UTF8String.fromString("x"), StringType("UTF8_LCASE")).sql
===
"'x' collate UTF8_LCASE")
}
+
+ test("valueSizeInBytes") {
+ // A null value reports the type's default size.
+ assert(Literal.create(null, StringType).valueSizeInBytes ===
Some(StringType.defaultSize))
+ assert(Literal.create(null, IntegerType).valueSizeInBytes ===
Some(IntegerType.defaultSize))
+
+ // Fixed-length types report their default size.
+ assert(Literal(1).valueSizeInBytes === Some(IntegerType.defaultSize))
+ assert(Literal(1L).valueSizeInBytes === Some(LongType.defaultSize))
+ assert(Literal(1.0).valueSizeInBytes === Some(DoubleType.defaultSize))
+ assert(Literal(true).valueSizeInBytes === Some(BooleanType.defaultSize))
+ assert(Literal(Decimal(1), DecimalType(10, 0)).valueSizeInBytes ===
+ Some(DecimalType(10, 0).defaultSize))
+
+ // Variable-length string / binary report their real byte length, not the
default size.
+ assert(Literal(UTF8String.fromString(""), StringType).valueSizeInBytes ===
Some(0))
+ assert(Literal(UTF8String.fromString("abc"), StringType).valueSizeInBytes
=== Some(3))
+ // A multi-byte UTF-8 character counts its encoded bytes (U+00E9 encodes
to 2 bytes). Build the
+ // character with `Character.toChars` rather than a unicode escape:
scalariform decodes such an
+ // escape (even in a comment) before scalastyle's NonASCIICharacterChecker
sees it, so it would
+ // trip the nonascii lint despite the source being pure ASCII.
+ assert(Literal(UTF8String.fromString(new String(Character.toChars(0xE9))),
StringType)
+ .valueSizeInBytes === Some(2))
+ assert(Literal(Array[Byte](1, 2, 3, 4), BinaryType).valueSizeInBytes ===
Some(4))
+
+ // Array sums element sizes.
+ assert(Literal.create(Array("a", "bc"),
ArrayType(StringType)).valueSizeInBytes === Some(3))
+ assert(Literal.create(Array(1, 2, 3),
ArrayType(IntegerType)).valueSizeInBytes ===
+ Some(3 * IntegerType.defaultSize))
+
+ // Map sums key and value sizes.
+ assert(Literal.create(Map("a" -> "bc"), MapType(StringType,
StringType)).valueSizeInBytes ===
+ Some(3))
+
+ // Struct sums field sizes.
+ val structType = StructType(Seq(
+ StructField("s", StringType), StructField("i", IntegerType)))
+ assert(Literal.create(Row("abc", 1), structType).valueSizeInBytes ===
+ Some(3 + IntegerType.defaultSize))
+
+ // CalendarInterval reports its (fixed) default size.
+ assert(Literal(new CalendarInterval(1, 2, 3),
CalendarIntervalType).valueSizeInBytes ===
+ Some(CalendarIntervalType.defaultSize))
+
+ // An unmeasurable variable-length type (e.g. variant) returns None.
+ assert(Literal(new VariantVal(Array[Byte](1, 2), Array[Byte](3)),
VariantType).valueSizeInBytes
+ === None)
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 752ba33bdad2..f8f1e9eeeb35 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -206,8 +206,7 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
}
private def checkHintNonEquiJoin(hint: JoinHint): Unit = {
- if (hintToShuffleHashJoin(hint) || hintToPreferShuffleHashJoin(hint) ||
- hintToSortMergeJoin(hint)) {
+ if (hintToShuffleHashJoin(hint) || hintToSortMergeJoin(hint)) {
assert(hint.leftHint.orElse(hint.rightHint).isDefined)
hintErrorHandler.joinHintNotSupported(hint.leftHint.orElse(hint.rightHint).get,
"no equi-join keys")
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
index 3c23930090ab..0a2c2090e7eb 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AQEOptimizer.scala
@@ -41,7 +41,7 @@ class AQEOptimizer(conf: SQLConf,
extendedRuntimeOptimizerRules: Seq[Rule[Logica
AQEPropagateEmptyRelation,
ConvertToLocalRelation,
UpdateAttributeNullability),
- Batch("Dynamic Join Selection", Once, DynamicJoinSelection),
+ Batch("Demote Broadcast Hash Join", Once, DemoteBroadcastHashJoin),
Batch("Eliminate Limits", fixedPoint, EliminateLimits),
Batch("Optimize One Row Plan", fixedPoint, OptimizeOneRowPlan)) :+
Batch("User Provided Runtime Optimizers", fixedPoint,
extendedRuntimeOptimizerRules: _*) :+
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
index 9a483076ff56..7040ab51cf51 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
@@ -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
@@ -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,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ConvertSortMergeJoinToShuffledHashJoin.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ConvertSortMergeJoinToShuffledHashJoin.scala
new file mode 100644
index 000000000000..734404f210bb
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ConvertSortMergeJoinToShuffledHashJoin.scala
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.adaptive
+
+import scala.annotation.tailrec
+
+import org.apache.spark.MapOutputStatistics
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, CaseWhen,
Cast, Coalesce, Expression, If, Literal, String2TrimExpression, Substring,
UnsafeRow}
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
JoinSelectionHelper}
+import org.apache.spark.sql.catalyst.plans.LeftExistence
+import org.apache.spark.sql.catalyst.plans.logical.{Join, SHUFFLE_MERGE}
+import
org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.{CollectMetricsExec, FilterExec,
ProjectExec, SortExec, SparkPlan}
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS,
EnsureRequirements}
+import org.apache.spark.sql.execution.joins.{BaseJoinExec,
ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.window.{WindowExecBase,
WindowGroupLimitExec}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Converts a [[SortMergeJoinExec]] into a [[ShuffledHashJoinExec]] during
adaptive execution when
+ * a build side's materialized shuffle statistics show it is small enough for
a local hash map.
+ *
+ * This runs on the physical plan and owns the shuffled-hash-over-sort-merge
selection that AQE
+ * makes from materialized shuffle statistics. It is gated by
+ * `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled`
(default true, the master
+ * switch), and has two modes:
+ * - Default: it looks through the sort merge join's own required
[[SortExec]] to reach a
+ * *direct* input shuffle.
+ * - Behind
`...convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled`: it
+ * additionally looks through non-shuffle operators (aggregate, project,
filter, window,
+ * left-existence join) sitting above the shuffle.
+ *
+ * The swap is shuffle-free since both joins are `ShuffledJoin`s with the same
distribution and
+ * partitioning; only the child sorts become unnecessary. As a shuffled hash
join loses the sort
+ * merge join's output ordering, [[EnsureRequirements]] is re-run to restore
any ordering an
+ * ancestor still needs, and AQE's [[CostEvaluator]] decides whether to adopt
the converted plan.
+ *
+ * A shuffled hash join builds a non-spillable local hash map, so the
traversed operators must not
+ * blow up the build size that the input shuffle statistics estimate:
+ * - the traversal only looks through an operator whose output expressions
are all size-bounded
+ * (see [[isSizeBoundedExpr]]), so no operator can widen a row in a way
the shuffle statistics
+ * cannot see; and
+ * - the build-side estimate is scaled by [[wideningFactor]] to account for
the width change the
+ * traversed operators do introduce.
+ */
+case class ConvertSortMergeJoinToShuffledHashJoin(ensureRequirements:
EnsureRequirements)
+ extends Rule[SparkPlan] with JoinSelectionHelper {
+
+ private def preferShuffledHashJoin(
+ mapStats: MapOutputStatistics,
+ sizeInBytesFactor: Double): 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)
+ }
+
+ /**
+ * The estimated per-row byte-size ratio of the build subtree's output to
its input shuffle's
+ * output, i.e. how much the traversed operators widen each row. The
traversed operators never
+ * increase the row count (`N_build <= N_shuffle`), so scaling the input
shuffle bytes by this
+ * ratio keeps them a valid upper bound on the hash-map build size once row
width is accounted
+ * for: `buildSize = N_build * buildRowWidth <= shuffleBytes *
(buildRowWidth / shuffleRowWidth)`.
+ *
+ * Floored at
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.minWideningFactor`
+ * (default 1.0). Unlike `SizeInBytesOnlyStatsPlanVisitor`, which computes a
best-effort size and
+ * lets a narrowing operator shrink it, the default keeps a conservative
bound for a non-spillable
+ * build: `getSizePerRow` under-estimates a variable-width column (it uses
`defaultSize`), so a
+ * `factor < 1` could push the scaled bytes below the real build size and
reintroduce the
+ * out-of-memory risk, whereas the raw shuffle bytes are always a valid
bound when the build side
+ * is no wider than the shuffle row. Raising the floor above 1.0 is more
conservative still.
+ */
+ private def wideningFactor(buildOutput: Seq[Attribute], shuffleOutput:
Seq[Attribute]): Double = {
+ val buildRowSize = EstimationUtils.getSizePerRow(buildOutput).toDouble
+ val shuffleRowSize = EstimationUtils.getSizePerRow(shuffleOutput).toDouble
+ math.max(conf.convertSortMergeJoinToShuffledHashJoinMinWideningFactor,
+ buildRowSize / shuffleRowSize)
+ }
+
+ private def hasSortMergeJoinHint(smj: SortMergeJoinExec): Boolean =
smj.logicalLink.exists {
+ case j: Join =>
+ j.hint.leftHint.exists(_.strategy.contains(SHUFFLE_MERGE)) ||
+ j.hint.rightHint.exists(_.strategy.contains(SHUFFLE_MERGE))
+ case _ => false
+ }
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ if (!conf.convertSortMergeJoinToShuffledHashJoinEnabled) {
+ return plan
+ }
+ val lookThroughOperatorsEnabled =
+ conf.convertSortMergeJoinToShuffledHashJoinLookThroughOperatorsEnabled
+ val optimizedPlan = plan.transformUp {
+ case smj @ SortMergeJoinExec(leftKeys, rightKeys, joinType, condition,
left, right, false)
+ // Do not convert if the join keys are not hash-join-compatible
(e.g. collated or other
+ // non-binary-stable string keys), since a hash join matches keys by
binary equality and
+ // would return wrong results. This mirrors the guard on the other
SHJ-planning paths.
+ if !hasSortMergeJoinHint(smj) && hashJoinSupported(leftKeys,
rightKeys) =>
+ val leftStage = findShuffleStage(left, lookThroughOperatorsEnabled)
+ val rightStage = findShuffleStage(right, lookThroughOperatorsEnabled)
+ // `wideningFactor` scales the input shuffle bytes by how much the
traversed operators
+ // widen each row, so its second argument must be the input shuffle
stage's output (not the
+ // join child's own output, which would always yield a ratio of 1.0).
Compute it only when
+ // the stage is known-defined.
+ val leftFactor = leftStage.map(s => wideningFactor(smj.left.output,
s.output))
+ val rightFactor = rightStage.map(s => wideningFactor(smj.right.output,
s.output))
+ val canBuildLeft = leftStage.isDefined &&
canBuildShuffledHashJoinLeft(smj.joinType) &&
+ preferShuffledHashJoin(leftStage.get.mapStats.get, leftFactor.get)
+ val canBuildRight = rightStage.isDefined &&
canBuildShuffledHashJoinRight(smj.joinType) &&
+ preferShuffledHashJoin(rightStage.get.mapStats.get, rightFactor.get)
+ val buildSide = if (canBuildLeft && canBuildRight) {
+ val leftSize = leftStage.get.mapStats.get.bytesByPartitionId.sum *
leftFactor.get
+ val rightSize = rightStage.get.mapStats.get.bytesByPartitionId.sum *
rightFactor.get
+ if (leftSize < rightSize) Some(BuildLeft) else Some(BuildRight)
+ } else if (canBuildLeft) {
+ Some(BuildLeft)
+ } else if (canBuildRight) {
+ Some(BuildRight)
+ } else {
+ None
+ }
+
+ buildSide match {
+ case Some(buildSide) =>
+ ShuffledHashJoinExec(leftKeys, rightKeys, joinType, buildSide,
condition,
+ stripSort(smj.left), stripSort(smj.right))
+ case None => smj
+ }
+ }
+ if (optimizedPlan.fastEquals(plan)) {
+ plan
+ } else {
+ // A shuffled hash join does not preserve the sort merge join's output
ordering. Re-run
+ // EnsureRequirements so any ordering an ancestor still needs is
re-established, keeping the
+ // plan valid. AQE's CostEvaluator then decides between this plan and
the current one.
+ ensureRequirements.apply(optimizedPlan)
+ }
+ }
+
+ /**
+ * Drops a top-level [[SortExec]] since a shuffled hash join does not
require sorted input;
+ * [[RemoveRedundantSorts]] cleans up any remaining redundant sorts
afterwards.
+ */
+ private def stripSort(plan: SparkPlan): SparkPlan = plan match {
+ case s: SortExec if !s.global => s.child
+ case other => other
+ }
+
+ /**
+ * Finds a join child's input shuffle. Descent stops at the first
[[ShuffleQueryStageExec]], which
+ * is thus guaranteed to be the join's own input shuffle whose statistics
bound (or, for a
+ * reducing aggregate, upper-bound) the build side. The stage must be
materialized with stats and
+ * originate from [[EnsureRequirements]], so swapping the join type does not
change the shuffle.
+ *
+ * The traversal has two modes:
+ * - Default: it looks through the sort merge join's own required
[[SortExec]] to reach a
+ * *direct* input shuffle.
+ * - Behind
`...convertSortMergeJoinToShuffledHashJoin.lookThroughOperators.enabled`: it
+ * additionally looks through non-shuffle operators (aggregate, project,
filter, window,
+ * left-existence join) sitting above the shuffle.
+ *
+ * A [[ProjectExec]], [[BaseAggregateExec]] or [[WindowExecBase]] is only
traversed when all of
+ * its output expressions are size-bounded (see [[isSizeBoundedExpr]]);
otherwise the shuffle
+ * bytes could badly under-estimate the non-spillable hash-map build size
(e.g.
+ * `repeat(max(c2), 10000)` above a small shuffle), so descent stops and the
join is left as is.
+ */
+ @tailrec
+ private def findShuffleStage(
+ plan: SparkPlan,
+ lookThroughOperatorsEnabled: Boolean): Option[ShuffleQueryStageExec] =
plan match {
+ case s: ShuffleQueryStageExec if s.isMaterialized && s.mapStats.isDefined
&&
+ s.shuffle.shuffleOrigin == ENSURE_REQUIREMENTS => Some(s)
+ // Always on: look through the join's own required sort to reach a
direct input shuffle.
+ case _: SortExec => findShuffleStage(plan.children.head,
lookThroughOperatorsEnabled)
+ // The look-through capability below is gated by its own config.
+ case _ if !lookThroughOperatorsEnabled => None
+ case _: FilterExec | _: WindowGroupLimitExec | _: CollectMetricsExec =>
+ findShuffleStage(plan.children.head, lookThroughOperatorsEnabled)
+ case p: ProjectExec if p.projectList.forall(isSizeBoundedExpr) =>
+ findShuffleStage(p.child, lookThroughOperatorsEnabled)
+ case a: BaseAggregateExec if a.resultExpressions.forall(isSizeBoundedExpr)
=>
+ findShuffleStage(a.child, lookThroughOperatorsEnabled)
+ case w: WindowExecBase if w.windowExpression.forall(isSizeBoundedExpr) =>
+ findShuffleStage(w.child, lookThroughOperatorsEnabled)
+ case join: BaseJoinExec =>
+ join.joinType match {
+ case LeftExistence(_) => findShuffleStage(join.left,
lookThroughOperatorsEnabled)
+ case _ => None
+ }
+ case _ => None
+ }
+
+ /**
+ * Whether `expr`'s result byte-size is bounded by the values it reads, so
it cannot widen a row.
+ * An operator all of whose outputs are size-bounded keeps the input shuffle
bytes a valid bound
+ * on the non-spillable hash-map build size; an unbounded output (e.g.
`repeat` or `concat`, which
+ * synthesize a wider value) makes the shuffle bytes an under-estimate and
stops the traversal.
+ *
+ * An [[Attribute]] is always bounded: it refers to a value produced by a
descendant operator.
+ * The traversal checks every operator down to the input shuffle, so if a
descendant synthesized a
+ * wide value (e.g. a lower `ProjectExec` with `repeat(...)`) this rule
stops there; by induction
+ * any attribute that survives is grounded in the shuffle output. Note that
aggregate functions do
+ * not appear inline here - a physical aggregate exposes them as result
attributes - so an
+ * aggregate result (`max`, and equally an accumulating `collect_list` whose
bytes are already in
+ * the shuffle below) is bounded through this same [[Attribute]] case.
+ *
+ * A fixed-width result ([[UnsafeRow.isFixedLength]], stored in an 8-byte
word) is bounded
+ * regardless of inputs. A variable-width [[Literal]] is bounded only when
its actual byte size is
+ * no larger than the data type's `defaultSize` that [[wideningFactor]]
assumes; otherwise a large
+ * folded constant (e.g. `repeat('x', 100000)` constant-folded into a
`Literal`) would slip past
+ * the size estimate. Beyond that, only a whitelist of length-non-increasing
transforms over
+ * bounded children is accepted; anything else (e.g. `repeat`, `concat`,
`upper`/`lower` - Unicode
+ * case mapping can grow the UTF-8 byte length - arithmetic on strings) is
treated as potentially
+ * widening.
+ */
+ private def isSizeBoundedExpr(expr: Expression): Boolean = {
+ if (UnsafeRow.isFixedLength(expr.dataType)) {
+ return true
+ }
+ expr match {
+ case _: Attribute => true
+ // A variable-width literal is bounded only when its actual byte size is
known and no larger
+ // than the data type's `defaultSize` that `wideningFactor` assumes;
otherwise a large folded
+ // constant (e.g. `repeat('x', 100000)` constant-folded into a
`Literal`) would slip past the
+ // estimate. An unmeasurable literal (`valueSizeInBytes` is None) is
treated as widening.
+ case lit: Literal => lit.valueSizeInBytes.exists(_ <=
lit.dataType.defaultSize)
+ case e: Alias => isSizeBoundedExpr(e.child)
+ // Cast is a very common expression; it may slightly increase the size
in bytes
+ // but should be tolerated.
+ case e: Cast => isSizeBoundedExpr(e.child)
+ case e: Substring => isSizeBoundedExpr(e.str)
+ case e: String2TrimExpression => isSizeBoundedExpr(e.srcStr)
+ // Conditionals only pick one of their (bounded) branch values.
+ case If(_, t, f) => isSizeBoundedExpr(t) && isSizeBoundedExpr(f)
+ case CaseWhen(branches, elseValue) =>
+ branches.forall(b => isSizeBoundedExpr(b._2)) &&
elseValue.forall(isSizeBoundedExpr)
+ case Coalesce(children) => children.forall(isSizeBoundedExpr)
+ case _ => false
+ }
+ }
+}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DemoteBroadcastHashJoin.scala
similarity index 72%
rename from
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala
rename to
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DemoteBroadcastHashJoin.scala
index 217569ae645c..f86354247f56 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DemoteBroadcastHashJoin.scala
@@ -21,21 +21,15 @@ import org.apache.spark.MapOutputStatistics
import org.apache.spark.sql.catalyst.optimizer.JoinSelectionHelper
import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftOuter, RightOuter}
-import org.apache.spark.sql.catalyst.plans.logical.{HintInfo, Join,
JoinStrategyHint, LogicalPlan, NO_BROADCAST_HASH, PREFER_SHUFFLE_HASH,
SHUFFLE_HASH}
+import org.apache.spark.sql.catalyst.plans.logical.{HintInfo, Join,
JoinStrategyHint, LogicalPlan, NO_BROADCAST_HASH}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.internal.SQLConf
/**
- * This optimization rule includes three join selection:
- * 1. detects a join child that has a high ratio of empty partitions and
adds a
- * NO_BROADCAST_HASH hint to avoid it being broadcast, as shuffle join is
faster in this case:
- * many tasks complete immediately since one join side is empty.
- * 2. detects a join child that every partition size is less than local map
threshold and adds a
- * PREFER_SHUFFLE_HASH hint to encourage being shuffle hash join instead
of sort merge join.
- * 3. if a join satisfies both NO_BROADCAST_HASH and PREFER_SHUFFLE_HASH,
- * then add a SHUFFLE_HASH hint.
+ * This optimization rule detects a join child that has a high ratio of empty
partitions and adds a
+ * NO_BROADCAST_HASH hint to avoid it being broadcast, as shuffle join is
faster in this case: many
+ * tasks complete immediately since one join side is empty.
*/
-object DynamicJoinSelection extends Rule[LogicalPlan] with JoinSelectionHelper
{
+object DemoteBroadcastHashJoin extends Rule[LogicalPlan] with
JoinSelectionHelper {
private def hasManyEmptyPartitions(mapStats: MapOutputStatistics): Boolean =
{
val partitionCnt = mapStats.bytesByPartitionId.length
@@ -44,14 +38,6 @@ object DynamicJoinSelection extends Rule[LogicalPlan] with
JoinSelectionHelper {
(nonZeroCnt * 1.0 / partitionCnt) <
conf.nonEmptyPartitionRatioForBroadcastJoin
}
- private def preferShuffledHashJoin(mapStats: MapOutputStatistics): 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(_ <=
maxShuffledHashJoinLocalMapThreshold)
- }
-
private def selectJoinStrategy(
join: Join,
isLeft: Boolean): Option[JoinStrategyHint] = {
@@ -89,13 +75,8 @@ object DynamicJoinSelection extends Rule[LogicalPlan] with
JoinSelectionHelper {
false
}
- val preferShuffleHash = preferShuffledHashJoin(stage.mapStats.get)
- if (demoteBroadcastHash && preferShuffleHash) {
- Some(SHUFFLE_HASH)
- } else if (demoteBroadcastHash) {
+ if (demoteBroadcastHash) {
Some(NO_BROADCAST_HASH)
- } else if (preferShuffleHash) {
- Some(PREFER_SHUFFLE_HASH)
} else {
None
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala
index 28b757114ebe..84da42ff68a6 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala
@@ -18,18 +18,33 @@
package org.apache.spark.sql.execution.adaptive
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
ShuffledJoin}
/**
- * A simple implementation of [[Cost]], which takes a number of [[Long]] as
the cost value.
+ * A simple implementation of [[Cost]] produced by [[SimpleCostEvaluator]].
Its three components are
+ * compared lexicographically in priority order:
+ * 1. `numSkewJoins`: more skew joins means lower cost, so it is compared
descending and first;
+ * 2. `numShuffles`: fewer shuffles means lower cost;
+ * 3. `numLocalSorts`: the lowest-priority tiebreaker, so among plans with
the same number of skew
+ * joins and shuffles the one with fewer local sorts is preferred (e.g. a
shuffled hash join
+ * over a sort merge join when the conversion does not push extra sorts
elsewhere).
+ *
+ * `numSkewJoins` and `numLocalSorts` are `0` when the corresponding feature
is disabled in the
+ * evaluator, so they do not affect the comparison in that case.
*/
-case class SimpleCost(value: Long) extends Cost {
+case class SimpleCost(numSkewJoins: Int, numShuffles: Int, numLocalSorts: Int)
extends Cost {
override def compare(that: Cost): Int = that match {
- case SimpleCost(thatValue) =>
- if (value < thatValue) -1 else if (value > thatValue) 1 else 0
+ case SimpleCost(thatSkewJoins, thatShuffles, thatLocalSorts) =>
+ val bySkewJoins = Integer.compare(thatSkewJoins, numSkewJoins)
+ if (bySkewJoins != 0) {
+ bySkewJoins
+ } else {
+ val byShuffles = Integer.compare(numShuffles, thatShuffles)
+ if (byShuffles != 0) byShuffles else Integer.compare(numLocalSorts,
thatLocalSorts)
+ }
case _ =>
throw
QueryExecutionErrors.cannotCompareCostWithTargetCostError(that.toString)
}
@@ -37,24 +52,23 @@ case class SimpleCost(value: Long) extends Cost {
/**
* A skew join aware implementation of [[CostEvaluator]], which counts the
number of
- * [[ShuffleExchangeLike]] nodes and skew join nodes in the plan.
+ * [[ShuffleExchangeLike]] nodes, skew join nodes and (optionally) local
[[SortExec]] nodes in the
+ * plan. See [[SimpleCost]] for how the components are compared.
*/
-case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean) extends
CostEvaluator {
- override def evaluateCost(plan: SparkPlan): Cost = {
- val numShuffles = plan.collect {
- case s: ShuffleExchangeLike => s
- }.size
+case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean,
countLocalSort: Boolean)
+ extends CostEvaluator {
- if (forceOptimizeSkewedJoin) {
- val numSkewJoins = plan.collect {
- case j: ShuffledJoin if j.isSkewJoin => j
- case j: BroadcastHashJoinExec if j.isSkewJoin => j
- }.size
- // We put `-numSkewJoins` in the first 32 bits of the long value, so
that it's compared first
- // when comparing the cost, and larger `numSkewJoins` means lower cost.
- SimpleCost(-numSkewJoins.toLong << 32 | numShuffles)
- } else {
- SimpleCost(numShuffles)
+ override def evaluateCost(plan: SparkPlan): Cost = {
+ var numSkewJoins = 0
+ var numShuffles = 0
+ var numLocalSorts = 0
+ plan.foreach {
+ case j: ShuffledJoin if forceOptimizeSkewedJoin && j.isSkewJoin =>
numSkewJoins += 1
+ case j: BroadcastHashJoinExec if forceOptimizeSkewedJoin && j.isSkewJoin
=> numSkewJoins += 1
+ case _: ShuffleExchangeLike => numShuffles += 1
+ case s: SortExec if countLocalSort && !s.global => numLocalSorts += 1
+ case _ =>
}
+ SimpleCost(numSkewJoins, numShuffles, numLocalSorts)
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryPlanningTrackerEndToEndSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryPlanningTrackerEndToEndSuite.scala
index c23d4e354c8a..0b5307b467ff 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryPlanningTrackerEndToEndSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryPlanningTrackerEndToEndSuite.scala
@@ -87,13 +87,13 @@ class QueryPlanningTrackerEndToEndSuite extends StreamTest {
test("SPARK-57212: Track sub-query AQE rules") {
// The main query has no shuffle, so its AQE never re-optimizes and thus
never runs the AQE
// logical optimizer. The scalar sub-query does have a shuffle, so its
sub-AQE (planned on a
- // separate thread) re-optimizes and runs `DynamicJoinSelection`. That
rule is only visible in
- // the query tracker if every `AdaptiveSparkPlanExec` records into the
shared query tracker.
+ // separate thread) re-optimizes and runs `DemoteBroadcastHashJoin`. That
rule is only visible
+ // in the query tracker if every `AdaptiveSparkPlanExec` records into the
shared query tracker.
val df = spark.sql("SELECT id FROM range(10) WHERE id > (SELECT count(*)
FROM range(1000))")
df.collect()
val ruleNames = df.queryExecution.tracker.rules.keySet
assert(ruleNames.contains(
- "org.apache.spark.sql.execution.adaptive.DynamicJoinSelection"))
+ "org.apache.spark.sql.execution.adaptive.DemoteBroadcastHashJoin"))
}
test("The start times should be in order: parsing <= analysis <=
optimization <= planning") {
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 c3b8de4f37c1..4b57546475ac 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
@@ -29,11 +29,11 @@ import org.apache.spark.scheduler.{SparkListener,
SparkListenerEvent, SparkListe
import org.apache.spark.shuffle.sort.SortShuffleManager
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, EqualTo, IsNull, Or}
+import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute,
AttributeReference, EqualTo, IsNull, Or, SortOrder}
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Join, JoinHint,
LocalRelation, LogicalPlan}
-import
org.apache.spark.sql.catalyst.plans.physical.CoalescedNullAwareHashPartitioning
+import
org.apache.spark.sql.catalyst.plans.physical.{CoalescedNullAwareHashPartitioning,
SinglePartition}
import org.apache.spark.sql.classic.Strategy
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
@@ -50,6 +50,7 @@ import
org.apache.spark.sql.execution.ui.{SparkListenerSQLAdaptiveExecutionUpdat
import org.apache.spark.sql.execution.window.WindowExec
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
+import
org.apache.spark.sql.internal.SQLConf.{ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_LOOK_THROUGH_OPERATORS_ENABLED
=> LOOK_THROUGH_OPERATORS}
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.streaming.{OutputMode, StatefulProcessor,
TimeMode, TimerValues, TTLConfig, ValueState}
import org.apache.spark.sql.test.SharedSparkSession
@@ -70,6 +71,10 @@ class AdaptiveQueryExecSuite
setupTestData()
+ // Short alias for the long config key, to keep the SMJ-to-SHJ conversion
tests within the line
+ // length limit.
+ private val lookThroughOperatorsKey = LOOK_THROUGH_OPERATORS.key
+
override protected def sparkConf =
super.sparkConf.set(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key,
"0")
@@ -2461,6 +2466,447 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-58084: Convert sort merge join to shuffled hash join through
operators") {
+ withTempView("t1", "t2", "t3") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The t2 side has a non-shuffle operator (aggregate, optionally with a
filter) between the
+ // join and its input shuffle, so the default direct-shuffle path does
not reach the shuffle;
+ // only the look-through mode converts it.
+ val queries = Seq(
+ "SELECT t1.c1, x.cnt FROM t1 JOIN " +
+ "(SELECT c1, count(*) AS cnt FROM t2 GROUP BY c1) x ON t1.c1 = x.c1",
+ "SELECT t1.c1, x.cnt FROM t1 JOIN " +
+ "(SELECT c1, count(*) AS cnt FROM t2 GROUP BY c1 HAVING count(*) >=
0) x " +
+ "ON t1.c1 = x.c1")
+
+ // t1 partition size: [926, 729, 731]; t2 (aggregated) side: [372, 126,
0]. With a small
+ // advisory partition size and a local map threshold of 500, only the t2
side has all
+ // partitions within the threshold, so the join is converted with the t2
side as build side.
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500") {
+ queries.foreach { query =>
+ // Look-through enabled: the sort merge join is converted to a
shuffled hash join.
+ withSQLConf(
+ lookThroughOperatorsKey -> "true") {
+ val (origin, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelSortMergeJoin(origin).size === 1)
+ val shj = findTopLevelShuffledHashJoin(adaptive)
+ assert(shj.size === 1, s"expected a shuffled hash join for query:
$query")
+ assert(shj.head.buildSide == BuildRight)
+ assert(findTopLevelSortMergeJoin(adaptive).isEmpty)
+ }
+ // Look-through disabled (default): the aggregate blocks the
direct-shuffle path, so the
+ // join stays a sort merge join.
+ withSQLConf(
+ lookThroughOperatorsKey -> "false") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ s"expected no shuffled hash join for query: $query")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1,
+ s"expected a sort merge join for query: $query")
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-58084: Do not convert when an operator adds a variable-width
column") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The shuffle below the aggregate is tiny, but the aggregate widens
each build row with a
+ // large variable-width string (`repeat(max(c2), 500)`), so the shuffle
bytes badly
+ // under-estimate the non-spillable hash-map build size. The traversal
must stop at that
+ // widening operator and leave the join as a sort merge join, even
though the shuffle looks
+ // small enough for a local hash map. The wide column is selected in the
output so column
+ // pruning cannot drop it before the join.
+ val query =
+ "SELECT t1.c1, x.wide FROM t1 JOIN " +
+ "(SELECT c1, repeat(max(c2), 500) AS wide FROM t2 GROUP BY c1) x ON
t1.c1 = x.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500",
+ lookThroughOperatorsKey -> "true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ "a widening aggregate above the shuffle must keep the join as a sort
merge join")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: Convert through size-bounded (non-widening) operators") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The aggregate emits a variable-width string column, but only through
size-bounded
+ // expressions: `max` selects an existing value, `cast` and `substring`
cannot widen it. The
+ // traversal must look through them and still convert the join, unlike
the `repeat(...)` case.
+ val query =
+ "SELECT t1.c1, x.m, x.s FROM t1 JOIN " +
+ "(SELECT c1, substring(max(c2), 1, 1) AS m, cast(count(*) AS string)
AS s " +
+ "FROM t2 GROUP BY c1) x ON t1.c1 = x.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"100000",
+ lookThroughOperatorsKey -> "true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ val shj = findTopLevelShuffledHashJoin(adaptive)
+ assert(shj.size === 1,
+ "size-bounded operators above the shuffle must not block the
conversion")
+ assert(shj.head.buildSide == BuildRight)
+ assert(findTopLevelSortMergeJoin(adaptive).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-58084: MinWideningFactor makes the size bound more
conservative") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The t2 (aggregated) build side fits the local map threshold at the
default widening factor,
+ // so the join converts. A large minWideningFactor scales the estimated
build size past the
+ // threshold, so the conversion is rejected and the join stays a sort
merge join.
+ val query =
+ "SELECT t1.c1, x.cnt FROM t1 JOIN " +
+ "(SELECT c1, count(*) AS cnt FROM t2 GROUP BY c1) x ON t1.c1 = x.c1"
+
+ def convertsWith(minWideningFactor: String): Boolean = {
+ var converted = false
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500",
+ lookThroughOperatorsKey -> "true",
+
SQLConf.ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_MIN_WIDENING_FACTOR.key
->
+ minWideningFactor) {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ converted = findTopLevelShuffledHashJoin(adaptive).nonEmpty
+ }
+ converted
+ }
+
+ // Default factor: the build side fits, so the join converts.
+ assert(convertsWith("1.0"), "the join should convert at the default
widening factor")
+ // A large factor scales the estimated build size past the threshold,
rejecting the
+ // conversion.
+ assert(!convertsWith("1000.0"), "a large minWideningFactor should reject
the conversion")
+ }
+ }
+
+ test("SPARK-58084: Widening factor uses the input shuffle's row width, not
the join child's") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The aggregate widens each build row through a size-bounded
`cast(count(*) AS string)`:
+ // the input shuffle row is (c1: int, count: long) = 20 bytes/row, while
the build output is
+ // (c1: int, s: string) = 32 bytes/row, so the true widening factor is
32/20 = 1.6. The build
+ // side's largest shuffle partition is 372 bytes; scaled by 1.6 it is
~595, above the 500-byte
+ // local-map threshold, so the conversion must be rejected. If
`wideningFactor` were computed
+ // against the join child's own output (factor 1.0), the unscaled 372
would fit and the join
+ // would wrongly convert -- this test pins that the input shuffle's
width is used.
+ val query =
+ "SELECT t1.c1, x.s FROM t1 JOIN " +
+ "(SELECT c1, cast(count(*) AS string) AS s FROM t2 GROUP BY c1) x ON
t1.c1 = x.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500",
+ lookThroughOperatorsKey -> "true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ "the widened build side exceeds the threshold, so the join must stay
a sort merge join")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: Convert sort merge join keeps required ordering valid") {
+ withTempView("small1", "small2", "big") {
+ spark.sparkContext.parallelize(
+ (1 to 20).map(i => TestData(i, i.toString)), 4)
+ .toDF("c1", "c2").createOrReplaceTempView("small1")
+ spark.sparkContext.parallelize(
+ (1 to 20).map(i => TestData(i, i.toString)), 4)
+ .toDF("c1", "c2").createOrReplaceTempView("small2")
+ spark.sparkContext.parallelize(
+ (1 to 4000).map(i => TestData(i % 20 + 1, i.toString)), 4)
+ .toDF("c1", "c2").createOrReplaceTempView("big")
+
+ // The inner join over the two small tables is convertible to a shuffled
hash join. The outer
+ // join is pinned to a sort merge join with a MERGE hint, and it
requires its (left) child
+ // ordered on the join key. When the inner join is converted to a
shuffled hash join
+ // (ordering Nil), EnsureRequirements must re-insert the sort above it
so the outer sort merge
+ // join's required ordering is still satisfied and the result is correct.
+ val query = "SELECT /*+ MERGE(big) */ small1.c1 FROM " +
+ "small1 JOIN small2 ON small1.c1 = small2.c1 " +
+ "JOIN big ON small1.c1 = big.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"100000",
+
SQLConf.ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key ->
"true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ // The inner join is converted; the outer join stays a sort merge join
whose (left) child
+ // ordering is re-established by EnsureRequirements, so the plan
remains valid.
+ val smj = findTopLevelSortMergeJoin(adaptive)
+ assert(smj.size === 1)
+ assert(smj.head.left.outputOrdering.nonEmpty,
+ "outer sort merge join must keep its left child ordered on the join
key")
+ assert(findTopLevelShuffledHashJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: SimpleCostEvaluator counts local sorts as a
lower-priority tiebreaker") {
+ def leaf: SparkPlan = CostTestLeafExec()
+ def shuffle(child: SparkPlan): SparkPlan =
ShuffleExchangeExec(SinglePartition, child)
+ def localSort(child: SparkPlan): SparkPlan =
+ SortExec(SortOrder(child.output.head, Ascending) :: Nil, global = false,
child)
+
+ val evaluator = SimpleCostEvaluator(forceOptimizeSkewedJoin = false,
countLocalSort = true)
+ def cost(plan: SparkPlan): Cost = evaluator.evaluateCost(plan)
+
+ // Same number of shuffles: fewer local sorts is cheaper.
+ val oneShuffleTwoSorts = localSort(localSort(shuffle(leaf)))
+ val oneShuffleOneSort = localSort(shuffle(leaf))
+ val oneShuffleNoSort = shuffle(leaf)
+ assert(cost(oneShuffleOneSort).compare(cost(oneShuffleTwoSorts)) < 0)
+ assert(cost(oneShuffleNoSort).compare(cost(oneShuffleOneSort)) < 0)
+
+ // The number of shuffles dominates: a plan with more shuffles is costlier
even with no sorts.
+ val twoShufflesNoSort = shuffle(shuffle(leaf))
+ assert(cost(oneShuffleTwoSorts).compare(cost(twoShufflesNoSort)) < 0)
+
+ // When countLocalSort is disabled, local sorts do not affect the cost.
+ val noSortEvaluator = SimpleCostEvaluator(
+ forceOptimizeSkewedJoin = false, countLocalSort = false)
+ assert(noSortEvaluator.evaluateCost(oneShuffleTwoSorts)
+ .compare(noSortEvaluator.evaluateCost(oneShuffleNoSort)) === 0)
+
+ // Skew join dominates, ahead of shuffles and sorts: with
forceOptimizeSkewedJoin, a plan with
+ // a skew join is cheaper than one without, even if the skew-join plan has
more shuffles and
+ // local sorts.
+ def join(l: SparkPlan, r: SparkPlan, isSkew: Boolean): SparkPlan =
+ SortMergeJoinExec(l.output.take(1), r.output.take(1), Inner, None, l, r,
isSkewJoin = isSkew)
+ val skewEvaluator = SimpleCostEvaluator(forceOptimizeSkewedJoin = true,
countLocalSort = true)
+ // Skew-join plan: 1 skew join, 3 shuffles, 2 local sorts.
+ val withSkewJoin = skewEvaluator.evaluateCost(
+ join(localSort(shuffle(shuffle(leaf))), localSort(shuffle(leaf)), isSkew
= true))
+ // Non-skew plan: 0 skew joins, 2 shuffles, 0 local sorts.
+ val withoutSkewJoin = skewEvaluator.evaluateCost(
+ join(shuffle(leaf), shuffle(leaf), isSkew = false))
+ assert(withSkewJoin.compare(withoutSkewJoin) < 0)
+ }
+
+ test("SPARK-58084: Do not convert sort merge join when it adds local sorts")
{
+ withTempView("big", "small") {
+ spark.sparkContext.parallelize(
+ (1 to 2000).map(i => TestData(i % 20 + 1, i.toString)), 4)
+ .toDF("k", "v").createOrReplaceTempView("big")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 4)
+ .toDF("k", "v").createOrReplaceTempView("small")
+
+ // Both join sides are sort aggregates grouped by the join key, so each
child is already
+ // ordered on the key for free and the sort merge join needs no explicit
child sort. A parent
+ // window partitions by the right join key. A sort merge inner join
keeps both sides' key
+ // orderings, satisfying the window; a shuffled hash join with
build-right keeps only the left
+ // ordering (see HashJoin.outputOrdering), so converting it forces an
extra local sort above
+ // the window. The conversion is therefore only beneficial without
counting local sorts.
+ val query =
+ "SELECT l.k, count(*) OVER (PARTITION BY r.k) c " +
+ "FROM (SELECT k, count(*) c FROM big GROUP BY k) l " +
+ "JOIN (SELECT k, count(*) c FROM small GROUP BY k) r ON l.k = r.k"
+
+ def countLocalSorts(plan: SparkPlan): Int = collect(plan) {
+ case s: SortExec if !s.global => s
+ }.size
+
+ // Force sort aggregate so each join child is ordered on the key for
free.
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.USE_HASH_AGG.key -> "false",
+ SQLConf.USE_OBJECT_HASH_AGG.key -> "false",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500",
+ lookThroughOperatorsKey -> "true") {
+ // Not counting local sorts: the conversion is adopted even though it
adds a local sort.
+
withSQLConf(SQLConf.ADAPTIVE_COST_EVALUATOR_COUNT_LOCAL_SORT_ENABLED.key ->
"false") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).size === 1)
+ assert(findTopLevelSortMergeJoin(adaptive).isEmpty)
+ assert(countLocalSorts(adaptive) == 5)
+ }
+ // Counting local sorts: the converted plan has more local sorts, so
it is rejected and the
+ // sort merge join is kept.
+
withSQLConf(SQLConf.ADAPTIVE_COST_EVALUATOR_COUNT_LOCAL_SORT_ENABLED.key ->
"true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty)
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ assert(countLocalSorts(adaptive) == 4)
+ }
+ }
+ }
+ }
+
+ test("SPARK-58084: Do not convert sort merge join with non-binary-stable
(collated) keys") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, s"v$i")), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, s"v$i")), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // A UTF8_LCASE key is orderable (so a sort merge join is planned) but
not binary-stable. When
+ // the equi-condition wraps the key (here `concat(...)`),
`RewriteCollationJoin` does not
+ // inject a `CollationKey`, so the physical join keys stay
non-binary-stable. A shuffled hash
+ // join matches keys by `UnsafeRow` binary equality, which would return
wrong results, so the
+ // conversion must skip such joins even with the config enabled -
mirroring the
+ // `hashJoinSupported` guard on the other SHJ-planning paths.
+ val query =
+ "SELECT t1.c2 FROM t1 JOIN t2 ON " +
+ "concat(cast(t1.c2 AS STRING COLLATE UTF8_LCASE), 'x') = " +
+ "concat(cast(t2.c2 AS STRING COLLATE UTF8_LCASE), 'x')"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"100000",
+
SQLConf.ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key ->
"true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ "non-binary-stable collated keys must keep the join as a sort merge
join")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: Do not convert sort merge join requested with an explicit
MERGE hint") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // The join is convertible by size, but the user explicitly asked for a
sort merge join with
+ // a MERGE hint. The conversion must respect the hint and keep the sort
merge join, never
+ // overriding an existing SHUFFLE_MERGE join strategy hint.
+ val query = "SELECT /*+ MERGE(t1, t2) */ t1.c1, t2.c2 FROM t1 JOIN t2 ON
t1.c1 = t2.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"100000",
+
SQLConf.ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED.key ->
"true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ "an explicit MERGE hint must keep the join as a sort merge join")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: Do not convert when a project adds a large folded
constant") {
+ withTempView("t1", "t2") {
+ spark.sparkContext.parallelize(
+ (1 to 100).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ val query =
+ "SELECT t1.c1, x.wide FROM t1 JOIN " +
+ "(SELECT c2, coalesce(c2, repeat('x', 100)) AS wide FROM t2 GROUP BY
c2) x " +
+ "ON t1.c2 = x.c2"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "3",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"500",
+ lookThroughOperatorsKey -> "true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).isEmpty,
+ "a large folded constant above the shuffle must keep the join as a
sort merge join")
+ assert(findTopLevelSortMergeJoin(adaptive).size === 1)
+ }
+ }
+ }
+
+ test("SPARK-58084: Convert still fires when DemoteBroadcastHashJoin adds
NO_BROADCAST_HASH") {
+ withTempView("t1", "t2") {
+ // Both inputs have empty partitions. DemoteBroadcastHashJoin only
matches a direct
+ // LogicalQueryStage child, so it cannot demote the t1 side (behind the
aggregate) and
+ // instead tags the t2 side with a NO_BROADCAST_HASH hint. That hint
only forbids
+ // broadcasting and must not block converting the sort merge join to a
shuffled hash join.
+ // DemoteBroadcastHashJoin is left enabled (unlike the other conversion
tests) so the
+ // interaction is exercised.
+ spark.sparkContext.parallelize(
+ (1 to 2).map(i => TestData(i, i.toString)), 5)
+ .toDF("c1", "c2").createOrReplaceTempView("t1")
+ spark.sparkContext.parallelize(
+ (1 to 2).map(i => TestData(i, i.toString)), 10)
+ .toDF("c1", "c2").createOrReplaceTempView("t2")
+
+ // An aggregate sits between the join and t1's input shuffle, so the
logical
+ // DemoteBroadcastHashJoin rule cannot see a direct shuffle child and
only the physical rule
+ // can convert - which is exactly the look-through case this rule adds.
+ val query =
+ "SELECT x.c1, t2.c1 FROM " +
+ "(SELECT c1, count(*) AS cnt FROM t1 GROUP BY c1) x JOIN t2 ON x.c1
= t2.c1"
+
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "6",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NON_EMPTY_PARTITION_RATIO_FOR_BROADCAST_JOIN.key -> "1",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key ->
"100000",
+ lookThroughOperatorsKey -> "true") {
+ val (_, adaptive) = runAdaptiveAndVerifyResult(query)
+ assert(findTopLevelShuffledHashJoin(adaptive).size === 1,
+ "a NO_BROADCAST_HASH hint must not block the conversion")
+ assert(findTopLevelSortMergeJoin(adaptive).isEmpty)
+ }
+ }
+ }
+
test("SPARK-35650: Coalesce number of partitions by AEQ") {
withSQLConf(SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1") {
Seq("REPARTITION", "REBALANCE(key)")
@@ -3735,6 +4181,16 @@ class AdaptiveQueryExecSuite
}
}
+/**
+ * A minimal leaf plan with a single output attribute, used to build tiny
plans for cost tests.
+ */
+private case class CostTestLeafExec() extends LeafExecNode {
+ override protected def doExecute(): RDD[InternalRow] =
+ throw SparkException.internalError("should not be executed")
+ override def output: Seq[Attribute] =
+ AttributeReference("a", org.apache.spark.sql.types.IntegerType)() :: Nil
+}
+
/**
* Invalid implementation class for [[CostEvaluator]].
*/
@@ -3749,7 +4205,7 @@ private case class SimpleShuffleSortCostEvaluator()
extends CostEvaluator {
case s: ShuffleExchangeLike => s
case s: SortExec => s
}.size
- SimpleCost(cost)
+ SimpleCost(numSkewJoins = 0, numShuffles = cost, numLocalSorts = 0)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
