LuciferYang commented on code in PR #58631:
URL: https://github.com/apache/spark/pull/58631#discussion_r3995581911
##########
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:
Both `optimizeNullAwareAntiJoin=false` and setting this new config to -1
disable the same optimization and land on the same plan, yet neither doc
mentions the other. And the new threshold is completely inert when the flag is
off (the conf check inside `unapply` short-circuits first), so a user can
configure a value that does nothing and get no hint of it.
The pair right next to it in the same file shows the convention:
`spark.sql.bucketing.coalesceBucketsInJoin.enabled` and `...maxBucketRatio`
(:7333 / :7346) share a prefix, and the latter's doc states it only takes
effect when the former is true. This config is still `.internal()` and
unreleased, so renaming it to
`spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold` and adding that
sentence are both cheap right now. Non-blocking.
##########
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:
The NAAJ case sits before the generic join case and, once matched, returns
`BroadcastHashJoinExec(BuildRight)` without consulting hints or calling
`checkHintNonEquiJoin`; its guard reads only `j.right.stats.sizeInBytes`. The
size-based fallback is in your description, but the hint part is not: when both
sides are over `autoBroadcastJoinThreshold`, v4.2.0 could still broadcast the
left via `BROADCAST(l)` on an explicit anti join, whereas this PR ignores that
hint and broadcasts the multi-GB right side, risking a driver OOM or the
`spark.sql.maxBroadcastTableSize` limit once it is big enough. (In the `NOT IN`
spelling an outer-query hint is dropped before the join exists, and a subquery
hint attaches only as a right-side hint, which cannot ask for BuildLeft.)
This is not newly introduced: `v4.0.0:joins.scala:422` and
`v4.1.0:joins.scala:430` are `=> true`, and v3.2.0 has no such site at all, so
3.2 through 4.1 ignored hints too; only 4.2.0 (#55678) differs. Please don't
add hint checks to the guard: that is the reverted
`NullAwareAntiJoinPlanning.decide`, and since the same guard is the
aggregate-pushdown predicate it would re-couple pushdown to hints. Saying so in
the description and the config doc is enough. Non-blocking.
##########
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:
The doc only says the fallback "may still use a broadcast nested loop join",
which reads as if lowering this config can stop a large right side from being
broadcast. With no broadcast hint there is nothing else to pick for this shape:
the condition is non-equi so no SMJ/SHJ is available,
`canBuildBroadcastLeft(LeftAnti)` is false, and when neither side fits, the
final fallback branch in `createJoinWithoutHint` broadcasts the right side
anyway as an `Array[InternalRow]` rather than a compact `HashedRelation`, plus
O(M*N) matching. With a 5GB left and right and the default 10MB
`autoBroadcastJoinThreshold`, setting it to 100m costs more memory, not less.
The PR description explains this; the config doc, which is usually all an
operator sees via `SET`, does not. Worth a clause in the doc too:
`canPlanAsBroadcastHashJoin` is the pushdown guard in
`PushDownLeftSemiAntiJoin` (:68) too, so this threshold also decides whether a
NAAJ gets pushed below an Aggregate. Non-blocking.
--
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]
