cloud-fan commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4036554337
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,19 @@ 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 = dedicatedThreshold < 0 || {
Review Comment:
Good catch. Commit `0f5f063d366` updates this regression case to record the
intended floored behavior: when the automatic threshold admits the right side,
the specialized null-aware `BroadcastHashJoinExec` wins, ignores the left
broadcast hint, and builds right. The test now asserts no BNLJ, one null-aware
BHJ, `BuildRight`, and the unchanged answer. All four focused SPARK-36082 tests
pass.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- val overLongMaxRight = right.copy(
- rowCount = BigInt(Long.MaxValue) + 1,
- size = Some(BigInt(Long.MaxValue) + 1))
withSQLConf(
Review Comment:
Agreed. Commit `0f5f063d366` adds a default-semantics case that leaves the
dedicated configuration unset, disables automatic broadcasting, gives the right
side a size above `Long.MaxValue`, and asserts `Some(BuildRight)`. This
directly pins the default `-1` as unbounded.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- 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 -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get)
=== Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(autoThresholdRight), SQLConf.get) ===
Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
}
- 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.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "20MB") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
}
- withSQLConf(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 -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(),
SQLConf.get).isEmpty)
}
+ }
- withSQLConf(
- SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
- SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ test("NAAJ broadcast threshold short-circuits config-only decisions") {
+ case class ThrowingStatsPlan() extends LeafNode {
Review Comment:
Agreed. `ThrowingStatsPlan.computeStats()` now explicitly throws
`IllegalStateException` in commit `0f5f063d366`, so the short-circuit
assertions no longer depend on the current default implementation of
`LeafNode.computeStats`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:
##########
@@ -195,48 +200,73 @@ 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 largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
- val negativeSizeRight = right.copy(size = Some(-1))
- 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 -> "10MB",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get)
=== Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(autoThresholdRight), SQLConf.get) ===
Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
}
- 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.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "20MB") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) ===
Some(BuildRight))
+ nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
}
- withSQLConf(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 -> "-1",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
+ assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(),
SQLConf.get).isEmpty)
}
+ }
- withSQLConf(
- SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
- SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
- assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin,
SQLConf.get).isEmpty)
+ test("NAAJ broadcast threshold short-circuits config-only decisions") {
+ case class ThrowingStatsPlan() extends LeafNode {
+ override def output: Seq[Attribute] = right.output
}
+ val nullAwareAntiJoinWithoutStats = nullAwareAntiJoin(ThrowingStatsPlan())
- 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.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
assert(getBroadcastHashJoinBuildSide(
- nullAwareAntiJoin.copy(right = largeRight), SQLConf.get).isEmpty)
+ nullAwareAntiJoinWithoutStats, 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)
+ }
+ }
+
+ test("NAAJ broadcast threshold rejects unknown sizes and respects the
optimization flag") {
+ withSQLConf(
Review Comment:
Agreed. Commit `0f5f063d366` adds both missing cases: a positive 10 MB
dedicated threshold admits the normal right side while automatic broadcasting
is disabled, and an 8 MB right side is admitted when the dedicated threshold is
5 MB and the automatic threshold is 10 MB. The latter specifically pins the
larger-of-the-two behavior.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,19 @@ 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 = dedicatedThreshold < 0 || {
+ val effectiveThreshold = math.max(dedicatedThreshold,
conf.autoBroadcastJoinThreshold)
Review Comment:
Agreed. Commit `0f5f063d366` uses `canBroadcastBySize` for the automatic
side of the decision, so runtime statistics select
`spark.sql.adaptive.autoBroadcastJoinThreshold` when configured. The positive
dedicated threshold remains a separate size gate. Explicit checks preserve the
no-statistics paths for a negative dedicated threshold and for fully disabled
thresholds. New runtime-statistics cases cover both directions: static 10
MB/adaptive 1 MB rejects a 5 MB right side, while static 1 MB/adaptive 10 MB
admits it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -438,12 +438,19 @@ 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 = dedicatedThreshold < 0 || {
+ val effectiveThreshold = math.max(dedicatedThreshold,
conf.autoBroadcastJoinThreshold)
+ effectiveThreshold > 0 && {
+ val rightSize = j.right.stats.sizeInBytes
+ rightSize >= 0 && rightSize <= effectiveThreshold
+ }
+ }
+ if (canBroadcast) {
Review Comment:
The widening is intentional. The dedicated NAAJ threshold already feeds
`canPlanAsBroadcastHashJoin`, which is the existing aggregate-pushdown gate, so
the floored eligibility should apply consistently to both join selection and
pushdown. Commit `0f5f063d366` keeps the shared predicate, clarifies that
relationship in the config documentation, and adds a
`LeftSemiAntiJoinPushDownSuite` case showing that the floor enables pushdown
while disabled thresholds do not. I did not add a separate unfloored predicate.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7452,8 +7452,11 @@ 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 effective " +
Review Comment:
Updated in commit `0f5f063d366`. The config documentation now names
`spark.sql.optimizeNullAwareAntiJoin=false` as the way to disable the
specialized optimization without changing automatic thresholds, and explains
the adaptive runtime threshold. The PR description now says to use dedicated
threshold `0` plus disabled applicable automatic thresholds; it explicitly
notes that a negative dedicated threshold is unbounded. One nuance: leaving the
dedicated config unset means its `-1` unbounded default, so a lower nonnegative
value remains distinguishable for inputs above the automatic threshold.
--
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]
