LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4034934870
##########
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 || {
Review Comment:
**HIGH**
`SPARK-36082: left-broadcast NAAJ fallback uses nested-loop join` in
`JoinSuite.scala:1311` will fail. It sets `autoBroadcastJoinThreshold =
Long.MaxValue` with the dedicated threshold at `0`, which pins exactly the case
where regular planning would broadcast the right side and the dedicated
threshold disables the hash optimization. With the floor, `max(0,
Long.MaxValue)` puts any right side under the threshold, so the plan becomes a
null-aware `BroadcastHashJoinExec` and all three of that test's plan assertions
fail, the first one aborting it. The answer itself is unchanged.
That test came with SPARK-36082 itself, so either the behavior it records is
no longer wanted, in which case this PR should update it and say why, or the
floor needs to not cover this configuration. It is also the only case covering
the hint-broadcasts-the-left-side path down to the nested-loop fallback, so
nothing guards that path once it changes.
##########
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:
**MEDIUM**
`canPlanAsBroadcastHashJoin` has a second consumer,
`PushDownLeftSemiAntiJoin` (`:68`), which uses the answer to decide whether to
push a LeftSemi/Anti join below an `Aggregate`. The question there is whether
the join stays an O(M) hash join after the push, measured against the
pre-aggregation row count, so "the fallback would broadcast the right side
anyway" does not transfer: a BNLJ below the aggregate is O(M * N) with an
un-aggregated M.
The risk already existed for size-gated values, a positive dedicated
threshold, and the floor adds the 0-to-static-threshold band to it; nothing
lifts a pushed-down join back above the aggregate. If only the cost decision in
join selection is meant to change, `PushDownLeftSemiAntiJoin` could ask an
unfloored predicate, or the description could state that widening the rewrite
is intended. Either way, `SQLConf.scala:7463` ("This configuration also
controls whether a null-aware anti join can be pushed below an aggregate") is
now incomplete, since the floor puts `spark.sql.autoBroadcastJoinThreshold` in
charge of that pushdown too; and no case in `LeftSemiAntiJoinPushDownSuite`
sets either conf today, so one would record whichever decision you take.
##########
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)
Review Comment:
**MEDIUM**
The floor reads `conf.autoBroadcastJoinThreshold`, but once this branch
returns `None`, whether the fallback broadcasts the right side is decided by
`canBroadcastBySize` (`joins.scala:363`), which switches to
`spark.sql.adaptive.autoBroadcastJoinThreshold` for runtime stats. Under AQE
the two can use different thresholds: with the adaptive limit at 1MB, the
static one at 10MB and the dedicated threshold at 0, a right stage measured at
5MB is admitted here and broadcast, while regular planning would have broadcast
the 512KB left side instead. With the adaptive limit above the static one it
errs the other way and rejects a case the fallback really would broadcast.
Replacing only the `math.max` half with the planner's own predicate, so the
shape becomes `dedicatedThreshold < 0 || canBroadcastBySize(j.right, conf) ||
(dedicatedThreshold > 0 && rightSize >= 0 && rightSize <= dedicatedThreshold)`,
makes the rationale hold on its own and the new comment true as written. One
constraint: the second block of this PR's new `short-circuits config-only
decisions` pins that a config-only rejection never reads stats (`auto = -1`,
dedicated `0`, a right side whose stats throw), and `canBroadcastBySize` has to
read `stats.isRuntime` to pick its threshold, so that config-only rejection
needs to stay ahead of it.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- val overLongMaxRight = right.copy(
- rowCount = BigInt(Long.MaxValue) + 1,
- size = Some(BigInt(Long.MaxValue) + 1))
withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get) ===
Some(BuildRight))
+ 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.copy(right = largeRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(autoThresholdRight), SQLConf.get) ===
Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
}
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
+ 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.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
}
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(),
SQLConf.get).isEmpty)
}
+ }
- withSQLConf(
- SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
- SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ test("NAAJ broadcast threshold short-circuits config-only decisions") {
+ case class ThrowingStatsPlan() extends LeafNode {
Review Comment:
**LOW**
`ThrowingStatsPlan` overrides only `output`; "reading `stats` throws" comes
from `LeafNode.computeStats`. The test's name claims short-circuiting, but that
premise appears nowhere in the test. If `LeafNode` ever gains a fallback
estimate, both assertions stay green while proving nothing.
Overriding `computeStats()` in that class to throw explicitly, or one
comment naming `LeafNode.computeStats`, puts the premise inside the test.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- val overLongMaxRight = right.copy(
- rowCount = BigInt(Long.MaxValue) + 1,
- size = Some(BigInt(Long.MaxValue) + 1))
withSQLConf(
Review Comment:
**MEDIUM**
Each of the three new tests sets `NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD`
explicitly, so the old block that left the dedicated conf at its default and
asserted `Some(BuildRight)` for `largeRight` is gone. Nothing pins the default
now: change it from -1 to a byte size and the suite stays green, while NAAJ
planning goes from size-blind to size-gated, which is user visible. (Changing
it to 0 is caught by `JoinSuite.scala:1368`.)
Cheap to restore: add a block that leaves the dedicated conf alone and
asserts `largeRight -> Some(BuildRight)`, rather than repurposing one of the
three existing blocks, each of which pins a side of the floor.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- val overLongMaxRight = right.copy(
- rowCount = BigInt(Long.MaxValue) + 1,
- size = Some(BigInt(Long.MaxValue) + 1))
withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get) ===
Some(BuildRight))
+ 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.copy(right = largeRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(autoThresholdRight), SQLConf.get) ===
Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
}
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
+ 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.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
}
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(),
SQLConf.get).isEmpty)
}
+ }
- withSQLConf(
- SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
- SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ test("NAAJ broadcast threshold short-circuits config-only decisions") {
+ case class ThrowingStatsPlan() extends LeafNode {
+ override def output: Seq[Attribute] = right.output
}
+ val nullAwareAntiJoinWithoutStats = nullAwareAntiJoin(ThrowingStatsPlan())
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key ->
"10MB") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get) ===
Some(BuildRight))
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get).isEmpty)
+ nullAwareAntiJoinWithoutStats, SQLConf.get) === Some(BuildRight))
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoinWithoutStats,
SQLConf.get).isEmpty)
+ }
+ }
+
+ test("NAAJ broadcast threshold rejects unknown sizes and respects the
optimization flag") {
+ withSQLConf(
Review Comment:
**MEDIUM**
The first block of `NAAJ broadcast threshold rejects unknown sizes and
respects the optimization flag` is the only configuration with automatic
broadcasting off and a positive dedicated threshold, and it only asserts the
negative-size rejection. The old positive case under `threshold = 10MB` is
gone, so nothing covers that combination.
One more assertion in that block, that `nullAwareAntiJoin()` gives
`Some(BuildRight)`, closes it. Separately, no block lands in `0 < dedicated <
auto`, so an implementation written as `if (dedicatedThreshold == 0) auto else
dedicated` also survives the suite; one block in that band would record the
"larger of the two" the doc promises.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,8 +7452,11 @@ 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 effective " +
Review Comment:
**LOW**
With the floor, any nonnegative dedicated threshold at or below
`spark.sql.autoBroadcastJoinThreshold` is indistinguishable from not setting it
at all, so `0` is no longer an off switch. A user who wants automatic
broadcasting untouched but the NAAJ hash optimization off is left with the
internal `spark.sql.optimizeNullAwareAntiJoin=false`, and the NAAJ key
normalization it also turns off does not presuppose the hash join was chosen:
it rewrites the condition during logical optimization, so it applies on the
nested-loop path too, which is what `JoinSuite.scala:1343` covers.
The doc states the new semantics; naming that replacement switch would tell
a reader who set 0 under the old doc where to go instead. Also, the
description's "To disable both, set both thresholds to a nonpositive value" is
inverted: a negative dedicated threshold means unbounded, so following it does
the opposite. Neither this conf's commit nor its parent is in any tag yet, so
no migration note is needed.
--
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]
