ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-35282][SQL][FOLLOWUP] Simplify condition code of shuffled hash join

Browse source 
### What changes were proposed in this pull request?

Simplify the condition code which is introduced by [SPARK-35282](https://issues.apache.org/jira/browse/SPARK-35282).

### Why are the changes needed?

Reduce the code size and make code more readable.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Pass CI

Closes #33046 from ulysses-you/simplify-shj.

Authored-by: ulysses-you <ulyssesyou18@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
This commit is contained in:
ulysses-you 2021-06-24 08:42:24 +00:00 committed by Wenchen Fan
parent 5e77ca8071
commit ff9ba89dcb
2 changed files with 7 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
View file

@ -272,28 +272,16 @@ trait JoinSelectionHelper {
val buildLeft = if (hintOnly) { val buildLeft = if (hintOnly) {
hintToShuffleHashJoinLeft(hint) hintToShuffleHashJoinLeft(hint)
} else { } else {
if (hintToPreferShuffleHashJoinLeft(hint)) { hintToPreferShuffleHashJoinLeft(hint) ||
true (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(left, conf) &&
} else { muchSmaller(left, right))
if (!conf.preferSortMergeJoin) {
canBuildLocalHashMapBySize(left, conf) && muchSmaller(left, right)
} else {
false
}
}
} }
val buildRight = if (hintOnly) { val buildRight = if (hintOnly) {
hintToShuffleHashJoinRight(hint) hintToShuffleHashJoinRight(hint)
} else { } else {
if (hintToPreferShuffleHashJoinRight(hint)) { hintToPreferShuffleHashJoinRight(hint) ||
true (!conf.preferSortMergeJoin && canBuildLocalHashMapBySize(right, conf) &&
} else { muchSmaller(right, left))
if (!conf.preferSortMergeJoin) {
canBuildLocalHashMapBySize(right, conf) && muchSmaller(right, left)
} else {
false
}
}
} }
getBuildSide( getBuildSide(
canBuildShuffledHashJoinLeft(joinType) && buildLeft, canBuildShuffledHashJoinLeft(joinType) && buildLeft,

5
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala
View file

@ -45,11 +45,8 @@ object DynamicJoinSelection extends Rule[LogicalPlan] {
val maxShuffledHashJoinLocalMapThreshold = val maxShuffledHashJoinLocalMapThreshold =
conf.getConf(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD) conf.getConf(SQLConf.ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD)
val advisoryPartitionSize = conf.getConf(SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES) val advisoryPartitionSize = conf.getConf(SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES)
if (advisoryPartitionSize <= maxShuffledHashJoinLocalMapThreshold) { advisoryPartitionSize <= maxShuffledHashJoinLocalMapThreshold &&
mapStats.bytesByPartitionId.forall(_ <= maxShuffledHashJoinLocalMapThreshold) mapStats.bytesByPartitionId.forall(_ <= maxShuffledHashJoinLocalMapThreshold)
} else {
false
}
} }
private def selectJoinStrategy(plan: LogicalPlan): Option[JoinStrategyHint] = plan match { private def selectJoinStrategy(plan: LogicalPlan): Option[JoinStrategyHint] = plan match {