You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "pratham76 (via GitHub)" <gi...@apache.org> on 2026/10/22 05:39:19 UTC
[PR] [SPARK-56906][SQL] Add SKEW_JOIN hint with SkewKeyJoinExec [spark]
pratham76 opened a new pull request, #59700:
URL: https://github.com/apache/spark/pull/59700
### What changes were proposed in this pull request?
We add a `SKEW_JOIN` join hint and a new physical operator `SkewKeyJoinExec` for joins where a key is frequent on both sides.
Syntax: `/*+ SKEW_JOIN(t(col1, col2)) */` (`SKEWJOIN` is an alias). The columns must be equi-join keys of the hinted relation.
* **Hint:** `SKEW_JOIN` is a new `JoinStrategyHint`, and `HintInfo` gets a `skewColumns` field. `ResolveJoinStrategyHints` reads `t(col)` (parsed as a function call, so no grammar change) and attaches the columns to the relation `t`.
* **Planning:** `JoinSelection` plans an equi-join with the hint as a `SkewKeyJoinExec`. If the join cannot be split, the hint is ignored with the usual hint warning and the join is planned as without the hint. Supported are inner joins, left outer joins hinted on the left side, and right outer joins hinted on the right side, with join keys of integral, decimal, boolean, date, timestamp or binary-collated string types that are the same on both sides.
* **Execution:** `SkewKeyJoinExec` samples the hinted side to find the hot keys and splits the join in two:
* the rows of the hot keys of both sides are joined with a `BroadcastHashJoinExec`, streaming the hinted side in its existing partitioning, and
* all the other rows are joined with a `SortMergeJoinExec`.
The results are unioned. Null keys are never hot and are handled by the sort merge join. If no key is hot, only the sort merge join runs. Since each side is read twice, a join with non-deterministic expressions is rejected with the new error `INVALID_SKEW_JOIN_HINT` (as are a column that is not in the relation and a malformed hint).
* **Hot keys:** a key is hot if its estimated number of rows (from the sample) is at least `spark.sql.skewJoinHint.minHotKeyRows` and at least the average rows of a shuffle partition (estimated rows of the hinted side divided by `spark.sql.shuffle.partitions`). At most `spark.sql.skewJoinHint.maxHotKeys` keys are used, the most frequent first.
* **Configs:** `spark.sql.skewJoinHint.sampleFraction` (0.1), `spark.sql.skewJoinHint.minHotKeyRows` (1000), `spark.sql.skewJoinHint.maxHotKeys` (1000).
### Why are the changes needed?
In a fact-to-fact join, a key that is frequent on both sides sends the Cartesian product of its rows to a single task, e.g. 5,000 x 10,000 = 50M output rows for one key. When that key is spread thinly over many mappers, all the shuffle blocks are small, so `OptimizeSkewedJoin`, which only looks at the sizes of the shuffle blocks, does not detect it. Splitting the partition and replicating the other side would not help either, as the problem is the output of a key and not the size of a block. Splitting the hot keys out of the join and broadcasting them spreads their output over all the tasks that hold their rows.
### Does this PR introduce _any_ user-facing change?
Yes. A new `SKEW_JOIN` join hint, the new error condition `INVALID_SKEW_JOIN_HINT`, and three new configs. Without the hint, nothing changes.
### How was this patch tested?
Added `SkewKeyJoinSuite`, which checks that the results are the same as without the hint, with AQE on and off, for inner joins hinted on either side, multiple keys, a subset of the join keys, non equi-join conditions, left and right outer joins, and null keys. It also covers no hot keys, the cap on the number of hot keys, sampling, the cases where the hint is ignored with a warning (unsupported join type, hinted side that cannot be split, column that is not a join key, different key types, double keys), and the error cases.
Also tested with existing UT
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56906][SQL] Add SKEW_JOIN hint with SkewKeyJoinExec [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #59700:
URL: https://github.com/apache/spark/pull/59700#discussion_r4361243912
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -512,6 +512,14 @@ trait JoinSelectionHelper extends Logging {
hint.rightHint.exists(_.strategy.contains(SHUFFLE_MERGE))
}
+ def hintToSkewJoinLeft(hint: JoinHint): Boolean = {
Review Comment:
**Nit (P3):** These helpers are only used in the equi-join branch. `SELECT /*+ SKEW_JOIN(t1(k1)) */ * FROM t1 JOIN t2 ON t1.k1 < t2.k1` takes the non-equi branch, where `checkHintNonEquiJoin` only reports SHUFFLE_HASH and MERGE. The hint then disappears without the 'hint not supported' warning that every other unsplittable case logs. Adding SKEW_JOIN to `checkHintNonEquiJoin` with the 'no equi-join keys' reason, plus a test, would close the gap.
**Verification:**
- **Regression:** A SKEW_JOIN-hinted join on a non-equality condition returns the unhinted result and logs a warning naming skew_join and the missing equi-join keys.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -256,6 +256,66 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case j @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, nonEquiCond,
_, left, right, hint) =>
val hashJoinSupport = hashJoinSupported(leftKeys, rightKeys)
+
+ // Splits the join on the hot keys of the hinted side, see `SkewKeyJoinExec`. The hint is
+ // ignored if the join cannot be split, in which case the join is planned without it.
+ def createSkewKeyJoin(): Option[Seq[SparkPlan]] = {
+ val skewLeft = hintToSkewJoinLeft(hint) &&
+ joins.SkewKeyJoinExec.supportsJoinType(joinType, skewLeft = true)
+ val skewRight = !skewLeft && hintToSkewJoinRight(hint) &&
+ joins.SkewKeyJoinExec.supportsJoinType(joinType, skewLeft = false)
+ val hintInfo = if (skewLeft || (!skewRight && hintToSkewJoinLeft(hint))) {
+ hint.leftHint
+ } else {
+ hint.rightHint
+ }
+ def notSupported(reason: String): Option[Seq[SparkPlan]] = {
+ hintInfo.foreach(hintErrorHandler.joinHintNotSupported(_, reason))
+ None
+ }
+
+ if (!hintToSkewJoinLeft(hint) && !hintToSkewJoinRight(hint)) {
+ None
+ } else if (!skewLeft && !skewRight) {
+ notSupported(s"${joinType.sql.toLowerCase(Locale.ROOT)} join with the hinted " +
+ "side as the side to split")
+ } else if (!left.deterministic || !right.deterministic) {
+ throw QueryCompilationErrors.invalidSkewJoinHintError(
+ "the join has non-deterministic expressions, but both sides of the join are " +
+ "read twice")
+ } else if (!RowOrdering.isOrderable(leftKeys)) {
+ notSupported("the join keys are not sortable")
+ } else {
+ val skewChild = if (skewLeft) left else right
+ val skewKeys = if (skewLeft) leftKeys else rightKeys
+ val otherKeys = if (skewLeft) rightKeys else leftKeys
+ val skewColumns = hintInfo.get.skewColumns.map { column =>
+ skewChild.output.find(a => conf.resolver(a.name, column)).getOrElse {
Review Comment:
**Nit (P3):** The names are looked up in `skewChild.output`, which is the optimized join child, after `ColumnPruning` and after `EliminateResolvedHint` has lifted the hint through any unary nodes, rather than in the relation named in the hint. With `t1(k1, k2, v)`:
- `SELECT /*+ SKEW_JOIN(t1(v)) */ * FROM t1 JOIN t2 ON t1.k1 = t2.k1` keeps `v`, so the hint is ignored with the 'not all equi-join keys' warning (the suite's case);
- `SELECT /*+ SKEW_JOIN(t1(v)) */ t1.k1, t2.w FROM t1 JOIN t2 ON t1.k1 = t2.k1` prunes `v`, so planning throws `INVALID_SKEW_JOIN_HINT: the column v is not found in the relation`, although `t1` has `v`.
A real typo is also reported only at physical planning (EXPLAIN or an action), not at analysis. And the comment above `createSkewKeyJoin` says the hint is ignored whenever the join cannot be split, which neither throw path matches.
**Recommended change:** Validate SKEW_JOIN column names against the hinted relation once it is resolved, for example with a check over ResolvedHint nodes that carry SKEW_JOIN during analysis checking. Raise INVALID_SKEW_JOIN_HINT only for names the relation does not have. In the planner, treat a name missing from the optimized join child as not an equi-join key and ignore the hint with the existing warning. Update the planner comment to say which cases fail the query and which only ignore the hint. Add analysis and planning tests for both outcomes.
**Why this works:** The relation the user named decides existence before pruning or lifting can change what is visible, so an error appears at analysis and only for real typos. The planner then decides only join-key membership, where a name absent from the child always means the column cannot be a key.
**Scope:** Move the column-existence check for SKEW_JOIN to analysis and make the planner treat missing names as not join keys.
**Risks:** The check must run after the hinted relation is resolved and must not reject hints on views or subquery aliases whose output names the column with different case under case-insensitive resolution.
**Constraints:** A column the relation does not have still fails with INVALID_SKEW_JOIN_HINT, as the change declares.
**Exclusions:**
- Accepting SKEW_JOIN without columns or with DataFrame string parameters, which depends on the open question about the supported hint forms.
**Success:** A hinted column that exists in the relation but is not an equi-join key is ignored with the warning, whatever the query selects. A hinted column the relation does not have fails at analysis, before EXPLAIN or execution.
**Verification:**
- **Regression:** SKEW_JOIN on a non-key column of the hinted relation is ignored with the not-all-equi-join-keys warning both when the query selects that column and when column pruning removes it.
- **Behavior:** SKEW_JOIN on a column the hinted relation lacks fails during analysis with INVALID_SKEW_JOIN_HINT.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -256,6 +256,66 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
case j @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, nonEquiCond,
_, left, right, hint) =>
val hashJoinSupport = hashJoinSupported(leftKeys, rightKeys)
+
+ // Splits the join on the hot keys of the hinted side, see `SkewKeyJoinExec`. The hint is
+ // ignored if the join cannot be split, in which case the join is planned without it.
+ def createSkewKeyJoin(): Option[Seq[SparkPlan]] = {
+ val skewLeft = hintToSkewJoinLeft(hint) &&
+ joins.SkewKeyJoinExec.supportsJoinType(joinType, skewLeft = true)
+ val skewRight = !skewLeft && hintToSkewJoinRight(hint) &&
+ joins.SkewKeyJoinExec.supportsJoinType(joinType, skewLeft = false)
+ val hintInfo = if (skewLeft || (!skewRight && hintToSkewJoinLeft(hint))) {
+ hint.leftHint
+ } else {
+ hint.rightHint
+ }
+ def notSupported(reason: String): Option[Seq[SparkPlan]] = {
+ hintInfo.foreach(hintErrorHandler.joinHintNotSupported(_, reason))
+ None
+ }
+
+ if (!hintToSkewJoinLeft(hint) && !hintToSkewJoinRight(hint)) {
+ None
+ } else if (!skewLeft && !skewRight) {
+ notSupported(s"${joinType.sql.toLowerCase(Locale.ROOT)} join with the hinted " +
+ "side as the side to split")
+ } else if (!left.deterministic || !right.deterministic) {
+ throw QueryCompilationErrors.invalidSkewJoinHintError(
+ "the join has non-deterministic expressions, but both sides of the join are " +
+ "read twice")
+ } else if (!RowOrdering.isOrderable(leftKeys)) {
+ notSupported("the join keys are not sortable")
+ } else {
+ val skewChild = if (skewLeft) left else right
+ val skewKeys = if (skewLeft) leftKeys else rightKeys
+ val otherKeys = if (skewLeft) rightKeys else leftKeys
+ val skewColumns = hintInfo.get.skewColumns.map { column =>
+ skewChild.output.find(a => conf.resolver(a.name, column)).getOrElse {
+ throw QueryCompilationErrors.invalidSkewJoinHintError(
+ s"the column $column is not found in the relation")
+ }
+ }
+ val skewKeyIndices = skewColumns.map { column =>
+ skewKeys.indexWhere {
+ case a: Attribute => a.semanticEquals(column)
+ case _ => false
+ }
+ }
+ if (skewKeyIndices.contains(-1)) {
+ notSupported("the columns are not all equi-join keys")
+ } else if (!skewKeyIndices.forall { i =>
+ joins.SkewKeyJoinExec.isSupportedKeyType(skewKeys(i).dataType) &&
+ otherKeys(i).isInstanceOf[Attribute] &&
+ otherKeys(i).dataType == skewKeys(i).dataType}) {
+ notSupported("the types of the join keys are not supported or differ between " +
+ "the two sides")
+ } else {
+ Some(Seq(joins.SkewKeyJoinExec(leftKeys, rightKeys, skewKeyIndices, joinType,
Review Comment:
**Non-blocking (P2):** This path checks the join type, determinism, orderability and the hinted columns' types, but not `hashJoinSupport`, which is computed a few lines up. Once sampling finds a hot key, `SkewKeyJoinExec.doExecute` builds `BroadcastHashJoinExec(leftKeys, rightKeys, ...)` over all equi-join keys, and the `HashJoin` trait asserts that every key is binary-stable (HashJoin.scala:45-46). `RewriteCollationJoin` only wraps `Equality(AttributeReference, AttributeReference)`, so a collated key on an expression reaches this point unwrapped, e.g.
```sql
SELECT /*+ SKEW_JOIN(t1(id)) */ * FROM table1 t1 JOIN table3 t3
ON t1.id = t3.id AND t3.col1 = t1.col1 COLLATE UTF8_LCASE_RTRIM
```
The hinted key `id` passes every check, and as soon as it has a hot key the query fails with `java.lang.AssertionError`. Without the hint it runs as a sort merge join. Could the skew path fall back through `notSupported` when `hashJoinSupport` is false, as the broadcast and shuffled hash paths do?
**Verification:**
- **Regression:** A hinted join with a hot key on the hinted column whose other equi-join key is a non-binary-collated string expression, which the collation join rewrite leaves unwrapped, returns the same rows as the unhinted query instead of failing, and logs the hint-not-supported warning.
- **Behavior:** Hinted joins whose keys all support hash joins are still split and find their hot keys.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinExec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.joins
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.exchange.EnsureRequirements
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Joins two sides on equi-join keys, where some keys are so frequent on both sides that all their
+ * output rows would end up in a single task of a shuffled join. The shuffle blocks of such a key
+ * can all be small, as the key is spread thinly over many mappers, so adaptive skew join handling,
+ * which looks at the sizes of the shuffle blocks, does not help.
+ *
+ * This operator samples the skewed side to find the hot keys, and then splits the join in two:
+ * - the rows of the hot keys of both sides are joined with a [[BroadcastHashJoinExec]], which
+ * streams the skewed side in its existing partitioning, so that the output of a hot key is
+ * produced by all the tasks that hold its rows, and
+ * - all the other rows are joined with a [[SortMergeJoinExec]].
+ * The results are unioned. As every row is joined in exactly one of the two branches, this gives
+ * the same result as a single join. The rows with a null key are never hot, as they do not match
+ * any row, and so are only handled by the sort merge join.
+ *
+ * Each side is read twice, once per branch, so both sides must be deterministic. The rows of the
Review Comment:
**Nit (P3):** `findHotKeys` runs a separate job over `skewChild.execute()` before either branch runs. With hot keys the hinted side is therefore computed three times (sampling, broadcast-join stream, sort-merge shuffle), and twice when no key is hot; the other side is computed twice. The non-determinism error reason in `SparkStrategies` ('both sides of the join are read twice') and the PR description have the same undercount.
**Verification:**
- **Inspection:** The operator doc states that the hinted side is computed for sampling plus each branch and the other side once per branch.
- **Behavior:** The non-determinism rejection still raises INVALID_SKEW_JOIN_HINT, with the corrected reason.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinExec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.joins
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.exchange.EnsureRequirements
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Joins two sides on equi-join keys, where some keys are so frequent on both sides that all their
+ * output rows would end up in a single task of a shuffled join. The shuffle blocks of such a key
+ * can all be small, as the key is spread thinly over many mappers, so adaptive skew join handling,
+ * which looks at the sizes of the shuffle blocks, does not help.
+ *
+ * This operator samples the skewed side to find the hot keys, and then splits the join in two:
+ * - the rows of the hot keys of both sides are joined with a [[BroadcastHashJoinExec]], which
+ * streams the skewed side in its existing partitioning, so that the output of a hot key is
+ * produced by all the tasks that hold its rows, and
+ * - all the other rows are joined with a [[SortMergeJoinExec]].
+ * The results are unioned. As every row is joined in exactly one of the two branches, this gives
+ * the same result as a single join. The rows with a null key are never hot, as they do not match
+ * any row, and so are only handled by the sort merge join.
+ *
+ * Each side is read twice, once per branch, so both sides must be deterministic. The rows of the
+ * hot keys of the side that is not skewed are broadcast, so this is expected to be a small number.
+ * If no key is hot, the operator only runs the sort merge join.
+ *
+ * @param leftKeys the left join keys.
+ * @param rightKeys the right join keys.
+ * @param skewKeyIndices the indices of the join keys to find the hot keys on. These keys of the
+ * skewed side must be attributes of the supported key types, and the
+ * corresponding keys of the other side must have the same types.
+ * @param joinType the type of the join: inner, or an outer join whose preserved side is the
+ * skewed side, as the other side is broadcast.
+ * @param condition the non equi-join condition.
+ * @param skewLeft whether the left side is the skewed side, otherwise the right side is.
+ */
+case class SkewKeyJoinExec(
+ leftKeys: Seq[Expression],
+ rightKeys: Seq[Expression],
+ skewKeyIndices: Seq[Int],
+ joinType: JoinType,
+ condition: Option[Expression],
+ skewLeft: Boolean,
+ left: SparkPlan,
+ right: SparkPlan) extends BinaryExecNode {
+
+ override lazy val metrics = Map(
+ "numHotKeys" -> SQLMetrics.createMetric(sparkContext, "number of hot keys"))
+
+ // The output of the join is the same for all the join strategies.
+ override lazy val output: Seq[Attribute] = sortMergeJoin(left, right).output
+
+ private def skewChild: SparkPlan = if (skewLeft) left else right
+
+ private def skewSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) leftKeys else rightKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def otherSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) rightKeys else leftKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def sortMergeJoin(l: SparkPlan, r: SparkPlan): SparkPlan = {
+ SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, l, r)
+ }
+
+ private def hotKeyFilter(keys: Seq[Expression], hotKeys: Set[Seq[Any]], hot: Boolean)
+ : SparkPlan => SparkPlan = {
+ FilterExec(SkewHotKeyFilter(keys, hotKeys, hot), _)
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val hotKeys = findHotKeys()
+ longMetric("numHotKeys").set(hotKeys.size)
+ val plan = if (hotKeys.isEmpty) {
+ sortMergeJoin(left, right)
+ } else {
+ val (hotSkew, coldSkew) = (hotKeyFilter(skewSideKeys, hotKeys, hot = true),
+ hotKeyFilter(skewSideKeys, hotKeys, hot = false))
+ val (hotOther, coldOther) = (hotKeyFilter(otherSideKeys, hotKeys, hot = true),
+ hotKeyFilter(otherSideKeys, hotKeys, hot = false))
+ val (hotLeft, hotRight, coldLeft, coldRight) = if (skewLeft) {
+ (hotSkew(left), hotOther(right), coldSkew(left), coldOther(right))
+ } else {
+ (hotOther(left), hotSkew(right), coldOther(left), coldSkew(right))
+ }
+ // The skewed side is streamed and the other side is broadcast.
+ val buildSide = if (skewLeft) BuildRight else BuildLeft
+ UnionExec(Seq(
+ BroadcastHashJoinExec(
+ leftKeys, rightKeys, joinType, buildSide, condition, hotLeft, hotRight),
+ sortMergeJoin(coldLeft, coldRight)))
+ }
+ // The shuffles and the broadcast of the two branches are planned here, as which keys are hot
+ // is only known once the skewed side has been sampled.
+ EnsureRequirements().apply(plan).execute()
Review Comment:
**Non-blocking (P2):** Everything the split actually executes is built here at execution time: `FilterExec(SkewHotKeyFilter)`, the `BroadcastHashJoinExec`, the `SortMergeJoinExec`, the `UnionExec`, and the exchanges that `EnsureRequirements()` adds. None of it is part of the planned tree, so for every hinted join that finds hot keys:
- AQE never turns the cold-branch shuffles or the broadcast into query stages, so there is no partition coalescing, no `OptimizeSkewedJoin` on the cold branch and no runtime statistics;
- `CollapseCodegenStages` and exchange reuse never see the inner operators;
- EXPLAIN and the SQL UI show only `SkewKeyJoinExec`;
- query-scoped shuffle cleanup misses these shuffles, and the sampling `reduceByKey` shuffle too, because `SQLExecution.extractShuffleIds` only looks at the AQE context or at `ShuffleExchangeLike` nodes in the executed plan;
- because `SkewKeyJoinExec` declares no `requiredChildDistribution`, `DisableUnnecessaryBucketedScan` turns off bucketed scans below it, and the inner sort merge join reshuffles them.
The code comment says this is needed because the hot keys are only known after sampling. A data-dependent value does not force execution-time planning, though: `InjectRuntimeFilter` plans its bloom filter as a predicate over a `ScalarSubquery`. Even `UnionLoopExec`, which does build plans at runtime, sends them through `Dataset.ofRows(...).queryExecution` rather than hand-built physical operators.
**Recommended change:** Plan the skew split instead of building it at execution time. Compute the hot-key set with a planned subquery over a sample of the hinted side: an aggregate that counts rows per key and in total and keeps the qualifying keys. Then plan the join as a union of a broadcast hash join over rows whose keys are in that set and a sort merge join over the remaining rows, with both sides filtered by a membership predicate that reads the subquery result. Add join and AQE tests that assert the broadcast, the cold-branch shuffles and the subquery appear in the executed plan.
**Why this works:** As with the runtime bloom filter, the only data-dependent value becomes a subquery result. Every operator and exchange of the split is then part of the planned tree, where EnsureRequirements, AQE stages with coalescing and skew handling, whole-stage codegen, exchange reuse, EXPLAIN, the SQL UI and query-scoped shuffle cleanup already apply without manual relays.
**Scope:** Move hot-key discovery and the hot/cold split from SkewKeyJoinExec.doExecute into planning, using a hot-key subquery and a planned union of the two joins.
**Compatibility:** Every row is joined in exactly one branch, null keys only in the sort merge join branch, and outer-join semantics on the hinted side are unchanged.
**Risks:** Duplicating each join side in the planned union must keep the original output attribute IDs on the first branch and fresh IDs on the second, or references above the join break. The planned hot-key subquery must stay comparable in cost to the current sampling pass.
**Constraints:** Unhinted queries and joins whose SKEW_JOIN hint is ignored must plan exactly as they do now. Each row must still be joined in exactly one branch, with null keys only in the sort merge join branch.
**Exclusions:**
- Changing which keys count as hot or bounding the broadcast of the other side, both of which depend on the open questions about how hot keys are discovered and how large that broadcast may be.
**Success:** The executed plan of a hinted join with hot keys contains the broadcast exchange, both cold-branch shuffle exchanges and the hot-key subquery as planned nodes. With AQE on, the cold-branch shuffles are query stages subject to partition coalescing and skew handling. Query-scoped shuffle cleanup sees every shuffle the hinted join creates.
**Verification:**
- **Behavior:** For a hinted join with hot keys, with AQE on and off, the executed plan exposes the broadcast exchange, both cold-branch shuffle exchanges and the hot-key subquery, and the result matches the unhinted query.
- **Regression:** With AQE partition coalescing enabled, the cold-branch shuffles are coalesced like any other query shuffle. With shuffle-dependency file cleanup enabled, no shuffle created by the hinted join survives the end of the query.
- **Compatibility:** Unhinted joins and joins whose SKEW_JOIN hint is ignored produce the same physical plans as before.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinExec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.joins
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.exchange.EnsureRequirements
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Joins two sides on equi-join keys, where some keys are so frequent on both sides that all their
+ * output rows would end up in a single task of a shuffled join. The shuffle blocks of such a key
+ * can all be small, as the key is spread thinly over many mappers, so adaptive skew join handling,
+ * which looks at the sizes of the shuffle blocks, does not help.
+ *
+ * This operator samples the skewed side to find the hot keys, and then splits the join in two:
+ * - the rows of the hot keys of both sides are joined with a [[BroadcastHashJoinExec]], which
+ * streams the skewed side in its existing partitioning, so that the output of a hot key is
+ * produced by all the tasks that hold its rows, and
+ * - all the other rows are joined with a [[SortMergeJoinExec]].
+ * The results are unioned. As every row is joined in exactly one of the two branches, this gives
+ * the same result as a single join. The rows with a null key are never hot, as they do not match
+ * any row, and so are only handled by the sort merge join.
+ *
+ * Each side is read twice, once per branch, so both sides must be deterministic. The rows of the
+ * hot keys of the side that is not skewed are broadcast, so this is expected to be a small number.
+ * If no key is hot, the operator only runs the sort merge join.
+ *
+ * @param leftKeys the left join keys.
+ * @param rightKeys the right join keys.
+ * @param skewKeyIndices the indices of the join keys to find the hot keys on. These keys of the
+ * skewed side must be attributes of the supported key types, and the
+ * corresponding keys of the other side must have the same types.
+ * @param joinType the type of the join: inner, or an outer join whose preserved side is the
+ * skewed side, as the other side is broadcast.
+ * @param condition the non equi-join condition.
+ * @param skewLeft whether the left side is the skewed side, otherwise the right side is.
+ */
+case class SkewKeyJoinExec(
+ leftKeys: Seq[Expression],
+ rightKeys: Seq[Expression],
+ skewKeyIndices: Seq[Int],
+ joinType: JoinType,
+ condition: Option[Expression],
+ skewLeft: Boolean,
+ left: SparkPlan,
+ right: SparkPlan) extends BinaryExecNode {
+
+ override lazy val metrics = Map(
+ "numHotKeys" -> SQLMetrics.createMetric(sparkContext, "number of hot keys"))
+
+ // The output of the join is the same for all the join strategies.
+ override lazy val output: Seq[Attribute] = sortMergeJoin(left, right).output
+
+ private def skewChild: SparkPlan = if (skewLeft) left else right
+
+ private def skewSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) leftKeys else rightKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def otherSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) rightKeys else leftKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def sortMergeJoin(l: SparkPlan, r: SparkPlan): SparkPlan = {
+ SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, l, r)
+ }
+
+ private def hotKeyFilter(keys: Seq[Expression], hotKeys: Set[Seq[Any]], hot: Boolean)
+ : SparkPlan => SparkPlan = {
+ FilterExec(SkewHotKeyFilter(keys, hotKeys, hot), _)
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val hotKeys = findHotKeys()
+ longMetric("numHotKeys").set(hotKeys.size)
Review Comment:
**Nit (P3):** Driver-side `SQLMetric` updates have to be posted with `SQLMetrics.postDriverMetricUpdates(sparkContext, executionId, ...)`; see the `SQLMetric` doc, and `BroadcastExchangeExec` or `UnionLoopExec` for examples. Without that call the SQL UI and history server never receive the value. It is the only signal of what the hint did, because the inner plan is otherwise invisible. The suite reads the metric from the plan object, so it cannot catch this.
**Verification:**
- **Behavior:** The number of hot keys for a hinted join is visible in the metrics recorded for its SQL execution, not only on the in-memory plan object.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinExec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.joins
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.exchange.EnsureRequirements
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Joins two sides on equi-join keys, where some keys are so frequent on both sides that all their
+ * output rows would end up in a single task of a shuffled join. The shuffle blocks of such a key
+ * can all be small, as the key is spread thinly over many mappers, so adaptive skew join handling,
+ * which looks at the sizes of the shuffle blocks, does not help.
+ *
+ * This operator samples the skewed side to find the hot keys, and then splits the join in two:
+ * - the rows of the hot keys of both sides are joined with a [[BroadcastHashJoinExec]], which
+ * streams the skewed side in its existing partitioning, so that the output of a hot key is
+ * produced by all the tasks that hold its rows, and
+ * - all the other rows are joined with a [[SortMergeJoinExec]].
+ * The results are unioned. As every row is joined in exactly one of the two branches, this gives
+ * the same result as a single join. The rows with a null key are never hot, as they do not match
+ * any row, and so are only handled by the sort merge join.
+ *
+ * Each side is read twice, once per branch, so both sides must be deterministic. The rows of the
+ * hot keys of the side that is not skewed are broadcast, so this is expected to be a small number.
+ * If no key is hot, the operator only runs the sort merge join.
+ *
+ * @param leftKeys the left join keys.
+ * @param rightKeys the right join keys.
+ * @param skewKeyIndices the indices of the join keys to find the hot keys on. These keys of the
+ * skewed side must be attributes of the supported key types, and the
+ * corresponding keys of the other side must have the same types.
+ * @param joinType the type of the join: inner, or an outer join whose preserved side is the
+ * skewed side, as the other side is broadcast.
+ * @param condition the non equi-join condition.
+ * @param skewLeft whether the left side is the skewed side, otherwise the right side is.
+ */
+case class SkewKeyJoinExec(
+ leftKeys: Seq[Expression],
+ rightKeys: Seq[Expression],
+ skewKeyIndices: Seq[Int],
+ joinType: JoinType,
+ condition: Option[Expression],
+ skewLeft: Boolean,
+ left: SparkPlan,
+ right: SparkPlan) extends BinaryExecNode {
+
+ override lazy val metrics = Map(
+ "numHotKeys" -> SQLMetrics.createMetric(sparkContext, "number of hot keys"))
+
+ // The output of the join is the same for all the join strategies.
+ override lazy val output: Seq[Attribute] = sortMergeJoin(left, right).output
+
+ private def skewChild: SparkPlan = if (skewLeft) left else right
+
+ private def skewSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) leftKeys else rightKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def otherSideKeys: Seq[Expression] = {
+ val keys = if (skewLeft) rightKeys else leftKeys
+ skewKeyIndices.map(keys)
+ }
+
+ private def sortMergeJoin(l: SparkPlan, r: SparkPlan): SparkPlan = {
+ SortMergeJoinExec(leftKeys, rightKeys, joinType, condition, l, r)
+ }
+
+ private def hotKeyFilter(keys: Seq[Expression], hotKeys: Set[Seq[Any]], hot: Boolean)
+ : SparkPlan => SparkPlan = {
+ FilterExec(SkewHotKeyFilter(keys, hotKeys, hot), _)
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val hotKeys = findHotKeys()
+ longMetric("numHotKeys").set(hotKeys.size)
+ val plan = if (hotKeys.isEmpty) {
+ sortMergeJoin(left, right)
+ } else {
+ val (hotSkew, coldSkew) = (hotKeyFilter(skewSideKeys, hotKeys, hot = true),
+ hotKeyFilter(skewSideKeys, hotKeys, hot = false))
+ val (hotOther, coldOther) = (hotKeyFilter(otherSideKeys, hotKeys, hot = true),
+ hotKeyFilter(otherSideKeys, hotKeys, hot = false))
+ val (hotLeft, hotRight, coldLeft, coldRight) = if (skewLeft) {
+ (hotSkew(left), hotOther(right), coldSkew(left), coldOther(right))
+ } else {
+ (hotOther(left), hotSkew(right), coldOther(left), coldSkew(right))
+ }
+ // The skewed side is streamed and the other side is broadcast.
+ val buildSide = if (skewLeft) BuildRight else BuildLeft
+ UnionExec(Seq(
+ BroadcastHashJoinExec(
+ leftKeys, rightKeys, joinType, buildSide, condition, hotLeft, hotRight),
+ sortMergeJoin(coldLeft, coldRight)))
+ }
+ // The shuffles and the broadcast of the two branches are planned here, as which keys are hot
+ // is only known once the skewed side has been sampled.
+ EnsureRequirements().apply(plan).execute()
+ }
+
+ /**
+ * Samples the skewed side to find the keys that are hot. A key is hot if its estimated number
+ * of rows is at least `spark.sql.skewJoinHint.minHotKeyRows` and at least the average number of
+ * rows of a shuffle partition, so that all the rows of the key alone fill a task. At most
+ * `spark.sql.skewJoinHint.maxHotKeys` of the most frequent keys are returned.
+ */
+ private def findHotKeys(): Set[Seq[Any]] = {
+ val fraction = conf.getConf(SQLConf.SKEW_JOIN_HINT_SAMPLE_FRACTION)
+ val minHotKeyRows = conf.getConf(SQLConf.SKEW_JOIN_HINT_MIN_HOT_KEY_ROWS)
+ val maxHotKeys = conf.getConf(SQLConf.SKEW_JOIN_HINT_MAX_HOT_KEYS)
+ val numPartitions = conf.numShufflePartitions
+
+ val keys = skewSideKeys
+ val keyTypes = keys.map(_.dataType)
+ val childOutput = skewChild.output
+ val sampledRows = sparkContext.longAccumulator("sampled rows of the skewed side")
+
+ val keyCounts = skewChild.execute()
+ .sample(withReplacement = false, fraction, SkewKeyJoinExec.SAMPLE_SEED)
+ .mapPartitionsInternal { iter =>
+ val project = UnsafeProjection.create(keys, childOutput)
+ iter.flatMap { row =>
+ sampledRows.add(1L)
Review Comment:
**Nit (P3):** Accumulator updates made inside transformations may be applied more than once when tasks or stages are re-executed (see 'Accumulators' in the RDD programming guide). If a fetch failure in the `top()` stage resubmits the sampling map stage, the recomputed tasks add their rows to `sampledRows` again. The per-key counts stay correct, because the reducer reads each map output once, but `estimatedRows` grows, and with it `max(minHotKeyRows, estimatedRows / numPartitions)`, so real hot keys can drop out. Deriving the total from the job's result, for example by carrying per-partition counts through the aggregation, avoids this.
**Verification:**
- **Inspection:** No accumulator updated inside a transformation feeds the hot-key threshold; the sampled total comes from the job's returned result.
- **Behavior:** Hot-key detection on the existing fixtures finds the same keys as before.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinSuite.scala:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.joins
+
+import org.apache.logging.log4j.Level
+
+import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest}
+import org.apache.spark.sql.catalyst.plans.logical.{HintInfo, Join, JoinHint, SKEW_JOIN}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+class SkewKeyJoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper {
+
+ // The key 7 is hot on both sides, the key 9 is less frequent, and every tenth row has a null key.
+ // The second key has three values.
+ private def createTables(): Unit = {
+ sql(
+ """SELECT CASE WHEN id % 10 = 0 THEN NULL WHEN id % 2 = 0 THEN 7 WHEN id % 10 = 5 THEN 9
+ | ELSE id END AS k1,
+ | id % 3 AS k2, id AS v
+ |FROM range(1000)""".stripMargin).createOrReplaceTempView("t1")
+ sql(
+ """SELECT CASE WHEN id % 10 = 0 THEN NULL WHEN id % 2 = 0 THEN 7 WHEN id % 10 = 5 THEN 9
+ | ELSE id END AS k1,
+ | id % 3 AS k2, id AS w
+ |FROM range(300)""".stripMargin).createOrReplaceTempView("t2")
+ sql("SELECT CAST(k1 AS INT) AS i, w FROM t2").createOrReplaceTempView("t3")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTables()
+ }
+
+ private def withSkewJoinConfs(f: => Unit): Unit = {
+ withSQLConf(
+ SQLConf.SHUFFLE_PARTITIONS.key -> "4",
+ SQLConf.SKEW_JOIN_HINT_SAMPLE_FRACTION.key -> "1.0",
+ SQLConf.SKEW_JOIN_HINT_MIN_HOT_KEY_ROWS.key -> "10",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1")(f)
+ }
+
+ private def skewKeyJoins(df: DataFrame): Seq[SkewKeyJoinExec] = {
+ collect(df.queryExecution.executedPlan) { case j: SkewKeyJoinExec => j }
+ }
+
+ /**
+ * Checks that the query with the hint is planned as a [[SkewKeyJoinExec]] that finds hot keys,
+ * and returns the same rows as the query without the hint.
+ */
+ private def checkSkewKeyJoin(hint: String, joinClause: String, hasHotKeys: Boolean = true)
+ : Unit = {
+ val hinted = sql(s"SELECT /*+ $hint */ * FROM $joinClause")
+ val expected = sql(s"SELECT * FROM $joinClause")
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ checkAnswer(hinted, expected)
+ val joins = skewKeyJoins(hinted)
+ assert(joins.size == 1)
+ assert((joins.head.metrics("numHotKeys").value > 0) == hasHotKeys)
+ }
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
+ checkAnswer(sql(s"SELECT /*+ $hint */ * FROM $joinClause"), expected)
+ }
+ }
+
+ test("SKEW_JOIN hint is resolved on the hinted relation") {
+ val df = sql("SELECT /*+ SKEW_JOIN(t1(k1, k2)) */ * FROM t1 JOIN t2 ON t1.k1 = t2.k1")
+ val hints = df.queryExecution.optimizedPlan.collect { case j: Join => j.hint }
+ assert(hints == Seq(
+ JoinHint(Some(HintInfo(Some(SKEW_JOIN), Seq("k1", "k2"))), None)))
+
+ val df2 = sql("SELECT /*+ SKEWJOIN(t2(k1)) */ * FROM t1 JOIN t2 ON t1.k1 = t2.k1")
+ val hints2 = df2.queryExecution.optimizedPlan.collect { case j: Join => j.hint }
+ assert(hints2 == Seq(JoinHint(None, Some(HintInfo(Some(SKEW_JOIN), Seq("k1"))))))
+ }
+
+ test("inner join with a hint on the left side") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1))", "t1 JOIN t2 ON t1.k1 = t2.k1")
+ }
+ }
+
+ test("inner join with a hint on the right side") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t2(k1))", "t1 JOIN t2 ON t1.k1 = t2.k1")
+ }
+ }
+
+ test("inner join on multiple keys") {
+ withSkewJoinConfs {
+ // The rows of the key 7 are split over the values of the second key, so use more
+ // partitions for the pairs to be hot.
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "100") {
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1, k2))",
+ "t1 JOIN t2 ON t1.k1 = t2.k1 AND t1.k2 = t2.k2")
+ }
+ }
+ }
+
+ test("hint on a subset of the join keys") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1))", "t1 JOIN t2 ON t1.k1 = t2.k1 AND t1.k2 = t2.k2")
+ }
+ }
+
+ test("inner join with a non equi-join condition") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1))", "t1 JOIN t2 ON t1.k1 = t2.k1 AND t1.v > t2.w")
+ }
+ }
+
+ test("left outer join with a hint on the left side") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1))", "t1 LEFT JOIN t2 ON t1.k1 = t2.k1")
+ checkSkewKeyJoin("SKEW_JOIN(t1(k1))",
+ "t1 LEFT JOIN t2 ON t1.k1 = t2.k1 AND t1.v > t2.w")
+ }
+ }
+
+ test("right outer join with a hint on the right side") {
+ withSkewJoinConfs {
+ checkSkewKeyJoin("SKEW_JOIN(t2(k1))", "t1 RIGHT JOIN t2 ON t1.k1 = t2.k1")
+ checkSkewKeyJoin("SKEW_JOIN(t2(k1))",
+ "t1 RIGHT JOIN t2 ON t1.k1 = t2.k1 AND t1.v > t2.w")
+ }
+ }
+
+ test("null keys are not hot and keep their outer join semantics") {
+ withSkewJoinConfs {
+ val query = "SELECT /*+ SKEW_JOIN(t1(k1)) */ t1.v, t2.w FROM t1 LEFT JOIN t2 " +
+ "ON t1.k1 = t2.k1 WHERE t1.k1 IS NULL"
Review Comment:
**Nit (P3):** A few of the behaviors the suite names cannot actually fail:
1. This test filters `WHERE t1.k1 IS NULL` on the preserved side, and that filter is pushed below the join. The operator only sees null keys, finds nothing hot and runs the plain sort merge join, and nothing asserts the operator or `numHotKeys`. The cap test would not notice either: with 100 partitions, null (100 rows) still ranks below 7 (401) and 9 (101).
2. The cap test only checks the count, so keeping the least frequent key passes.
3. With fraction 0.5, `minHotKeyRows` 10 and 4 partitions, key 7 stays hot with or without the 1/fraction scaling.
4. No test merges SKEW_JOIN with another hint on the same relation. If `HintInfo.merge` dropped `skewColumns`, every row would be hot and the whole other side broadcast, and the suite would still pass.
5. The both-sides test uses the same column name on both relations and only an inner join, so picking the wrong side's columns for a right outer join passes.
**Verification:**
- **Regression:** A test fails if null keys are counted as hot, if the cap keeps a less frequent key, if sampled counts are not scaled by the sampling fraction, if merging SKEW_JOIN with another hint on the same relation drops its columns, or if a right outer join hinted on both sides uses the left side's columns.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1579,6 +1579,38 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("256MB")
+ val SKEW_JOIN_HINT_SAMPLE_FRACTION =
+ buildConf("spark.sql.skewJoinHint.sampleFraction")
+ .doc("The fraction of the rows of the hinted side that is sampled to find the hot keys " +
+ "for a join with a SKEW_JOIN hint.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doubleConf
+ .checkValue(v => v > 0.0 && v <= 1.0, "The sample fraction must be in (0, 1].")
+ .createWithDefault(0.1)
+
+ val SKEW_JOIN_HINT_MIN_HOT_KEY_ROWS =
+ buildConf("spark.sql.skewJoinHint.minHotKeyRows")
+ .doc("The minimum estimated number of rows a key must have on the hinted side to be " +
+ "treated as hot for a join with a SKEW_JOIN hint. A key is hot if it has at least this " +
+ "many rows and at least as many rows as the average shuffle partition, which is the " +
+ s"estimated row count of the hinted side divided by '${SHUFFLE_PARTITIONS.key}'.")
Review Comment:
**Nit (P3):** `findHotKeys` divides by `conf.numShufflePartitions`. That returns `spark.sql.adaptive.coalescePartitions.initialPartitionNum` when AQE and partition coalescing are enabled (the default) and that config is set, so with `initialPartitionNum=2000` and `spark.sql.shuffle.partitions=200` the threshold uses 2000, not the documented 200. The cold-branch shuffle also uses the effective number, so the doc should describe that rather than name `spark.sql.shuffle.partitions`.
**Verification:**
- **Inspection:** The minHotKeyRows doc names the same partition count that hot-key selection uses, including when adaptive partition coalescing sets an initial partition number.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -339,7 +399,8 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
if (hint.isEmpty) {
createJoinWithoutHint()
} else {
- createBroadcastHashJoin(true)
+ createSkewKeyJoin()
Review Comment:
**Nit (P3):** With `createSkewKeyJoin()` first, `/*+ SKEW_JOIN(t1(k)), BROADCAST(t2) */` runs a sampling job plus a shuffled sort merge join instead of the requested broadcast join, with no `hintOverridden` warning. The SQL hints reference, `sql-performance-tuning.md`, the `JoinStrategyHint` scaladoc and the hint-order comment in `JoinSelection` all still say BROADCAST over MERGE over SHUFFLE_HASH over SHUFFLE_REPLICATE_NL. The hint itself is not documented either: there is nothing on the `relation(columns)` syntax, the supported join types and sides, which cases are ignored and which rejected, or the `spark.sql.skewJoinHint.*` configs. Broadcasting the other side already avoids the single-task hot key, so keeping BROADCAST first and ranking SKEW_JOIN right after it seems the natural order. Whichever order you pick, please document it consistently and add a test that combines the hints.
**Verification:**
- **Behavior:** A join with SKEW_JOIN on one side and BROADCAST on the other plans a broadcast hash join, and with MERGE on the other side it plans the skew split.
- **Inspection:** The SQL hints reference, performance tuning guide, JoinStrategyHint scaladoc and planner hint-order comment state the same priority order, including SKEW_JOIN, and the reference documents its syntax, supported joins, error cases and configs.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SkewKeyJoinExec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.joins
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, SparkPlan, UnionExec}
+import org.apache.spark.sql.execution.exchange.EnsureRequirements
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Joins two sides on equi-join keys, where some keys are so frequent on both sides that all their
+ * output rows would end up in a single task of a shuffled join. The shuffle blocks of such a key
+ * can all be small, as the key is spread thinly over many mappers, so adaptive skew join handling,
+ * which looks at the sizes of the shuffle blocks, does not help.
Review Comment:
**Nit (P3):** `OptimizeSkewedJoin` decides skew from `mapStats.bytesByPartitionId`, the total size of each reduce partition summed over all mappers. A key spread thinly over many mappers therefore still shows up as a large partition and gets split. The case AQE misses is a key whose partition stays below `max(skewedPartitionThresholdInBytes, skewedPartitionFactor * median)` while its output (rows on one side times rows on the other) explodes. The SKEW_JOIN scaladoc in `hints.scala` ('even though no shuffle block is oversized') uses the same framing.
**Verification:**
- **Inspection:** The operator and hint docs attribute AQE's skew detection to reduce-partition size totals, not block sizes, and describe the output-explosion case the hint targets.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56906][SQL] Add SKEW_JOIN hint with SkewKeyJoinExec [spark]
Posted by "pratham76 (via GitHub)" <gi...@apache.org>.
pratham76 commented on PR #59700:
URL: https://github.com/apache/spark/pull/59700#issuecomment-6339223075
@cloud-fan @ulysses-you @pan3793 Could you PTAL? Thanks!
FYI @xumingming
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org