dongjoon-hyun commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4064278570
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,33 @@ trait JoinSelectionHelper extends Logging {
getBroadcastBuildSide(join, hintOnly = true, conf).orElse {
if (noShufflePlannedBefore) getBroadcastBuildSide(join, hintOnly =
false, conf) else None
}
- // `JoinSelection` always builds from the right for this shape. A negative
threshold preserves
- // the original unbounded NAAJ behavior, while zero disables the broadcast
hash optimization.
+ // `JoinSelection` always builds from the right for this shape. A
dedicated threshold that
+ // admits the right side takes precedence over join hints. The automatic
threshold is only
+ // used as a floor when regular planning would also broadcast the right
side; a left-only
+ // broadcast hint makes the fallback broadcast the left side instead.
case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) =>
- val threshold = conf.nullAwareAntiJoinBroadcastThreshold
- val rightSize = j.right.stats.sizeInBytes
- if (threshold < 0 || (threshold > 0 && rightSize >= 0 && rightSize <=
threshold)) {
+ val dedicatedThreshold = conf.nullAwareAntiJoinBroadcastThreshold
+ val canBroadcast = if (dedicatedThreshold < 0) {
+ true
+ } else {
+ val fallbackBuildsRight =
+ !hintToBroadcastLeft(j.hint) || hintToBroadcastRight(j.hint)
+ val automaticBroadcastDisabled = conf.autoBroadcastJoinThreshold < 0 &&
+ conf.getConf(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_
< 0)
+ if (dedicatedThreshold == 0 && automaticBroadcastDisabled) {
+ // Avoid potentially expensive statistics computation when the
configurations alone
+ // determine the result. Both automatic thresholds must be disabled
because selecting
+ // between them requires reading `stats.isRuntime`.
+ false
+ } else {
+ (fallbackBuildsRight && canBroadcastBySize(j.right, conf)) ||
Review Comment:
The floor only consults `canBroadcastBySize(j.right, conf)`, so a
`BROADCAST` hint on the **right** side is not treated as a reason the fallback
would broadcast the right side. That misses the case this PR targets.
With `spark.sql.autoBroadcastJoinThreshold=10MB`,
`spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold=0`, and
```sql
SELECT /*+ BROADCAST(r) */ * FROM l LEFT ANTI JOIN r ON l.k = r.k OR
isnull(l.k = r.k)
```
where `r` is estimated at 50MB: `fallbackBuildsRight` is `true`,
`canBroadcastBySize(r)` is `false`, and `dedicatedThreshold > 0` is `false`, so
this returns `None`. `JoinSelection` then reaches the non-equi case, where
`createBroadcastNLJoin(onlyLookingAtHint = true)` sets `buildRight =
hintToBroadcastRight(hint)` and plans `BroadcastNestedLoopJoinExec(...,
BuildRight)`. The 50MB right side is broadcast anyway, now with the nested-loop
`O(M * N)` representation instead of the `O(M + N)` hash lookup.
This also makes the new config doc inaccurate for that case: regular join
planning *would* broadcast the right side here, but the optimization is not
allowed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,33 @@ trait JoinSelectionHelper extends Logging {
getBroadcastBuildSide(join, hintOnly = true, conf).orElse {
if (noShufflePlannedBefore) getBroadcastBuildSide(join, hintOnly =
false, conf) else None
}
- // `JoinSelection` always builds from the right for this shape. A negative
threshold preserves
- // the original unbounded NAAJ behavior, while zero disables the broadcast
hash optimization.
+ // `JoinSelection` always builds from the right for this shape. A
dedicated threshold that
+ // admits the right side takes precedence over join hints. The automatic
threshold is only
+ // used as a floor when regular planning would also broadcast the right
side; a left-only
+ // broadcast hint makes the fallback broadcast the left side instead.
case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) =>
- val threshold = conf.nullAwareAntiJoinBroadcastThreshold
- val rightSize = j.right.stats.sizeInBytes
- if (threshold < 0 || (threshold > 0 && rightSize >= 0 && rightSize <=
threshold)) {
+ val dedicatedThreshold = conf.nullAwareAntiJoinBroadcastThreshold
+ val canBroadcast = if (dedicatedThreshold < 0) {
+ true
+ } else {
+ val fallbackBuildsRight =
+ !hintToBroadcastLeft(j.hint) || hintToBroadcastRight(j.hint)
Review Comment:
`fallbackBuildsRight` does not account for
`hintToNotBroadcastAndReplicateRight`, which the fallback *does* veto on, so
the comment's invariant above ("the automatic threshold is only used as a floor
when regular planning would also broadcast the right side") can break.
For a NAAJ-shaped join carrying `JoinHint(None,
Some(HintInfo(Some(NO_BROADCAST_AND_REPLICATION))))` with
`broadcastThreshold=0`, `autoBroadcastJoinThreshold=10MB` and a 1MB right side,
`fallbackBuildsRight` and `canBroadcastBySize(j.right, conf)` are both `true`,
so a right-side broadcast is planned. Regular planning would not have done
that: `createBroadcastNLJoin(false)` computes `buildRight =
canBroadcastBySize(right) && !hintToNotBroadcastAndReplicateRight(hint)` =
`false`, and `getBroadcastNestedLoopJoinBuildSide` forces `BuildLeft`.
Practical reach is limited today -- `NO_BROADCAST_AND_REPLICATION` has no
hint aliases, and `RewriteMergeIntoTable` only ever attaches it on the left --
so this is mostly about keeping the invariant true, and it falls out of the
same predicate change as the other comment.
--
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]
