ulysses-you commented on code in PR #59703:
URL: https://github.com/apache/spark/pull/59703#discussion_r4355920098
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala:
##########
@@ -211,67 +211,138 @@ case class OptimizeSkewedJoin(ensureRequirements:
EnsureRequirements)
}
}
- def optimizeSkewJoin(plan: SparkPlan): SparkPlan = {
- val localShuffleReaderEnabled =
conf.getConf(SQLConf.LOCAL_SHUFFLE_READER_ENABLED)
- plan.transformUp {
+ /**
+ * The candidate `plan` stands for, when it is a sort-merge, shuffled-hash
or broadcast-hash join
+ * with a skewed shuffled side this rule can split.
+ */
+ private def splitSkewJoin(
+ plan: SparkPlan,
+ localShuffleReaderEnabled: Boolean): Option[SkewJoinCandidate] =
+ plan match {
case smj @ SortMergeJoinExec(_, _, joinType, _,
s1 @ SortExec(_, _, ShuffleStage(left: ShuffleQueryStageExec), _),
s2 @ SortExec(_, _, ShuffleStage(right: ShuffleQueryStageExec), _),
false) =>
tryOptimizeJoinChildren(left, right, joinType).map {
case (newLeft, newRight) =>
- smj.copy(
- left = s1.copy(child = newLeft), right = s2.copy(child =
newRight), isSkewJoin = true)
- }.getOrElse(smj)
+ SkewJoinCandidate(smj, joinType, smj.copy(
+ left = s1.copy(child = newLeft), right = s2.copy(child =
newRight),
+ isSkewJoin = true))
+ }
case shj @ ShuffledHashJoinExec(_, _, joinType, _, _,
ShuffleStage(left: ShuffleQueryStageExec),
ShuffleStage(right: ShuffleQueryStageExec), false) =>
tryOptimizeJoinChildren(left, right, joinType).map {
case (newLeft, newRight) =>
- shj.copy(left = newLeft, right = newRight, isSkewJoin = true)
- }.getOrElse(shj)
+ SkewJoinCandidate(
+ shj, joinType, shj.copy(left = newLeft, right = newRight,
isSkewJoin = true))
+ }
// When localShuffleReader is disabled, the streamedPlan of
BroadcastHashJoin may be skewed,
// so we try to optimize it.
- case bhj @ BroadcastHashJoinExec(_, _, _, BuildRight, _,
+ case bhj @ BroadcastHashJoinExec(_, _, joinType, BuildRight, _,
ShuffleStage(left: ShuffleQueryStageExec), _, _, false) if
!localShuffleReaderEnabled =>
tryOptimizeBroadcastHashJoinStreamedPlan(left).map {
- case newLeft => bhj.copy(left = newLeft, isSkewJoin = true)
- }.getOrElse(bhj)
- case bhj @ BroadcastHashJoinExec(_, _, _, BuildLeft, _, _,
+ case newLeft =>
+ SkewJoinCandidate(bhj, joinType, bhj.copy(left = newLeft,
isSkewJoin = true))
+ }
+ case bhj @ BroadcastHashJoinExec(_, _, joinType, BuildLeft, _, _,
ShuffleStage(right: ShuffleQueryStageExec), _, false) if
!localShuffleReaderEnabled =>
tryOptimizeBroadcastHashJoinStreamedPlan(right).map {
- case newRight => bhj.copy(right = newRight, isSkewJoin = true)
- }.getOrElse(bhj)
+ case newRight =>
+ SkewJoinCandidate(bhj, joinType, bhj.copy(right = newRight,
isSkewJoin = true))
+ }
+
+ case _ => None
}
+
+ /**
+ * Every join in `plan` this rule can split, in the bottom-up order a
`transformUp` visits them.
+ */
+ private def collectSplitSkewJoins(plan: SparkPlan): Seq[SkewJoinCandidate] =
{
+ val localShuffleReaderEnabled =
conf.getConf(SQLConf.LOCAL_SHUFFLE_READER_ENABLED)
+ val candidates = mutable.ArrayBuffer.empty[SkewJoinCandidate]
+ plan.foreachUp { operator =>
+ splitSkewJoin(operator, localShuffleReaderEnabled).foreach(candidates +=
_)
+ }
+ candidates.toSeq
}
- override def apply(plan: SparkPlan): SparkPlan = {
- if (!conf.getConf(SQLConf.SKEW_JOIN_ENABLED)) {
+ /**
+ * `plan` with `candidates` replaced by their splits. Candidates are matched
by node identity,
+ * so structurally equal joins in different places are distinct candidates.
A subset may be
+ * passed, keeping the others out of the result.
+ *
+ * The walk is top-down: a sort-merge or shuffled-hash candidate has shuffle
stages as children
+ * and contains nothing else, but a broadcast-hash join only constrains its
streamed side, so
+ * another candidate can sit in its build side. Replacing the outer join
first keeps that inner
+ * candidate an instance `plan` holds, and the walk then splits it inside
the replacement.
+ */
+ private[adaptive] def applySkewJoins(
+ plan: SparkPlan,
+ candidates: Seq[SkewJoinCandidate]): SparkPlan = {
+ if (candidates.isEmpty) {
return plan
}
+ plan.transformDown { operator =>
+ candidates.find(_.join eq
operator).map(_.replacement).getOrElse(operator)
+ }
+ }
- // We try to optimize every skewed sort-merge/shuffle-hash joins in the
query plan. If this
- // introduces extra shuffles, we give up the optimization and return the
original query plan, or
- // accept the extra shuffles if the force-apply config is true.
- // TODO: It's possible that only one skewed join in the query plan leads
to extra shuffles and
- // we only need to skip optimizing that join. We should make the
strategy smarter here.
- val optimized = optimizeSkewJoin(plan)
- val requirementSatisfied = if
(ensureRequirements.requiredDistribution.isDefined) {
- ValidateRequirements.validate(optimized,
ensureRequirements.requiredDistribution.get)
- } else {
- ValidateRequirements.validate(optimized)
+ private def requirementsSatisfied(plan: SparkPlan): Boolean = {
+ ensureRequirements.requiredDistribution match {
+ case Some(distribution) => ValidateRequirements.validate(plan,
distribution)
+ case None => ValidateRequirements.validate(plan)
}
- if (requirementSatisfied) {
- optimized.transform {
- case SkewJoinChildWrapper(child) => child
+ }
+
+ private def stripSkewJoinChildWrappers(plan: SparkPlan): SparkPlan = {
+ plan.transform { case SkewJoinChildWrapper(child) => child }
+ }
+
+ /**
+ * Applies the candidates one at a time, in the order a bottom-up pass
visits them, keeping a
+ * candidate when the plan holding the kept ones and this one still
satisfies every requirement.
+ * A split the plan cannot take therefore costs only its own join, where the
whole-plan fast path
+ * would have cost every join.
+ *
+ * Each trial is rebuilt from `plan` rather than from the previous trial, so
acceptance never
+ * depends on how a rebuild kept node instances; the handful of skewed joins
keeps that cheap.
+ */
+ private def applySkewJoinsSatisfyingRequirements(
+ plan: SparkPlan,
+ candidates: Seq[SkewJoinCandidate]): SparkPlan = {
+ var accepted = Seq.empty[SkewJoinCandidate]
+ candidates.foreach { candidate =>
+ val trial = applySkewJoins(plan, accepted :+ candidate)
+ if (requirementsSatisfied(trial)) {
+ accepted = accepted :+ candidate
+ } else {
+ logDebug(s"Skip splitting the skewed ${candidate.joinType} join " +
+ s"${candidate.join.nodeName}: with it the plan fails
ValidateRequirements")
}
+ }
+ logDebug(s"Optimized ${accepted.length} of ${candidates.length} skewed
joins")
+ stripSkewJoinChildWrappers(applySkewJoins(plan, accepted))
+ }
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ if (!conf.getConf(SQLConf.SKEW_JOIN_ENABLED)) {
+ return plan
+ }
+
+ // Try every split at once first: one pass to find the candidates and one
validation cover the
+ // common case. When that breaks a requirement of the plan, either accept
the extra shuffles if
+ // the force-apply config is true, or keep only the joins whose splits the
plan can take. The
+ // candidates are collected once, so the fallback reuses the splits the
fast path built.
+ val candidates = collectSplitSkewJoins(plan)
+ val optimized = applySkewJoins(plan, candidates)
+ if (requirementsSatisfied(optimized)) {
+ stripSkewJoinChildWrappers(optimized)
} else if (conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN)) {
- ensureRequirements.apply(optimized).transform {
- case SkewJoinChildWrapper(child) => child
- }
+ stripSkewJoinChildWrappers(ensureRequirements.apply(optimized))
} else {
- plan
+ applySkewJoinsSatisfyingRequirements(plan, candidates)
Review Comment:
Fixed in 98db64897e8.
`CoalesceShufflePartitions.childrenNeedCompatiblePartitioning` now also reads
`u.outputPartitioning`, so a union whose layout a split took away coalesces its
children independently instead of giving up for the whole group. The condition
short-circuits on plain unions, so they do not pay the extra `rawPartitioning`.
The first test asserts the rejected join's two reads are coalesced; with the
old predicate it fails with `List() had size 0 instead of expected size 2`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala:
##########
@@ -211,67 +211,138 @@ case class OptimizeSkewedJoin(ensureRequirements:
EnsureRequirements)
}
}
- def optimizeSkewJoin(plan: SparkPlan): SparkPlan = {
- val localShuffleReaderEnabled =
conf.getConf(SQLConf.LOCAL_SHUFFLE_READER_ENABLED)
- plan.transformUp {
+ /**
+ * The candidate `plan` stands for, when it is a sort-merge, shuffled-hash
or broadcast-hash join
+ * with a skewed shuffled side this rule can split.
+ */
+ private def splitSkewJoin(
+ plan: SparkPlan,
+ localShuffleReaderEnabled: Boolean): Option[SkewJoinCandidate] =
+ plan match {
case smj @ SortMergeJoinExec(_, _, joinType, _,
s1 @ SortExec(_, _, ShuffleStage(left: ShuffleQueryStageExec), _),
s2 @ SortExec(_, _, ShuffleStage(right: ShuffleQueryStageExec), _),
false) =>
tryOptimizeJoinChildren(left, right, joinType).map {
case (newLeft, newRight) =>
- smj.copy(
- left = s1.copy(child = newLeft), right = s2.copy(child =
newRight), isSkewJoin = true)
- }.getOrElse(smj)
+ SkewJoinCandidate(smj, joinType, smj.copy(
+ left = s1.copy(child = newLeft), right = s2.copy(child =
newRight),
+ isSkewJoin = true))
+ }
case shj @ ShuffledHashJoinExec(_, _, joinType, _, _,
ShuffleStage(left: ShuffleQueryStageExec),
ShuffleStage(right: ShuffleQueryStageExec), false) =>
tryOptimizeJoinChildren(left, right, joinType).map {
case (newLeft, newRight) =>
- shj.copy(left = newLeft, right = newRight, isSkewJoin = true)
- }.getOrElse(shj)
+ SkewJoinCandidate(
+ shj, joinType, shj.copy(left = newLeft, right = newRight,
isSkewJoin = true))
+ }
// When localShuffleReader is disabled, the streamedPlan of
BroadcastHashJoin may be skewed,
// so we try to optimize it.
- case bhj @ BroadcastHashJoinExec(_, _, _, BuildRight, _,
+ case bhj @ BroadcastHashJoinExec(_, _, joinType, BuildRight, _,
ShuffleStage(left: ShuffleQueryStageExec), _, _, false) if
!localShuffleReaderEnabled =>
tryOptimizeBroadcastHashJoinStreamedPlan(left).map {
- case newLeft => bhj.copy(left = newLeft, isSkewJoin = true)
- }.getOrElse(bhj)
- case bhj @ BroadcastHashJoinExec(_, _, _, BuildLeft, _, _,
+ case newLeft =>
+ SkewJoinCandidate(bhj, joinType, bhj.copy(left = newLeft,
isSkewJoin = true))
+ }
+ case bhj @ BroadcastHashJoinExec(_, _, joinType, BuildLeft, _, _,
ShuffleStage(right: ShuffleQueryStageExec), _, false) if
!localShuffleReaderEnabled =>
tryOptimizeBroadcastHashJoinStreamedPlan(right).map {
- case newRight => bhj.copy(right = newRight, isSkewJoin = true)
- }.getOrElse(bhj)
+ case newRight =>
+ SkewJoinCandidate(bhj, joinType, bhj.copy(right = newRight,
isSkewJoin = true))
+ }
+
+ case _ => None
}
+
+ /**
+ * Every join in `plan` this rule can split, in the bottom-up order a
`transformUp` visits them.
+ */
+ private def collectSplitSkewJoins(plan: SparkPlan): Seq[SkewJoinCandidate] =
{
+ val localShuffleReaderEnabled =
conf.getConf(SQLConf.LOCAL_SHUFFLE_READER_ENABLED)
+ val candidates = mutable.ArrayBuffer.empty[SkewJoinCandidate]
+ plan.foreachUp { operator =>
+ splitSkewJoin(operator, localShuffleReaderEnabled).foreach(candidates +=
_)
+ }
+ candidates.toSeq
}
- override def apply(plan: SparkPlan): SparkPlan = {
- if (!conf.getConf(SQLConf.SKEW_JOIN_ENABLED)) {
+ /**
+ * `plan` with `candidates` replaced by their splits. Candidates are matched
by node identity,
+ * so structurally equal joins in different places are distinct candidates.
A subset may be
+ * passed, keeping the others out of the result.
+ *
+ * The walk is top-down: a sort-merge or shuffled-hash candidate has shuffle
stages as children
+ * and contains nothing else, but a broadcast-hash join only constrains its
streamed side, so
+ * another candidate can sit in its build side. Replacing the outer join
first keeps that inner
+ * candidate an instance `plan` holds, and the walk then splits it inside
the replacement.
+ */
+ private[adaptive] def applySkewJoins(
+ plan: SparkPlan,
+ candidates: Seq[SkewJoinCandidate]): SparkPlan = {
+ if (candidates.isEmpty) {
return plan
}
+ plan.transformDown { operator =>
+ candidates.find(_.join eq
operator).map(_.replacement).getOrElse(operator)
+ }
+ }
- // We try to optimize every skewed sort-merge/shuffle-hash joins in the
query plan. If this
- // introduces extra shuffles, we give up the optimization and return the
original query plan, or
- // accept the extra shuffles if the force-apply config is true.
- // TODO: It's possible that only one skewed join in the query plan leads
to extra shuffles and
- // we only need to skip optimizing that join. We should make the
strategy smarter here.
- val optimized = optimizeSkewJoin(plan)
- val requirementSatisfied = if
(ensureRequirements.requiredDistribution.isDefined) {
- ValidateRequirements.validate(optimized,
ensureRequirements.requiredDistribution.get)
- } else {
- ValidateRequirements.validate(optimized)
+ private def requirementsSatisfied(plan: SparkPlan): Boolean = {
+ ensureRequirements.requiredDistribution match {
+ case Some(distribution) => ValidateRequirements.validate(plan,
distribution)
+ case None => ValidateRequirements.validate(plan)
}
- if (requirementSatisfied) {
- optimized.transform {
- case SkewJoinChildWrapper(child) => child
+ }
+
+ private def stripSkewJoinChildWrappers(plan: SparkPlan): SparkPlan = {
+ plan.transform { case SkewJoinChildWrapper(child) => child }
+ }
+
+ /**
+ * Applies the candidates one at a time, in the order a bottom-up pass
visits them, keeping a
+ * candidate when the plan holding the kept ones and this one still
satisfies every requirement.
+ * A split the plan cannot take therefore costs only its own join, where the
whole-plan fast path
+ * would have cost every join.
+ *
+ * Each trial is rebuilt from `plan` rather than from the previous trial, so
acceptance never
+ * depends on how a rebuild kept node instances; the handful of skewed joins
keeps that cheap.
+ */
+ private def applySkewJoinsSatisfyingRequirements(
+ plan: SparkPlan,
+ candidates: Seq[SkewJoinCandidate]): SparkPlan = {
+ var accepted = Seq.empty[SkewJoinCandidate]
+ candidates.foreach { candidate =>
+ val trial = applySkewJoins(plan, accepted :+ candidate)
+ if (requirementsSatisfied(trial)) {
+ accepted = accepted :+ candidate
+ } else {
+ logDebug(s"Skip splitting the skewed ${candidate.joinType} join " +
+ s"${candidate.join.nodeName}: with it the plan fails
ValidateRequirements")
}
+ }
+ logDebug(s"Optimized ${accepted.length} of ${candidates.length} skewed
joins")
+ stripSkewJoinChildWrappers(applySkewJoins(plan, accepted))
+ }
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ if (!conf.getConf(SQLConf.SKEW_JOIN_ENABLED)) {
+ return plan
+ }
+
+ // Try every split at once first: one pass to find the candidates and one
validation cover the
+ // common case. When that breaks a requirement of the plan, either accept
the extra shuffles if
+ // the force-apply config is true, or keep only the joins whose splits the
plan can take. The
+ // candidates are collected once, so the fallback reuses the splits the
fast path built.
+ val candidates = collectSplitSkewJoins(plan)
+ val optimized = applySkewJoins(plan, candidates)
+ if (requirementsSatisfied(optimized)) {
+ stripSkewJoinChildWrappers(optimized)
} else if (conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN)) {
- ensureRequirements.apply(optimized).transform {
- case SkewJoinChildWrapper(child) => child
- }
+ stripSkewJoinChildWrappers(ensureRequirements.apply(optimized))
} else {
Review Comment:
Added in 98db64897e8: `spark.sql.adaptive.skewJoin.perJoinFallback.enabled`,
`.internal()`, default true, `version("4.4.0")`. Turning it off returns `plan`,
as this branch did before the PR, and the first test runs its query with the
config off to assert neither join is split.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -1901,6 +1901,144 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-60458: skip only the skewed join whose split the plan cannot
take") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100") {
+ withTempView("skewData1", "skewData2") {
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 3 as key1", "id as value1")
+ .createOrReplaceTempView("skewData1")
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 1 as key2", "id as value2")
+ .createOrReplaceTempView("skewData2")
+
+ val query =
+ """
+ |SELECT key1, count(*) FROM skewData1 JOIN skewData2 ON key1 =
key2 GROUP BY key1
+ |UNION ALL
+ |SELECT key1, 1 FROM skewData1 JOIN skewData2 ON key1 = key2
+ |""".stripMargin
+
+ // The aggregate above the first join needs its input on `key1`, which
the split breaks.
+ // The second join feeds the union directly, so its split breaks
nothing: it is kept.
+ val (_, adaptivePlan) = runAdaptiveAndVerifyResult(query)
+
+ // key2 is always 0, so only the 334 rows of key1 = 0 join, each
against all 1000 rows of
+ // the other side: the aggregated branch collapses them into one row,
the other keeps one
+ // row per joined row.
+ val rows = sql(query).collect()
+ assert(rows.length == 334001)
+ assert(rows.count(_ == Row(0L, 334000L)) == 1)
+ assert(rows.count(_ == Row(0L, 1L)) == 334000)
+
+ val aggregateJoins = collect(adaptivePlan) {
+ case agg: BaseAggregateExec =>
+ collect(agg) { case j: SortMergeJoinExec => j }
+ }.flatten
+ assert(aggregateJoins.size == 1)
Review Comment:
Changed in 98db64897e8 to `collect { case agg: BaseAggregateExec =>
findTopLevelSortMergeJoin(agg) }.flatten.distinct`. Verified by running the
test with `spark.sql.execution.combineAdjacentAggregation=false`: the previous
form failed, the new form passes.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -1901,6 +1901,144 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-60458: skip only the skewed join whose split the plan cannot
take") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100") {
+ withTempView("skewData1", "skewData2") {
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 3 as key1", "id as value1")
+ .createOrReplaceTempView("skewData1")
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 1 as key2", "id as value2")
+ .createOrReplaceTempView("skewData2")
+
+ val query =
+ """
+ |SELECT key1, count(*) FROM skewData1 JOIN skewData2 ON key1 =
key2 GROUP BY key1
+ |UNION ALL
+ |SELECT key1, 1 FROM skewData1 JOIN skewData2 ON key1 = key2
+ |""".stripMargin
+
+ // The aggregate above the first join needs its input on `key1`, which
the split breaks.
+ // The second join feeds the union directly, so its split breaks
nothing: it is kept.
+ val (_, adaptivePlan) = runAdaptiveAndVerifyResult(query)
+
+ // key2 is always 0, so only the 334 rows of key1 = 0 join, each
against all 1000 rows of
+ // the other side: the aggregated branch collapses them into one row,
the other keeps one
+ // row per joined row.
+ val rows = sql(query).collect()
+ assert(rows.length == 334001)
+ assert(rows.count(_ == Row(0L, 334000L)) == 1)
+ assert(rows.count(_ == Row(0L, 1L)) == 334000)
+
+ val aggregateJoins = collect(adaptivePlan) {
+ case agg: BaseAggregateExec =>
+ collect(agg) { case j: SortMergeJoinExec => j }
+ }.flatten
+ assert(aggregateJoins.size == 1)
+ assert(!aggregateJoins.head.isSkewJoin)
+ val unionJoins = findTopLevelSortMergeJoin(adaptivePlan)
+ .filterNot(j => aggregateJoins.exists(_ eq j))
+ assert(unionJoins.size == 1)
+ assert(unionJoins.head.isSkewJoin)
+ assert(collect(unionJoins.head) {
+ case r: AQEShuffleReadExec => r
+
}.exists(_.partitionSpecs.exists(_.isInstanceOf[PartialReducerPartitionSpec])))
Review Comment:
Switched to `hasSkewedPartition` in 98db64897e8.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -1901,6 +1901,144 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-60458: skip only the skewed join whose split the plan cannot
take") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100") {
+ withTempView("skewData1", "skewData2") {
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 3 as key1", "id as value1")
+ .createOrReplaceTempView("skewData1")
+ spark
+ .range(0, 1000, 1, 10)
+ .selectExpr("id % 1 as key2", "id as value2")
+ .createOrReplaceTempView("skewData2")
+
+ val query =
+ """
+ |SELECT key1, count(*) FROM skewData1 JOIN skewData2 ON key1 =
key2 GROUP BY key1
+ |UNION ALL
+ |SELECT key1, 1 FROM skewData1 JOIN skewData2 ON key1 = key2
+ |""".stripMargin
+
+ // The aggregate above the first join needs its input on `key1`, which
the split breaks.
+ // The second join feeds the union directly, so its split breaks
nothing: it is kept.
+ val (_, adaptivePlan) = runAdaptiveAndVerifyResult(query)
+
+ // key2 is always 0, so only the 334 rows of key1 = 0 join, each
against all 1000 rows of
+ // the other side: the aggregated branch collapses them into one row,
the other keeps one
+ // row per joined row.
+ val rows = sql(query).collect()
+ assert(rows.length == 334001)
+ assert(rows.count(_ == Row(0L, 334000L)) == 1)
+ assert(rows.count(_ == Row(0L, 1L)) == 334000)
+
+ val aggregateJoins = collect(adaptivePlan) {
Review Comment:
Both now use `findTopLevelAggregate` and `findTopLevelSortMergeJoin` in
98db64897e8.
--
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]
