cloud-fan commented on code in PR #58631:
URL: https://github.com/apache/spark/pull/58631#discussion_r4000377527
##########
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:
Thanks for raising this. I prefer to keep the current `-1` semantics. This
threshold is not intended to follow `spark.sql.autoBroadcastJoinThreshold`:
disabling the NAAJ optimization is already controlled by
`spark.sql.optimizeNullAwareAntiJoin=false`, while this threshold needs an
unbounded default to restore the original behavior. `Long.MaxValue` is not
sufficient because `sizeInBytes` is a `BigInt`; the `threshold < 0` branch is
genuinely unbounded and covers estimates above `Long.MaxValue`.
`createOptional` would encode the same unbounded state as `None`, but would not
improve correctness here. The config is internal and its documentation
explicitly calls out the negative-value semantics, so I prefer to keep `-1`.
##########
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:
Fixed in 36a284541db. The test now calls `checkAnswer` first to execute the
query, then collects from `queryExecution.executedPlan` through
`AdaptiveSparkPlanHelper`, so the AQE-enabled cases verify that the final
adaptive plan actually uses the null-aware broadcast hash join.
##########
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:
Fixed in 36a284541db. The positive-threshold path now also requires
`rightSize >= 0`, and `JoinSelectionHelperSuite` covers a negative reported
size.
--
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]
