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/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.
The 20MB block added in `47fceb0e550` covers the rejection side of
`LeftSemiAntiJoinPushDownSuite`. The positive block at :172 is still an empty
`LocalRelation` whose `sizeInBytes` is 0, which clears any nonnegative
threshold, so it 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]
