LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4055928686
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,12 +7452,22 @@ object SQLConf {
"single-column null-aware anti join for which Spark uses the broadcast
hash join " +
"optimization. This configuration takes effect only when " +
"spark.sql.optimizeNullAwareAntiJoin is enabled. A negative value
allows the " +
- "optimization regardless of the estimated size, while zero disables
it. If the " +
- "estimated size exceeds a positive value, Spark falls back to regular
join planning. " +
+ "optimization regardless of the estimated size. For a nonnegative
value, the " +
+ "optimization is also allowed when regular join planning considers the
right side " +
+ "broadcastable. " +
+ "For join selection, regular planning uses " +
+ "spark.sql.adaptive.autoBroadcastJoinThreshold for runtime statistics
when it is set, " +
+ "and spark.sql.autoBroadcastJoinThreshold otherwise. The same
eligibility decision " +
+ "controls whether a null-aware anti join can be pushed below an
aggregate; this " +
Review Comment:
**MEDIUM**
`canPlanAsBroadcastHashJoin` has a second consumer,
`PushDownLeftSemiAntiJoin.scala:68`, which uses the answer to decide whether a
LeftSemi/LeftAnti join can be pushed below an `Aggregate`. The floor widens
that decision too, but "the fallback would broadcast the right side anyway"
says nothing there: the pushdown leaves the right side alone and replaces the
left side with the aggregate's input, so the cost turns on how many rows the
aggregate removes, which the automatic broadcast threshold does not describe.
With a dedicated threshold of 0, the default 10MB automatic threshold, a 5MB
right side and `SELECT DISTINCT k FROM t` on the left (10^9 rows, 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, the cost estimate
is what changed. Default-configured Spark (dedicated threshold -1) answers true
either way and is unaffected.
If sharing one predicate is intended, this sentence should say that the
floor's rationale only covers join selection; if not, the pushdown gate can
keep the plain dedicated-threshold test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,26 @@ 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 = if (dedicatedThreshold < 0) {
+ true
+ } else {
+ val automaticBroadcastDisabled = conf.autoBroadcastJoinThreshold < 0 &&
+ conf.getConf(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_
< 0)
+ if (dedicatedThreshold == 0 && automaticBroadcastDisabled) {
Review Comment:
**LOW**
When `automaticBroadcastDisabled` holds, the threshold `canBroadcastBySize`
picks is negative and the check is always false, and `dedicatedThreshold == 0`
makes the right half of the `||` false. This branch therefore changes no return
value in any configuration; it only decides whether `j.right.stats` is read.
Skipping that read is worth something (a DSv2 `estimateStatistics` is not
cheap), but the optimization cannot be made complete: deciding which threshold
applies means reading `stats.isRuntime` first, so the conservative "both
thresholds off" test is the only sound one. A comment saying that, and a test
name that says what it actually pins (statistics are not read in these
configurations), would keep the next reader from treating the
positive-adaptive-threshold case as a missing condition.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,26 @@ 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 = if (dedicatedThreshold < 0) {
+ true
+ } else {
+ val automaticBroadcastDisabled = conf.autoBroadcastJoinThreshold < 0 &&
+ conf.getConf(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_
< 0)
+ if (dedicatedThreshold == 0 && automaticBroadcastDisabled) {
+ false
+ } else {
+ canBroadcastBySize(j.right, conf) || (dedicatedThreshold > 0 && {
Review Comment:
**HIGH**
The floor's rationale is that the fallback would broadcast the right side
anyway, but the fallback's build side is decided in
`SparkStrategies.scala:364-433` from the hints: with `BROADCAST` on the left
and nothing on the right it is `BuildLeft`, and the right side is never
broadcast. `JoinSuite.scala:1340` pins exactly that configuration, and the
assertion it replaced was `BuildLeft`. With `autoBroadcastJoinThreshold=100MB`,
a dedicated threshold of 0, a 1KB hinted left side and a 90MB right side, the
old plan broadcast 1KB and the new one broadcasts 90MB; the default dedicated
threshold of -1 is unaffected.
Even when the fallback does build right, the two plans do not ship the same
bytes. `BroadcastNestedLoopJoinExec` uses `IdentityBroadcastMode` and ships the
rows as they are, while the `HashedRelation` the hash plan ships keeps a copy
of the key row per entry (`HashedRelation.scala:477-498`).
Either add a "the fallback really would build right" condition to the floor
(the test is the negation of `hintToBroadcastLeft(j.hint) &&
!hintToBroadcastRight(j.hint)`, at the cost of reverting the assertions in
`JoinSuite.scala:1338-1340`), or restate the tradeoff in the comment, the
config doc and the PR description: the O(M) hash plan is worth a larger
broadcast even when the fallback would broadcast the left side.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,12 +7452,22 @@ object SQLConf {
"single-column null-aware anti join for which Spark uses the broadcast
hash join " +
"optimization. This configuration takes effect only when " +
"spark.sql.optimizeNullAwareAntiJoin is enabled. A negative value
allows the " +
- "optimization regardless of the estimated size, while zero disables
it. If the " +
- "estimated size exceeds a positive value, Spark falls back to regular
join planning. " +
+ "optimization regardless of the estimated size. For a nonnegative
value, the " +
+ "optimization is also allowed when regular join planning considers the
right side " +
+ "broadcastable. " +
+ "For join selection, regular planning uses " +
+ "spark.sql.adaptive.autoBroadcastJoinThreshold for runtime statistics
when it is set, " +
+ "and spark.sql.autoBroadcastJoinThreshold otherwise. The same
eligibility decision " +
+ "controls whether a null-aware anti join can be pushed below an
aggregate; this " +
+ "pushdown runs before adaptive execution and therefore uses estimated
statistics and " +
+ "spark.sql.autoBroadcastJoinThreshold. Thus, zero disables the
optimization only when " +
+ "automatic broadcasting is also disabled. When neither threshold
admits the right " +
+ "side, Spark falls back to regular join planning. " +
Review Comment:
**LOW**
The sentence at :7466-7467, `a nested-loop representation that uses more
memory`, comes from #58631 and is not supported. On broadcast bytes the hash
plan is the larger one: `BroadcastNestedLoopJoinExec` uses
`IdentityBroadcastMode` and returns the rows unchanged
(`broadcastMode.scala:37-46`), while a `HashedRelation` adds a key-row copy per
entry or a whole index array serialized at capacity
(`HashedRelation.scala:477-498`, `:916-932`). Resident object overhead does
lean the other way (BNLJ keeps an `UnsafeRow` plus an array slot per row), but
nobody has measured that dimension.
Left in place it tells the reader that admitting the hash plan always saves
memory over the fallback. This paragraph is already being rewritten, so either
qualify which kind of memory it means or drop the clause.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,144 @@ class JoinSelectionHelperSuite extends PlanTest with
JoinSelectionHelper {
}
}
- test("getBroadcastHashJoinBuildSide uses the null-aware anti join broadcast
threshold") {
- val leftKey = left.output.head
- val rightKey = right.output.head
- val condition = Or(EqualTo(leftKey, rightKey), IsNull(EqualTo(leftKey,
rightKey)))
- val nullAwareAntiJoin = Join(left, right, LeftAnti, Some(condition),
JoinHint.NONE)
+ test("NAAJ broadcast threshold is floored by the automatic broadcast
threshold") {
+ val autoThresholdRight = right.copy(
+ rowCount = 10 * 1024 * 1024,
+ size = Some(10 * 1024 * 1024))
+ val betweenThresholdsRight = right.copy(
+ rowCount = 8 * 1024 * 1024,
+ size = Some(8 * 1024 * 1024))
val largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
+ val emptyRight = right.copy(rowCount = 0, size = Some(0))
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get)
=== Some(BuildRight))
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(autoThresholdRight), SQLConf.get) ===
Some(BuildRight))
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "20MB") {
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
Review Comment:
**MEDIUM**
The automatic threshold's inclusive boundary is pinned (`autoThresholdRight`
is exactly 10MB and :218-219 asserts it is admitted); the dedicated one is not.
The only two places where the dedicated threshold is the deciding term are this
block (20000000 bytes against 20MB) and :332 (1000 bytes against 10MB), both
strictly below the threshold, so changing `rightSize <= dedicatedThreshold` to
`<` keeps all three suites green.
In `LeftSemiAntiJoinPushDownSuite.scala:172` the right side of the positive
block is an empty `LocalRelation` whose `sizeInBytes` is 0, which clears any
nonnegative threshold, so the block cannot show that the automatic threshold is
what admitted it; the 10MB at :170 could be 1 byte and it would still pass.
Two cases close both: a right side exactly at the dedicated threshold here,
and a nonzero `StatsTestPlan` right side in the pushdown block, with
`pushedDownQuery` rebuilt to match it.
--
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]
