LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4056293303
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,19 @@ 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. Do not
reject the hash
+ // optimization when regular join planning would broadcast the right side,
as the fallback
+ // would still broadcast it with a slower nested-loop join.
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 = dedicatedThreshold < 0 || {
+ val effectiveThreshold = math.max(dedicatedThreshold,
conf.autoBroadcastJoinThreshold)
+ effectiveThreshold > 0 && {
+ val rightSize = j.right.stats.sizeInBytes
+ rightSize >= 0 && rightSize <= effectiveThreshold
+ }
+ }
+ if (canBroadcast) {
Review Comment:
The intentional widening is fine by me. One number worth having in the
thread, since the doc now points both consumers at the same predicate:
dedicated threshold `0`, the default 10MB automatic threshold, a 5MB right
side, and `SELECT DISTINCT k FROM t` on the left with 10^9 rows and 10^6
distinct keys. The anti join used to stay above the aggregate and probe 10^6
rows; it is now pushed below and probes 10^9. Both plans are correct, only the
cost estimate moved, and it takes an explicitly set dedicated threshold to get
there.
One clause in the config doc would help: the fallback-broadcasts-anyway
argument is what justifies the floor for join selection, while the pushdown's
cost turns on how many rows the aggregate removes.
--
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]
