You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "LuciferYang (via GitHub)" <gi...@apache.org> on 2026/09/21 03:34:55 UTC
Re: [PR] [SPARK-59673][SQL] Floor the NAAJ broadcast threshold [spark]
LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4059088142
##########
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 =
Review Comment:
**LOW**
`!hintToBroadcastLeft(j.hint) || hintToBroadcastRight(j.hint)` really tests "there is no broadcast hint on the left side only", which is weaker than the name: with no hint at all, a broadcastable left side and a right side that is not, it is true while the fallback builds the left side (`SparkStrategies.scala:376-403`). Nothing shows today because it is chained with `canBroadcastBySize(j.right, conf)` by `&&`, and that case makes the whole term false anyway.
Renaming it (`noLeftOnlyBroadcastHint`, say), or noting beside it that it only holds inside this `&&`, would keep a later change from reusing it elsewhere.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,161 @@ 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)
- val largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
+ 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 = 20 * 1024 * 1024,
+ size = Some(20 * 1024 * 1024))
+ 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(autoThresholdRight).copy(
+ hint = JoinHint(hintBroadcast, None)), SQLConf.get).isEmpty)
Review Comment:
**MEDIUM**
The three new hint assertions (:222-224, :235-237, :338-340) all carry BROADCAST on the left only, and no NAAJ case anywhere hints the right side or both sides, so nothing exercises `|| hintToBroadcastRight(j.hint)`: dropping it, or turning the `||` into `&& !`, leaves every assertion in the three suites green.
Two assertions close it: a both-sides-hinted case in this block (right side `autoThresholdRight`, expecting `Some(BuildRight)`), and a left-hinted case in `LeftSemiAntiJoinPushDownSuite.scala:174-178` expecting no pushdown. That suite carries no hint at all today.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,12 +7452,24 @@ 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 would broadcast the right " +
Review Comment:
**LOW**
Two sentences in this paragraph do not match the code. :7456-7457 promises the optimization whenever regular planning would broadcast the right side, while the code requires the right side to pass the automatic threshold. The fallback does not consult size: with a BROADCAST hint on the right, or with no hint and both sides over the automatic threshold, it builds the right side, while the floor rejects once the right side is over the automatic threshold and the dedicated threshold does not admit it on its own.
:7465-7466 runs the other way: it makes "the right side is ineligible for automatic broadcasting" the condition that has to accompany a dedicated threshold of zero, but with zero and a left-only BROADCAST hint the optimization is off while the right side is eligible.
Since the paragraph is being rewritten anyway, all three can go in one pass: narrow the first to the automatic threshold, and add "or carries a broadcast hint only on the left" to the two that follow.
##########
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:
**LOW**
The comment at :441-444 explains the hint term purely in join-selection terms, but `canPlanAsBroadcastHashJoin` has a second caller, `PushDownLeftSemiAntiJoin.scala:68`, which takes the same answer to decide whether a LeftSemi/LeftAnti join can move below an `Aggregate`. That answer is correct (with a left-only hint the plan really does become a BNLJ, and declining the pushdown is right), but nothing in the comment points at the second caller.
Half a sentence — that this term also decides the aggregate pushdown — would cover it.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,161 @@ 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)
- val largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
+ 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 = 20 * 1024 * 1024,
+ size = Some(20 * 1024 * 1024))
+ 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(autoThresholdRight).copy(
+ hint = JoinHint(hintBroadcast, None)), SQLConf.get).isEmpty)
+ 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))
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(largeRight).copy(
+ hint = JoinHint(hintBroadcast, None)), SQLConf.get) === Some(BuildRight))
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "5MB") {
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(betweenThresholdsRight), 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(nullAwareAntiJoin(), SQLConf.get).isEmpty)
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoin(emptyRight), SQLConf.get) === Some(BuildRight))
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get).isEmpty)
+ }
+ }
+
+ test("NAAJ broadcast threshold is unlimited by default") {
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 -> "-1") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) === Some(BuildRight))
+ nullAwareAntiJoin(overLongMaxRight), SQLConf.get) === Some(BuildRight))
+ }
+ }
+
+ test("NAAJ broadcast threshold uses the adaptive threshold for runtime statistics") {
+ case class RuntimeStatsPlan(size: BigInt) extends LeafNode {
+ override def output: Seq[Attribute] = right.output
+ override def computeStats(): Statistics = Statistics(sizeInBytes = size, isRuntime = true)
+ }
+ val runtimeRight = RuntimeStatsPlan(5 * 1024 * 1024)
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(runtimeRight), SQLConf.get).isEmpty)
+ }
+
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1MB",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight))
+ nullAwareAntiJoin(runtimeRight), SQLConf.get) === Some(BuildRight))
}
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight))
+ nullAwareAntiJoin(runtimeRight), SQLConf.get) === Some(BuildRight))
}
+ }
- withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get).isEmpty)
+ test("NAAJ broadcast threshold does not read stats when eligibility is already known") {
+ case class ThrowingStatsPlan() extends LeafNode {
+ override def output: Seq[Attribute] = right.output
+ override def computeStats(): Statistics =
+ throw new IllegalStateException("statistics should not be read")
}
+ val nullAwareAntiJoinWithoutStats = nullAwareAntiJoin(ThrowingStatsPlan())
withSQLConf(
- SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
- SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get).isEmpty)
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
+ assert(getBroadcastHashJoinBuildSide(
+ nullAwareAntiJoinWithoutStats, SQLConf.get) === Some(BuildRight))
}
- 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.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoinWithoutStats, 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 -> "0") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get).isEmpty)
+ nullAwareAntiJoinWithoutStats.copy(
+ hint = JoinHint(hintBroadcast, None)), SQLConf.get).isEmpty)
+ }
+ }
+
+ test("NAAJ broadcast threshold rejects unknown sizes and respects the optimization flag") {
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "10MB") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = negativeSizeRight), SQLConf.get).isEmpty)
+ nullAwareAntiJoin(right.copy(size = Some(-1))), SQLConf.get).isEmpty)
Review Comment:
**LOW**
The block on master that asserted `isEmpty` for a 20000000-byte right side under a 10MB dedicated threshold is gone. All three places in this suite where the dedicated threshold is positive now assert admission, and the only `isEmpty` left, at :351, is the `size = Some(-1)` case, which covers `rightSize >= 0` alone: cutting `rightSize >= 0 && rightSize <= dedicatedThreshold` down to `rightSize >= 0` upsets neither this suite nor `LeftSemiAntiJoinPushDownSuite`.
The upper bound is still guarded, but only by the end-to-end case at `JoinSuite.scala:1285`, so a catalyst-only run cannot see it break. The old assertion still holds under the new code (20000000 bytes exceeds both the 10MB dedicated threshold and the default 10MB automatic one), so putting it back as it was is the cheapest fix.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org