LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4038770335
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,12 +7452,18 @@ 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. " +
+ "Regular planning uses spark.sql.adaptive.autoBroadcastJoinThreshold
for runtime " +
Review Comment:
**MEDIUM**
The adaptive sentence holds for join selection but cannot apply to the other
consumer. `Statistics(isRuntime = true)` originates only in
`QueryStageExec.computeStats` and reaches logical plans through
`LogicalQueryStage`, while `PushDownLeftSemiAntiJoin` is registered only in the
main optimizer's `operatorOptimizationRuleSet`
(`catalyst/optimizer/Optimizer.scala:109`) and in no `AQEOptimizer` batch.
Every plan that rule sees therefore reports estimated statistics and always
uses the static threshold. Read together with the last sentence, which ties the
same decision to the aggregate pushdown, the doc suggests
`spark.sql.adaptive.autoBroadcastJoinThreshold` can widen that pushdown. It
cannot; scoping the sentence to join selection would say what the code does.
Separately, dropping "If the estimated size exceeds the effective threshold,
Spark falls back to regular join planning" left the next sentence's "The
fallback" with no antecedent in the paragraph.
##########
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala:
##########
@@ -1308,7 +1308,7 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
- test("SPARK-36082: left-broadcast NAAJ fallback uses nested-loop join") {
+ test("SPARK-36082: automatic threshold enables NAAJ hash join despite left
broadcast hint") {
Review Comment:
**MEDIUM**
The hint in this query is now unobservable. `JoinSelection` matches the NAAJ
shape before any hint handling and hardcodes the build side
(`SparkStrategies.scala:343`), never reading `j.hint`, so deleting `/*+
BROADCAST(naajHintedLeft) */` leaves all four assertions passing. The positive
fact this case now records is already covered at unit level by the `auto =
10MB` plus dedicated `0` block in `JoinSelectionHelperSuite`.
What the old version pinned was a differential: the same relations and query
giving `BuildLeft` because of the hint, with the no-hint case below it as the
control giving `BuildRight`. Nothing now covers a broadcast hint naming the
left side while the NAAJ hash join is rejected. This was the only `BROADCAST`
hint on a NAAJ query in the repo, and `JoinSelectionHelperSuite`'s NAAJ fixture
hardcodes `JoinHint.NONE`. The path stays reachable, for instance `auto = 0`
with dedicated `0`. One more `withSQLConf` block here would restore it, and
would also be the only case asserting rows through a `BuildLeft` anti
nested-loop join, since the case above expects an empty answer.
##########
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
&&
Review Comment:
**LOW**
`automaticBroadcastDisabled` compares with `<= 0` while `canBroadcastBySize`
admits `size >= 0 && size <= threshold`, so at threshold `0` with an estimated
size of `0` the two disagree: this short-circuit returns false where the
disjunct below it would have returned true.
Reachable with `spark.sql.autoBroadcastJoinThreshold=0`, the adaptive
threshold unset, dedicated `0`, and `SELECT * FROM t WHERE c NOT IN (SELECT id
FROM range(0))`. `Range.computeStats` reports `sizeInBytes = 0` for zero
elements, and `PropagateEmptyRelation` does not remove the join, its `isEmpty`
matching only an empty `LocalRelation`.
The consequence is negligible, since both plans are cheap over an empty
right side, but it contradicts this branch's own rationale and the new doc
sentence about regular planning considering the right side broadcastable. The
same corner lets the adaptive threshold change the decision for a plan with no
runtime statistics. Comparing `< 0` in both halves removes it, and the
short-circuit case uses -1 and -2, so it stays green.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,125 @@ 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))
+
+ 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))
+ }
+
+ 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)
+ }
+ }
+
+ 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))
- assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) ===
Some(BuildRight))
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), 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.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
+ withSQLConf(
+ SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1MB",
Review Comment:
**MEDIUM**
Both blocks of this case use a runtime-statistics right side, and these two
lines are the only places in the suite that set
`spark.sql.adaptive.autoBroadcastJoinThreshold`, each alongside a positive
static threshold. Two mutations survive that.
Drop the `plan.stats.isRuntime` test in `canBroadcastBySize`, so the
adaptive threshold applies whenever it is set, and both blocks still pass (5MB
> 1MB, then 5MB <= 10MB). A block with `auto = 10MB`, `adaptive = 1MB`,
dedicated `0` and an ordinary non-runtime 5MB right side, asserting
`Some(BuildRight)`, would catch it.
Delete `&& conf.getConf(ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_ <=
0)` from `automaticBroadcastDisabled` and the whole suite stays green: every
block that reaches that gate either has a positive static threshold or leaves
the adaptive one unset. A block with `auto = -1`, `adaptive = 10MB`, dedicated
`0` and the runtime 5MB right side, asserting `Some(BuildRight)`, would pin the
new conjunct.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala:
##########
@@ -142,6 +142,33 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}
+ test("Aggregate: NAAJ pushdown follows the effective broadcast threshold") {
Review Comment:
**LOW**
`testRelation` and `testRelation1` are empty `LocalRelation`s, so
`computeStats` gives `sizeInBytes = getSizePerRow(output) * data.length`, which
is 0 (`LocalRelation.scala:107`). `0 <= 10MB` then holds trivially, and the
first block pins only that automatic broadcasting is enabled, not any size
comparison: an implementation that admits whenever the automatic threshold is
positive, ignoring the size, passes it. The second block is sound, and the
analyzed plan really is a fixpoint of this suite's batch once the pushdown is
refused.
A third block with a `StatsTestPlan` right side of 20MB, `auto = 10MB` and
dedicated `0`, expecting no pushdown, would pin the size dimension too.
--
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]
