sunchao commented on code in PR #58631:
URL: https://github.com/apache/spark/pull/58631#discussion_r3993604221
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -340,18 +340,10 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
.getOrElse(createJoinWithoutHint())
}
- case j: logical.Join if
ExtractSingleColumnNullAwareAntiJoin.extract(j).isDefined =>
- val (leftKeys, rightKeys) =
ExtractSingleColumnNullAwareAntiJoin.extract(j).get
- NullAwareAntiJoinPlanning.decide(j, conf) match {
- case NullAwareAntiJoinPlanning.BroadcastHash =>
- Seq(joins.BroadcastHashJoinExec(leftKeys, rightKeys, LeftAnti,
BuildRight,
- None, planLater(j.left), planLater(j.right), isNullAwareAntiJoin
= true))
- case NullAwareAntiJoinPlanning.BroadcastNestedLoop =>
- checkHintNonEquiJoin(j.hint)
- val buildSide = getBroadcastNestedLoopJoinBuildSide(j, conf)
- Seq(joins.BroadcastNestedLoopJoinExec(
- planLater(j.left), planLater(j.right), buildSide, LeftAnti,
j.condition))
- }
+ case j @ ExtractSingleColumnNullAwareAntiJoin(leftKeys, rightKeys)
Review Comment:
### [P2] Restored hash joins can return incorrect floating-point results
Setting `spark.sql.autoBroadcastJoinThreshold=0` now selects the null-aware
hash join.
For nullable double keys:
- Left: `[-0.0, 2.0, NULL]`
- Right: `[0.0, 1.0]`
- Parent `NOT IN` result: `[2.0]`
- PR result: **`[-0.0, 2.0]`**
Reproduced with adaptive execution and code generation independently enabled
and disabled. This re-exposes an existing hash-key normalization defect in
previously correct fallback paths. Normalize the extracted keys and retain the
deleted floating-point regression tests.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -498,13 +438,12 @@ 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) {
- Some(BuildRight)
- } else {
- None
- }
+ // `JoinSelection` always builds from the right for this shape. Its
dedicated threshold
+ // defaults to Long.MaxValue to preserve the original NAAJ planning
behavior.
+ case j @ ExtractSingleColumnNullAwareAntiJoin(_, _)
+ if j.right.stats.sizeInBytes >= 0 &&
+ j.right.stats.sizeInBytes <=
conf.nullAwareAntiJoinBroadcastThreshold =>
Review Comment:
### [P2] The default threshold does not fully restore original planning
Statistics use `BigInt`, so they can exceed `Long.MaxValue`.
A `UNION ALL` of two tiny RDD-backed inputs without statistics produces an
estimated size of `18446744073709551614`. I confirmed that the default still
selects the slow nested-loop join, and no representable threshold enables
hashing.
This is an **incomplete restoration**, rather than a new parent-to-head
regression. An explicit unlimited sentinel would preserve the original behavior.
--
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]
