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: [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]
