cloud-fan commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4047422069
##########
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:
Agreed. Commit `47fceb0e550` keeps the automatically admitted case and adds
a rejected case using the same left broadcast hint with automatic threshold
`-1` and dedicated threshold `0`. It asserts one `BroadcastNestedLoopJoinExec`
with `BuildLeft`, no null-aware hash join, and the non-empty `Row(2.0d)` result.
##########
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:
Agreed. Commit `47fceb0e550` changes both disabled-threshold comparisons
from `<= 0` to `< 0`, matching `canBroadcastBySize` at the
zero-size/zero-threshold boundary. It also adds regression assertions that a
zero-byte right side is admitted at threshold `0` while a non-empty side is
rejected.
##########
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:
Agreed. Commit `47fceb0e550` scopes the adaptive-threshold text to join
selection, explains that aggregate pushdown runs before adaptive execution and
therefore uses estimated statistics plus
`spark.sql.autoBroadcastJoinThreshold`, and restores an explicit
regular-planning fallback sentence as the antecedent for "The fallback."
##########
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:
Agreed. Commit `47fceb0e550` adds a `StatsTestPlan` right side of `20MB`
with automatic threshold `10MB` and dedicated threshold `0`, and asserts that
the join remains above the aggregate. This covers the size-rejection dimension
in the pushdown suite.
##########
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:
Agreed on the remaining adaptive-only coverage gap, and thanks for the
correction on the first half. Commit `47fceb0e550` adds the runtime-statistics
case with automatic threshold `-1`, adaptive threshold `10MB`, and dedicated
threshold `0`, asserting `Some(BuildRight)`. This pins the new
adaptive-threshold conjunct in `automaticBroadcastDisabled`.
--
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]
