LuciferYang commented on code in PR #58870:
URL: https://github.com/apache/spark/pull/58870#discussion_r4038992967
##########
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:
Correcting the first half of this: dropping the `plan.stats.isRuntime` test
survives this suite but not the repo. `AdaptiveQueryExecSuite`'s `Change
broadcast join to merge join` sets `spark.sql.autoBroadcastJoinThreshold=10000`
with the adaptive threshold at `-1` and asserts one top-level broadcast hash
join in the initial physical plan, which `runAdaptiveAndVerifyResult` hands
back as `sparkPlan`, so estimated statistics reading the adaptive `-1` would
take that join away. That gate is also not this PR's code.
So the ask reduces to the second half: nothing pins the new `forall(_ <= 0)`
conjunct, and a block with `auto = -1`, `adaptive = 10MB`, dedicated `0` and
the runtime 5MB right side would.
--
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]
