LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4089451741
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +437,18 @@ 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. The
applicable automatic
+ // broadcast threshold floors a nonnegative dedicated threshold. As
before, threshold
+ // eligibility takes precedence over join hints. This same decision
intentionally controls
+ // aggregate pushdown.
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 ||
+ (dedicatedThreshold > 0 && {
+ val rightSize = j.right.stats.sizeInBytes
+ rightSize >= 0 && rightSize <= dedicatedThreshold
+ }) || canBroadcastBySize(j.right, conf)
Review Comment:
This is the disjunct the aggregate pushdown and join selection can read
differently. `canBroadcastBySize` picks
`spark.sql.adaptive.autoBroadcastJoinThreshold` when `stats.isRuntime` and the
static one otherwise (`:364-366`), while the pushdown gate
(`PushDownLeftSemiAntiJoin.scala:68`) runs in the main optimizer and
`AQEOptimizer` never re-runs that rule, so it always sees estimated statistics
and the static threshold.
With `broadcastThreshold=0`, `autoBroadcastJoinThreshold=10MB`,
`adaptive.autoBroadcastJoinThreshold=1MB` and a right side estimated at 5MB,
the anti join is pushed below the Aggregate and AQE then rejects the broadcast
on runtime statistics, which is exactly what your `uses the adaptive threshold
for runtime statistics` case pins. The plan lands on a BroadcastNestedLoopJoin
over a left side that no longer carries the Aggregate, so M is larger than it
was before this PR, and at threshold 0 the pushdown never fired at all.
The config description already says join selection "may reevaluate it with
runtime statistics"; one more clause saying the two can reach opposite answers
would make that cost visible.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +437,18 @@ 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. The
applicable automatic
+ // broadcast threshold floors a nonnegative dedicated threshold. As
before, threshold
+ // eligibility takes precedence over join hints. This same decision
intentionally controls
+ // aggregate pushdown.
Review Comment:
Since this is where the floor's rationale lives: for a LeftAnti with no
hints, `JoinSelection`'s fallback broadcasts the right side regardless.
`desiredBuildSide` is `BuildRight` for LeftAnti
(`SparkStrategies.scala:366-374`), `createBroadcastNLJoin(false)` and
`createCartesianProduct()` both decline once the automatic threshold is
negative, and the last resort at `:413-424` builds the right side.
So `broadcastThreshold=0` does not keep the right side from being broadcast;
it swaps O(M) hash probing for O(M*N) nested-loop probing. Worth a clause here,
or the opposite question: if the fallback broadcasts it either way, why gate
the floor on size at all?
--
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]
