dongjoon-hyun commented on PR #58870:
URL: https://github.com/apache/spark/pull/58870#issuecomment-5779406953
Thanks for the follow-up. I went through this in detail; notes below,
grouped by severity. Line numbers are against `64b6d905ec4`.
## Correctness / design
**1. The optimizer and AQE can now disagree, and the Aggregate pushdown is
not reversible** — `joins.scala:463`
Routing the decision through `canBroadcastBySize` makes it depend on
`stats.isRuntime`, and the same predicate gates two things at two different
times:
- `PushDownLeftSemiAntiJoin.scala:68` — logical optimizer, estimated stats,
`isRuntime = false`, so `spark.sql.autoBroadcastJoinThreshold`;
- `JoinSelection` under AQE — `LogicalQueryStage` runtime stats, `isRuntime
= true`, so `spark.sql.adaptive.autoBroadcastJoinThreshold`.
With `broadcastThreshold=0`, `autoBroadcastJoinThreshold=10MB`, and a right
side estimated at 5MB but 50MB at runtime (or with the adaptive threshold set
lower), the optimizer pushes the NAAJ below the `Aggregate` — `pushDownJoin`
makes the join's left input `agg.child`, i.e. pre-aggregation rows — and then
`reOptimize` re-runs `JoinSelection`, rejects the hash NAAJ on runtime stats,
and lands on `BroadcastNestedLoopJoinExec` over that enlarged left side.
`AQEOptimizer.defaultBatches` doesn't include `PushDownLeftSemiAntiJoin`, so it
can't be undone.
The window looks structural rather than incidental: `reOptimize`
(`AdaptiveSparkPlanExec.scala:407`) runs before `createQueryStages(...,
firstRun = false)`, so whenever the right side contains a shuffle there is an
iteration where the right child is a `LogicalQueryStage` over a materialized
*shuffle* stage. `LogicalQueryStageStrategy`'s null-aware case doesn't match
that, so the plan reaches `SparkStrategies.scala:343` with `isRuntime = true`.
On master this couldn't happen — the old predicate never read `isRuntime`,
so both call sites always agreed. The new doc describes the two-clock situation
in prose, but nothing guards it.
**2. The floor doesn't cover the case where the fallback broadcasts the
right side anyway** — `joins.scala:464`
For `LeftAnti`, `canBuildBroadcastLeft` is false, so `desiredBuildSide` is
`BuildRight`, and `createJoinWithoutHint`'s final `getOrElse`
(`SparkStrategies.scala:418-423`) forces a `BroadcastNestedLoopJoinExec` on the
right side with no size check at all.
So with no hints, `broadcastThreshold=0`, `autoBroadcastJoinThreshold=10MB`,
left 200GB and right 50MB: `canBroadcastBySize` is false, the hash NAAJ is
declined, `createBroadcastNLJoin(false)` returns `None`, cartesian product
doesn't apply to `LeftAnti` — and the same 50MB is broadcast anyway, as
`IdentityBroadcastMode` (every column, not a single-key `HashedRelation`), with
the probe going from O(M+N) to O(M*N).
This is pre-existing rather than introduced here, and the new doc text
acknowledges it. But it means the floor closes only the `dedicated < size <=
auto` band, while the configuration still cannot prevent the broadcast in the
case where it would hurt most — it can only make it worse. Worth deciding
explicitly whether that's the intended contract. There's also no test for the
unhinted `AUTO` + `dedicated=0` combination.
**3. `NO_BROADCAST_AND_REPLICATION` is honored at threshold 0 and ignored
above it** — `joins.scala:465`
`!hintToNotBroadcastAndReplicateRight(j.hint)` guards only the disjunct at
line 461. The `dedicatedThreshold > 0` disjunct at 465 bypasses it:
```scala
rightBroadcastSelectedByHintOrSize ||
(dedicatedThreshold > 0 && {
val rightSize = j.right.stats.sizeInBytes
rightSize >= 0 && rightSize <= dedicatedThreshold
})
```
With `broadcastThreshold=20MB` and a 5MB right side carrying the hint, line
461 is false but 465 is true, so a `BroadcastHashJoinExec` broadcasts and
replicates exactly the side the hint forbids. Set the threshold to 0 and the
same plan is rejected. That contradicts the comment added at line 444 ("a
right-only no-broadcast-and-replication hint makes the fallback build the left
side instead").
Symmetrically, `getBroadcastNestedLoopJoinBuildSide` returns
`Some(BuildRight)` when the *left* side carries the hint, which the predicate
can't see either.
Latent today — `hintAliases` is `Set.empty` and the only producer
(`RewriteMergeIntoTable.scala:427`) sets it on the left of an inner join — so
this is a trap rather than a live bug, but the asymmetry reads as an oversight.
**4. For hint-free plans the whole predicate is `max(auto, dedicated)`** —
`joins.scala:448`
Expanding lines 448-470 with no hints gives `(rightSize <= auto) || (D > 0
&& 0 <= rightSize <= D)`, i.e. `rightSize <= max(auto, D)` for `D > 0` and
`rightSize <= auto` for `D == 0` — differing only at the degenerate point `auto
< 0, D == 0, rightSize == 0`.
So the 23 lines of nested branching and the 243-word doc exist entirely to
buy hint sensitivity, and per (2) and (3) that hint sensitivity is itself
incomplete. Given this is the fourth commit rewriting the same predicate
(7090c70cfef → f8b673c5a53 → 6456b573343 → this one), each time changing the
documented contract — the immediately preceding one said "Join hints do not
override this configuration", which this PR reverses — and `.version("4.2.1")`
is still unreleased, it may be worth settling it now: either state the rule as
`effective = max(auto, dedicated)` and drop the hint terms, or delegate to a
helper shared with `createBroadcastNLJoin` /
`getBroadcastNestedLoopJoinBuildSide` so the two copies of the build-side rule
can't drift.
**5. The method scaladoc now contradicts the branch it governs** —
`joins.scala:419`
Lines 419-420 still say `None` is returned "when one is ruled out by the
join shape or by a hint", and 425-426 document the precedence as "a hinted
broadcast first, a hinted shuffle hash join as a veto, then the sizes". After
this change the NAAJ branch returns `None` purely from sizes and thresholds
with no hint involved, and treats `BROADCAST(left)` as a veto of a right-side
broadcast — which appears nowhere in that order. The inline comment 10 lines
below was updated but the scaladoc wasn't.
This isn't cosmetic: `PushDownJoinThroughUnion.scala:113` explicitly cites
this contract when justifying that `getBroadcastHashJoinBuildSide` alone
"carries the requirement the removed `canPlanAsBroadcastHashJoin` conjunct used
to". Someone widening that rule's `Inner || LeftOuter` guard to `LeftAnti`
would read "None only by shape or hint" and inherit a right-side-duplicating
rewrite whose correctness now turns on two broadcast thresholds.
**6. The config doc's first sentence contradicts the new semantics** —
`SQLConf.scala:7451`
"Configures the maximum estimated size in bytes ... for which Spark uses the
broadcast hash join optimization" is no longer true in either direction:
- with `autoBroadcastJoinThreshold=30MB` and `broadcastThreshold=20MB`, a
25MB right side is admitted — any dedicated value below the automatic threshold
is inert;
- a user who sets the threshold to 1MB to cap the NAAJ `HashedRelation`
build gets a 10GB right side built anyway as soon as the query carries `/*+
BROADCAST(right) */`, since `rightBroadcastHint` short-circuits before any size
check.
Neither is covered: `JoinSelectionHelperSuite.scala:246-257` checks
`overDedicatedThresholdRight` without hints, and applies `hintBroadcast` only
to the left.
## Tests
**7.** `JoinSuite.scala:1311` — the rewritten test hand-rolls `plan.collect`
blocks instead of using `assertJoin` (`JoinSuite.scala:64`), which asserts
`canPlanAsBroadcastHashJoin(optimized, conf) ===
operators.head.isInstanceOf[BroadcastHashJoinExec]` ("not in sync with join
selection codepath!") and, for a BHJ, that
`getBroadcastHashJoinBuildSide(...).contains(bhj.buildSide)`. That is exactly
the invariant this PR changes, and it's the one assertion the new test drops.
The SPARK-32290 NAAJ test at line 1249 already uses this pattern. It would also
collapse three near-identical 14-line blocks (18 prefix-distinguished locals)
to about three lines each.
**8.** `LeftSemiAntiJoinPushDownSuite.scala:146` — this is the only NAAJ
test in the repo that doesn't pin `SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN`.
`ExtractSingleColumnNullAwareAntiJoin.unapply` reads
`SQLConf.get.optimizeNullAwareAntiJoin` — the session conf, not the one passed
in. If that flag is ever false, the extractor stops matching, and the two
`comparePlans(Optimize.execute(x), x)` negatives at lines 189 and 196 still
pass while only the positive fails, so the failure points at the threshold
logic rather than at the flag.
**9.** `JoinSelectionHelperSuite.scala:246` — two changed behaviors have no
assertion: a right `BROADCAST` hint overriding a positive dedicated threshold
(the bypass in (6)), and the `BigInt` size vs `Long` threshold comparison. The
surviving `overLongMaxRight` case at 281-292 runs only with `dedicatedThreshold
< 0`, which returns `true` at line 449 before any comparison happens — so a
"simplification" of line 467 to `rightSize.toLong <= dedicatedThreshold` would
truncate `Long.MaxValue + 1` to `Long.MinValue` and broadcast a 9-exabyte right
side, uncaught. Both are one assertion each.
**10.** `JoinSelectionHelperSuite.scala:295` — `RuntimeStatsPlan` hardcodes
`isRuntime = true`, which no leaf can be in production: `Statistics(isRuntime =
true)` comes only from `QueryStageExec.computeStats` and reaches a logical plan
only via `LogicalQueryStage.computeStats`, which falls back to `isRuntime =
false` while the stage is unmaterialized. So the ordering that matters under
AQE — the first planning pass, where `canBroadcastBySize` applies the
*non-adaptive* threshold — is untested, and the end-to-end `JoinSuite` test
runs entirely under `ADAPTIVE_EXECUTION_ENABLED = false`. Adding `isRuntime:
Boolean = false` to `StatsTestPlan` would fix the modeling gap and make it
reusable — it's used in 177 places and none can currently test runtime stats.
**11.** `JoinSelectionHelperSuite.scala:328` — the `ThrowingStatsPlan` test
pins the `&&`/`||` evaluation *order* rather than behavior. Two of its
assertions pass only because `!hintToBroadcastLeft(j.hint)` is written before
`canBroadcastBySize(...)` on line 463, and because `rightBroadcastHint ||`
precedes it on 462. Rewriting 463 as `canBroadcastBySize(j.right, conf) &&
!hintToBroadcastLeft(j.hint)` — semantically identical — fails the suite with
`IllegalStateException: statistics should not be read`.
## Cleanup
**12.** `joins.scala:454` — the `dedicatedThreshold == 0 &&
automaticBroadcastDisabled && !rightBroadcastHint` branch appears to be a
behavioral no-op. Under that guard `canBroadcastBySize` picks
`ADAPTIVE.getOrElse(auto)` or `auto`, both negative by the guard, so
`sizeInBytes >= 0 && sizeInBytes <= threshold` is unsatisfiable; and
`dedicatedThreshold > 0` is false. The `else` branch yields `false`
identically. It only avoids one memoized `stats` read, at the cost of lines
452-453 being a second copy of the adaptive-vs-static selection rule that
already lives in `canBroadcastBySize` (364-369) — if that ever changes, this
guard drifts with no compile error. It also never fires on a default cluster,
since `AUTO_BROADCASTJOIN_THRESHOLD` ships as `"10MB"`, so the "avoid
potentially expensive statistics computation" comment doesn't hold for the
configuration this PR is about.
**13.** `joins.scala:452` — `val` is eager, so
`conf.autoBroadcastJoinThreshold` is read on every invocation with a
nonnegative threshold although it's consumed only by the `dedicatedThreshold ==
0` conjunct on 454. That read goes through
`ConfigEntryWithDefaultString.readFrom`, which unconditionally runs
`byteFromString` → `JavaUtils.byteStringAs`: a
`toLowerCase(Locale.ROOT).trim()` plus two `Matcher` allocations and a regex
match, even when unset. With `dedicatedThreshold == 0` and auto broadcasting on
it happens twice per call (452, and again inside `canBroadcastBySize` at 368);
master did zero. Inlining both conf reads as the trailing conjuncts of the `if`
at 454 lets `&&` short-circuit them, without the `LazyRef` a `lazy val` would
allocate.
**14.** `SQLConf.scala:7455` — the doc hardcodes four config key literals
where the file's convention is `${ENTRY.key}` interpolation (98 occurrences,
including `ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD` itself at line 1527
referencing `${AUTO_BROADCASTJOIN_THRESHOLD.key}`). All three entries are
defined earlier, so `.doc(s"...")` compiles as is, and a key rename becomes a
compile error rather than silently wrong docs. On length: across the ~800
`.doc` strings in this file the median is 29 words and p99 is 135; this one is
243, the longest in the file, on an `internal()` config that never reaches the
published docs. There's also a stray `"uses " +` fragment alone on line 7460 —
the split is needed for the 100-char limit, but could sit at a phrase boundary.
**15.** `JoinSuite.scala:1311` — renaming the test from `SPARK-36082:` to
`SPARK-59673:` drops the regression marker from a test that still contains the
SPARK-36082 guard: the first `withSQLConf` block (1327-1357:
`AUTO=Long.MaxValue`, `dedicated=0`, left `BROADCAST` hint, asserting
`BroadcastNestedLoopJoinExec` with `buildSide === BuildLeft`) is that guard
verbatim. `grep -c SPARK-36082 JoinSuite.scala` gives 4 on master and 3 here,
so the PR description's `-z SPARK-36082` command and its "4 matching
SPARK-36082" count no longer reproduce — and the tests this PR actually changed
aren't selected by that filter at all. Keeping both IDs in the title would fix
both.
---
(1) and (2) are the ones I'd want resolved before merge; the rest are
smaller. Happy to be wrong on any of these — particularly (2), where the
remaining nested-loop path may well be the deliberate contract you describe in
the doc, in which case it'd be good to say so explicitly.
Generated-by: Claude Opus 5
--
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]
