viirya commented on code in PR #57181:
URL: https://github.com/apache/spark/pull/57181#discussion_r3581103414
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -287,7 +288,7 @@ case object BuildRight extends BuildSide
case object BuildLeft extends BuildSide
-trait JoinSelectionHelper extends Logging {
+trait JoinSelectionHelper extends SQLConfHelper with Logging {
Review Comment:
This `SQLConfHelper` is now a leftover: it was added for
`preferShuffledHashJoin` when that lived here, but the method has moved into
the physical rule (which already gets `conf` from `Rule`), and every remaining
method in this trait takes `conf: SQLConf` explicitly. It can be reverted to
keep the trait's explicit-conf style.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/LiteralExpressionSuite.scala:
##########
@@ -855,4 +855,48 @@ 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).
+ assert(Literal(UTF8String.fromString("\u00e9"),
StringType).valueSizeInBytes === Some(2))
Review Comment:
This is why the lint job is still failing after the style fix: scalastyle
flags `nonascii` at 876:41 even though the source is now a pure-ASCII `\u00e9`
escape - the scalariform lexer decodes unicode escapes before
`NonASCIICharacterChecker` sees the text, so the escape counts as the character
itself. The escape can't dodge the check; build the string without one, e.g.:
```scala
// U+00E9 encodes to 2 bytes in UTF-8.
assert(Literal(UTF8String.fromString(new String(Character.toChars(0xE9))),
StringType)
.valueSizeInBytes === Some(2))
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ConvertSortMergeJoinToShuffledHashJoin.scala:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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, isSkewJoin)
Review Comment:
The previous revision matched `isSkewJoin = false` explicitly; this now
matches any value and propagates it. `true` is currently unreachable -
`OptimizeSkewedJoin` runs later in the same preparation-rule sequence, and the
skew-marked joins it produces have `AQEShuffleReadExec` children that
`findShuffleStage` doesn't match - but that's an implicit invariant. If some
rule ever breaks it, this would silently produce an SHJ claiming `isSkewJoin =
true` whose children lack the skew-split readers. I'd reinstate the explicit
`false` match (or a comment stating the invariant).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2461,6 +2466,412 @@ 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: 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") {
+ // t1 (the build candidate) has many empty partitions, so
DemoteBroadcastHashJoin demotes
Review Comment:
Minor: the comment attributes the `NO_BROADCAST_HASH` hint to t1, but
`DemoteBroadcastHashJoin` only matches a direct `LogicalQueryStage` child - the
t1 side is behind the aggregate, so the hint actually lands on the t2 side. The
interaction is still exercised (a `NO_BROADCAST_HASH` hint is present and
conversion fires), just worth fixing the attribution so the comment doesn't
mislead.
##########
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 {
Review Comment:
The rename needs a `docs/sql-migration-guide.md` entry: existing
`spark.sql.adaptive.optimizer.excludedRules` configs naming
`org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` now silently
stop matching (unknown names are ignored, not errored), and users who
previously disabled the shuffled-hash preference by excluding that rule must
now set
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled=false`
instead. Worth one entry covering both the rename and where the SHJ-over-SMJ
selection moved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
