ulysses-you commented on code in PR #57181: URL: https://github.com/apache/spark/pull/57181#discussion_r3584108921
########## 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: Reinstated in a40f1d68a1f: the pattern now matches `SortMergeJoinExec(..., false)` explicitly, so a skew-marked join no longer falls into this case. Since a converted join is always non-skew, I also dropped the trailing `isSkewJoin` argument from the `ShuffledHashJoinExec` construction (its param defaults to `false`). This keeps the invariant explicit rather than implicit -- if some future rule ever routed a skew-marked SMJ here, it simply wouldn't match. ########## 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: Fixed the attribution in a40f1d68a1f. The comment now says both inputs have empty partitions and, since `DemoteBroadcastHashJoin` only matches a direct `LogicalQueryStage` child, it cannot demote the t1 side (behind the aggregate) and instead tags the t2 side with `NO_BROADCAST_HASH`. The interaction is unchanged and still exercised. -- 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]
