[SPARK-36082][SQL][FOLLOWUP] Floor the NAAJ broadcast threshold

[cloud-fan]

What changes were proposed in this pull request?

This PR follows #58631.

For a nonnegative spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold, this PR uses the
larger of that value and spark.sql.autoBroadcastJoinThreshold as the effective threshold. A
negative dedicated threshold remains unlimited.

The implementation also avoids reading plan statistics when the dedicated threshold is unlimited,
or when both thresholds disable broadcasting. The configuration documentation and focused
JoinSelectionHelperSuite coverage are updated for these semantics.

Why are the changes needed?

If the dedicated NAAJ threshold is lower than the automatic broadcast threshold, it can reject the
specialized null-aware broadcast hash join even though regular join planning will still broadcast
the right side. That fallback uses a nested-loop representation and changes hash lookup from
O(M + N) to O(M * N) without avoiding the broadcast.

Flooring the dedicated threshold by the automatic threshold prevents this performance regression.
The dedicated threshold remains useful for allowing the specialized NAAJ hash join above the
automatic threshold.

Does this PR introduce any user-facing change?

Yes, on unreleased master and for an internal configuration. A nonnegative dedicated NAAJ
threshold can no longer disable the specialized hash optimization for inputs that are within
spark.sql.autoBroadcastJoinThreshold. To disable both, set both thresholds to a nonpositive
value.

How was this patch tested?

Added focused JoinSelectionHelperSuite tests for the automatic-threshold floor, unlimited and
disabled short-circuit paths, unknown statistics, and the NAAJ optimization flag.

The following local command was attempted:

build/sbt "catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.JoinSelectionHelperSuite"

The test process did not start because the local environment could not resolve Maven and SBT
repositories while loading missing SBT plugins (UnknownHostException).

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[cloud-fan]

cc @sunchao @LuciferYang

[LuciferYang]

Reviewed b381100da1a. Seven items, most severe first; each names the line it is about.

1. HIGH — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:446

SPARK-36082: left-broadcast NAAJ fallback uses nested-loop join in JoinSuite.scala:1311 will fail. It sets autoBroadcastJoinThreshold = Long.MaxValue with the dedicated threshold at 0, which pins exactly the case where regular planning would broadcast the right side and the dedicated threshold disables the hash optimization. With the floor, max(0, Long.MaxValue) puts any right side under the threshold, so the plan becomes a null-aware BroadcastHashJoinExec and all three of that test's plan assertions fail, the first one aborting it. The answer itself is unchanged.

That test came with SPARK-36082 itself, so either the behavior it records is no longer wanted, in which case this PR should update it and say why, or the floor needs to not cover this configuration. It is also the only case covering the hint-broadcasts-the-left-side path down to the nested-loop fallback, so nothing guards that path once it changes.

2. MEDIUM — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:447

The floor reads conf.autoBroadcastJoinThreshold, but once this branch returns None, whether the fallback broadcasts the right side is decided by canBroadcastBySize (joins.scala:363), which switches to spark.sql.adaptive.autoBroadcastJoinThreshold for runtime stats. Under AQE the two can use different thresholds: with the adaptive limit at 1MB, the static one at 10MB and the dedicated threshold at 0, a right stage measured at 5MB is admitted here and broadcast, while regular planning would have broadcast the 512KB left side instead. With the adaptive limit above the static one it errs the other way and rejects a case the fallback really would broadcast.

Replacing only the math.max half with the planner's own predicate, so the shape becomes dedicatedThreshold < 0 || canBroadcastBySize(j.right, conf) || (dedicatedThreshold > 0 && rightSize >= 0 && rightSize <= dedicatedThreshold), makes the rationale hold on its own and the new comment true as written. One constraint: the second block of this PR's new short-circuits config-only decisions pins that a config-only rejection never reads stats (auto = -1, dedicated 0, a right side whose stats throw), and canBroadcastBySize has to read stats.isRuntime to pick its threshold, so that config-only rejection needs to stay ahead of it.

3. MEDIUM — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:453

canPlanAsBroadcastHashJoin has a second consumer, PushDownLeftSemiAntiJoin (:68), which uses the answer to decide whether to push a LeftSemi/Anti join below an Aggregate. The question there is whether the join stays an O(M) hash join after the push, measured against the pre-aggregation row count, so "the fallback would broadcast the right side anyway" does not transfer: a BNLJ below the aggregate is O(M * N) with an un-aggregated M.

The risk already existed for size-gated values, a positive dedicated threshold, and the floor adds the 0-to-static-threshold band to it; nothing lifts a pushed-down join back above the aggregate. If only the cost decision in join selection is meant to change, PushDownLeftSemiAntiJoin could ask an unfloored predicate, or the description could state that widening the rewrite is intended. Either way, SQLConf.scala:7463 ("This configuration also controls whether a null-aware anti join can be pushed below an aggregate") is now incomplete, since the floor puts spark.sql.autoBroadcastJoinThreshold in charge of that pushdown too; and no case in LeftSemiAntiJoinPushDownSuite sets either conf today, so one would record whichever decision you take.

4. MEDIUM — sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:209

Each of the three new tests sets NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD explicitly, so the old block that left the dedicated conf at its default and asserted Some(BuildRight) for largeRight is gone. Nothing pins the default now: change it from -1 to a byte size and the suite stays green, while NAAJ planning goes from size-blind to size-gated, which is user visible. (Changing it to 0 is caught by JoinSuite.scala:1368.)

Cheap to restore: add a block that leaves the dedicated conf alone and asserts largeRight -> Some(BuildRight), rather than repurposing one of the three existing blocks, each of which pins a side of the floor.

5. MEDIUM — sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:258

The first block of NAAJ broadcast threshold rejects unknown sizes and respects the optimization flag is the only configuration with automatic broadcasting off and a positive dedicated threshold, and it only asserts the negative-size rejection. The old positive case under threshold = 10MB is gone, so nothing covers that combination.

One more assertion in that block, that nullAwareAntiJoin() gives Some(BuildRight), closes it. Separately, no block lands in 0 < dedicated < auto, so an implementation written as if (dedicatedThreshold == 0) auto else dedicated also survives the suite; one block in that band would record the "larger of the two" the doc promises.

6. LOW — sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala:237

ThrowingStatsPlan overrides only output; "reading stats throws" comes from LeafNode.computeStats. The test's name claims short-circuiting, but that premise appears nowhere in the test. If LeafNode ever gains a fallback estimate, both assertions stay green while proving nothing.

Overriding computeStats() in that class to throw explicitly, or one comment naming LeafNode.computeStats, puts the premise inside the test.

7. LOW — sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7455

With the floor, any nonnegative dedicated threshold at or below spark.sql.autoBroadcastJoinThreshold is indistinguishable from not setting it at all, so 0 is no longer an off switch. A user who wants automatic broadcasting untouched but the NAAJ hash optimization off is left with the internal spark.sql.optimizeNullAwareAntiJoin=false, and the NAAJ key normalization it also turns off does not presuppose the hash join was chosen: it rewrites the condition during logical optimization, so it applies on the nested-loop path too, which is what JoinSuite.scala:1343 covers.

The doc states the new semantics; naming that replacement switch would tell a reader who set 0 under the old doc where to go instead. Also, the description's "To disable both, set both thresholds to a nonpositive value" is inverted: a negative dedicated threshold means unbounded, so following it does the opposite. Neither this conf's commit nor its parent is in any tag yet, so no migration note is needed.