LuciferYang commented on code in PR #58631:
URL: https://github.com/apache/spark/pull/58631#discussion_r3999216000
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7366,15 +7366,34 @@ object SQLConf {
val OPTIMIZE_NULL_AWARE_ANTI_JOIN =
buildConf("spark.sql.optimizeNullAwareAntiJoin")
.internal()
- .doc("When true, NULL-aware anti join execution will be planed into " +
+ .doc("When true, NULL-aware anti join execution can be planned as " +
"BroadcastHashJoinExec with flag isNullAwareAntiJoin enabled, " +
"optimized from O(M*N) calculation into O(M) calculation " +
"using Hash lookup instead of Looping lookup. " +
- "Only support for singleColumn NAAJ for now.")
+ "Only support for singleColumn NAAJ for now. The optimization is also
controlled by " +
+ "spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold.")
.version("3.1.0")
.booleanConf
.createWithDefault(true)
+ val NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD =
+ buildConf("spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold")
Review Comment:
Negative meaning unlimited inverts the broadcast-threshold family:
`autoBroadcastJoinThreshold` and its adaptive twin both document `-1` as the
way to disable broadcasting, and `canBroadcastBySize` (`joins.scala:370`)
refuses any negative size. An operator whose driver is dying on a NAAJ
broadcast will set `-1` and get the size limit removed instead. `bytesConf`
strips the sign, so `-2g` lands in the same bucket.
`createOptional` avoids the collision: unset means no limit, a set value is
a real limit, and disabling stays with `spark.sql.optimizeNullAwareAntiJoin`.
One reader to update, plus the `-2` assertion in `JoinSelectionHelperSuite`. If
the current semantics stay, the doc should say they read the opposite way from
`autoBroadcastJoinThreshold`.
##########
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala:
##########
@@ -1410,6 +1365,50 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
}
}
+ test("SPARK-36082: NAAJ hash join preserves floating-point equality") {
+ Seq(false, true).foreach { adaptiveEnabled =>
Review Comment:
The plan assertion reads `queryExecution.sparkPlan`, the plan before AQE
wraps it, so both AQE arms inspect the same tree. If the NAAJ ever falls back
to a BNLJ under AQE, all four combinations still pass, since the BNLJ already
compares `-0.0` correctly. Asserting after `checkAnswer` and collecting through
`AdaptiveSparkPlanHelper` would cover it; `executedPlan` is still `initialPlan`
until the query runs.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -498,9 +438,11 @@ trait JoinSelectionHelper extends Logging {
getBroadcastBuildSide(join, hintOnly = true, conf).orElse {
if (noShufflePlannedBefore) getBroadcastBuildSide(join, hintOnly =
false, conf) else None
}
- case j if ExtractSingleColumnNullAwareAntiJoin.extract(j).isDefined =>
- if (NullAwareAntiJoinPlanning.decide(j, conf) ==
- NullAwareAntiJoinPlanning.BroadcastHash) {
+ // `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.
+ case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) =>
+ val threshold = conf.nullAwareAntiJoinBroadcastThreshold
+ if (threshold < 0 || (threshold > 0 && j.right.stats.sizeInBytes <=
threshold)) {
Review Comment:
The rewrite dropped `sizeInBytes >= 0`, which `canBroadcastBySize` in the
same file (:370) still carries. Spark's own estimation never goes negative, but
a DSv2 connector can report a negative size since `transformV2Stats` does not
clamp it, and then a positive threshold here would broadcast a right side that
:370 would refuse. Worth adding back to the size comparison.
--
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]
