cloud-fan commented on code in PR #58631:
URL: https://github.com/apache/spark/pull/58631#discussion_r3996658403
##########
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:
Fixed in 0b7c77bc06d. `NormalizeFloatingNumbers` now recognizes
`ExtractSingleColumnNullAwareAntiJoin` and normalizes both extracted keys. I
restored the two deleted floating-point fallback regression tests and added
broadcast hash join coverage with AQE and whole-stage codegen independently
enabled and disabled. The focused suites pass.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7375,6 +7375,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD =
+ buildConf("spark.sql.nullAwareAntiJoinBroadcastThreshold")
Review Comment:
Done. The config was renamed to
`spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold` in 0501e87536e, and
both config docs describe its dependency on
`spark.sql.optimizeNullAwareAntiJoin`. Following the P2 feedback above,
negative values now mean unlimited and zero disables the optimization.
##########
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:
Fixed in 0b7c77bc06d. The threshold now defaults to `-1`; any negative value
means unlimited, zero disables the hash optimization, and a positive value is
the size limit. The comparison stays in `BigInt`, and the test covers
statistics greater than `Long.MaxValue`.
##########
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:
Documented this in both the config doc and the updated PR description. The
specialized NAAJ hash path continues to ignore join hints and always builds the
right side, matching the behavior before #55678. No hint logic was added to the
shared guard.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7375,6 +7375,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD =
+ buildConf("spark.sql.nullAwareAntiJoinBroadcastThreshold")
+ .internal()
+ .doc("Configures the maximum estimated size in bytes of the right side
of a " +
Review Comment:
Updated the config doc in 0b7c77bc06d. It now explains that the fallback may
broadcast the right side using a nested-loop representation that consumes more
memory and runs in `O(M * N)` time, and that the threshold also controls
aggregate pushdown.
--
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]
