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: [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]
