Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cloud-fan wants to merge 3 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b381100
[SPARK-36082][SQL][FOLLOWUP] Floor the NAAJ broadcast threshold
cloud-fan Sep 16, 2026
0f5f063
[SPARK-36082][SQL] Address review comments
cloud-fan Sep 17, 2026
47fceb0
[SPARK-36082][SQL] Address follow-up review comments
cloud-fan Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 19 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -438,12 +438,26 @@ trait JoinSelectionHelper extends Logging {
getBroadcastBuildSide(join, hintOnly = true, conf).orElse {
if (noShufflePlannedBefore) getBroadcastBuildSide(join, hintOnly = false, conf) else None
}
// `JoinSelection` always builds from the right for this shape. A negative threshold preserves
// the original unbounded NAAJ behavior, while zero disables the broadcast hash optimization.
// `JoinSelection` always builds from the right for this shape. Do not reject the hash
// optimization when regular join planning would broadcast the right side, as the fallback
// would still broadcast it with a slower nested-loop join.
case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) =>
val threshold = conf.nullAwareAntiJoinBroadcastThreshold
val rightSize = j.right.stats.sizeInBytes
if (threshold < 0 || (threshold > 0 && rightSize >= 0 && rightSize <= threshold)) {
val dedicatedThreshold = conf.nullAwareAntiJoinBroadcastThreshold
val canBroadcast = if (dedicatedThreshold < 0) {
true
} else {
val automaticBroadcastDisabled = conf.autoBroadcastJoinThreshold < 0 &&
conf.getConf(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_ < 0)
if (dedicatedThreshold == 0 && automaticBroadcastDisabled) {
false
} else {
canBroadcastBySize(j.right, conf) || (dedicatedThreshold > 0 && {
val rightSize = j.right.stats.sizeInBytes
rightSize >= 0 && rightSize <= dedicatedThreshold
})
}
}
if (canBroadcast) {

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MEDIUM

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.

@cloud-fan cloud-fan Sep 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The widening is intentional. The dedicated NAAJ threshold already feeds canPlanAsBroadcastHashJoin, which is the existing aggregate-pushdown gate, so the floored eligibility should apply consistently to both join selection and pushdown. Commit 0f5f063d366 keeps the shared predicate, clarifies that relationship in the config documentation, and adds a LeftSemiAntiJoinPushDownSuite case showing that the floor enables pushdown while disabled thresholds do not. I did not add a separate unfloored predicate.

Some(BuildRight)
} else {
None
Expand Down
18 changes: 14 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7452,12 +7452,22 @@ object SQLConf {
"single-column null-aware anti join for which Spark uses the broadcast hash join " +
"optimization. This configuration takes effect only when " +
"spark.sql.optimizeNullAwareAntiJoin is enabled. A negative value allows the " +
"optimization regardless of the estimated size, while zero disables it. If the " +
"estimated size exceeds a positive value, Spark falls back to regular join planning. " +
"optimization regardless of the estimated size. For a nonnegative value, the " +
"optimization is also allowed when regular join planning considers the right side " +
"broadcastable. " +
"For join selection, regular planning uses " +
"spark.sql.adaptive.autoBroadcastJoinThreshold for runtime statistics when it is set, " +
"and spark.sql.autoBroadcastJoinThreshold otherwise. The same eligibility decision " +
"controls whether a null-aware anti join can be pushed below an aggregate; this " +
"pushdown runs before adaptive execution and therefore uses estimated statistics and " +
"spark.sql.autoBroadcastJoinThreshold. Thus, zero disables the optimization only when " +
"automatic broadcasting is also disabled. When neither threshold admits the right " +
"side, Spark falls back to regular join planning. " +
"The fallback may still broadcast the right side with a nested-loop representation " +
"that uses more memory and runs in O(M * N) time. Join hints do not override this " +
"configuration when the broadcast hash optimization is selected. This configuration " +
"also controls whether a null-aware anti join can be pushed below an aggregate.")
"configuration when the broadcast hash optimization is selected. Set " +
"spark.sql.optimizeNullAwareAntiJoin to false to disable the optimization without " +
"changing automatic broadcast thresholds.")
.version("4.2.1")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.bytesConf(ByteUnit.BYTE)
Expand Down
149 changes: 125 additions & 24 deletions ...yst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@
package org.apache.spark.sql.catalyst.optimizer

import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions.{AttributeMap, EqualTo, IsNull, Or}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, EqualTo, IsNull, Or}
import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, PlanTest}
import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, HintInfo, Join, JoinHint, NO_BROADCAST_HASH, SHUFFLE_HASH}
import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, HintInfo, Join, JoinHint, LeafNode, LogicalPlan, NO_BROADCAST_HASH, SHUFFLE_HASH, Statistics}
import org.apache.spark.sql.catalyst.statsEstimation.StatsTestPlan
import org.apache.spark.sql.internal.SQLConf

Expand All @@ -40,6 +40,11 @@ class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper {

private val join = Join(left, right, Inner, None, JoinHint(None, None))

private def nullAwareAntiJoin(rightPlan: LogicalPlan = right): Join = {
val equality = EqualTo(left.output.head, rightPlan.output.head)
Join(left, rightPlan, LeftAnti, Some(Or(equality, IsNull(equality))), JoinHint.NONE)
}

private val hintBroadcast = Some(HintInfo(Some(BROADCAST)))
private val hintNotToBroadcast = Some(HintInfo(Some(NO_BROADCAST_HASH)))
private val hintShuffleHash = Some(HintInfo(Some(SHUFFLE_HASH)))
Expand Down Expand Up @@ -195,48 +200,144 @@ class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper {
}
}

test("getBroadcastHashJoinBuildSide uses the null-aware anti join broadcast threshold") {
val leftKey = left.output.head
val rightKey = right.output.head
val condition = Or(EqualTo(leftKey, rightKey), IsNull(EqualTo(leftKey, rightKey)))
val nullAwareAntiJoin = Join(left, right, LeftAnti, Some(condition), JoinHint.NONE)
test("NAAJ broadcast threshold is floored by the automatic broadcast threshold") {
val autoThresholdRight = right.copy(
rowCount = 10 * 1024 * 1024,
size = Some(10 * 1024 * 1024))
val betweenThresholdsRight = right.copy(
rowCount = 8 * 1024 * 1024,
size = Some(8 * 1024 * 1024))
val largeRight = right.copy(rowCount = 20000000, size = Some(20000000))
val negativeSizeRight = right.copy(size = Some(-1))
val emptyRight = right.copy(rowCount = 0, size = Some(0))

withSQLConf(

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MEDIUM

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.

@cloud-fan cloud-fan Sep 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Commit 0f5f063d366 adds a default-semantics case that leaves the dedicated configuration unset, disables automatic broadcasting, gives the right side a size above Long.MaxValue, and asserts Some(BuildRight). This directly pins the default -1 as unbounded.

SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin(autoThresholdRight), SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin(largeRight), SQLConf.get).isEmpty)
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "20MB") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin(largeRight), SQLConf.get) === Some(BuildRight))
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "5MB") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin(betweenThresholdsRight), SQLConf.get) === Some(BuildRight))
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get).isEmpty)
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin(emptyRight), SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get).isEmpty)
}
}

test("NAAJ broadcast threshold is unlimited by default") {
val overLongMaxRight = right.copy(
rowCount = BigInt(Long.MaxValue) + 1,
size = Some(BigInt(Long.MaxValue) + 1))

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get) === Some(BuildRight))
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) === Some(BuildRight))
nullAwareAntiJoin(overLongMaxRight), SQLConf.get) === Some(BuildRight))
}
}

test("NAAJ broadcast threshold uses the adaptive threshold for runtime statistics") {
case class RuntimeStatsPlan(size: BigInt) extends LeafNode {
override def output: Seq[Attribute] = right.output
override def computeStats(): Statistics = Statistics(sizeInBytes = size, isRuntime = true)
}
val runtimeRight = RuntimeStatsPlan(5 * 1024 * 1024)

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1MB",

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MEDIUM

Both blocks of this case use a runtime-statistics right side, and these two lines are the only places in the suite that set spark.sql.adaptive.autoBroadcastJoinThreshold, each alongside a positive static threshold. Two mutations survive that.

Drop the plan.stats.isRuntime test in canBroadcastBySize, so the adaptive threshold applies whenever it is set, and both blocks still pass (5MB > 1MB, then 5MB <= 10MB). A block with auto = 10MB, adaptive = 1MB, dedicated 0 and an ordinary non-runtime 5MB right side, asserting Some(BuildRight), would catch it.

Delete && conf.getConf(ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD).forall(_ <= 0) from automaticBroadcastDisabled and the whole suite stays green: every block that reaches that gate either has a positive static threshold or leaves the adaptive one unset. A block with auto = -1, adaptive = 10MB, dedicated 0 and the runtime 5MB right side, asserting Some(BuildRight), would pin the new conjunct.

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correcting the first half of this: dropping the plan.stats.isRuntime test survives this suite but not the repo. AdaptiveQueryExecSuite's Change broadcast join to merge join sets spark.sql.autoBroadcastJoinThreshold=10000 with the adaptive threshold at -1 and asserts one top-level broadcast hash join in the initial physical plan, which runAdaptiveAndVerifyResult hands back as sparkPlan, so estimated statistics reading the adaptive -1 would take that join away. That gate is also not this PR's code.

So the ask reduces to the second half: nothing pins the new forall(_ <= 0) conjunct, and a block with auto = -1, adaptive = 10MB, dedicated 0 and the runtime 5MB right side would.

@cloud-fan cloud-fan Sep 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed on the remaining adaptive-only coverage gap, and thanks for the correction on the first half. Commit 47fceb0e550 adds the runtime-statistics case with automatic threshold -1, adaptive threshold 10MB, and dedicated threshold 0, asserting Some(BuildRight). This pins the new adaptive-threshold conjunct in automaticBroadcastDisabled.

SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(runtimeRight), SQLConf.get).isEmpty)
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1MB",
SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight))
nullAwareAntiJoin(runtimeRight), SQLConf.get) === Some(BuildRight))
}

withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight))
nullAwareAntiJoin(runtimeRight), SQLConf.get) === Some(BuildRight))
}
}

withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get).isEmpty)
test("NAAJ broadcast threshold short-circuits config-only decisions") {
case class ThrowingStatsPlan() extends LeafNode {

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LOW

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.

@cloud-fan cloud-fan Sep 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. ThrowingStatsPlan.computeStats() now explicitly throws IllegalStateException in commit 0f5f063d366, so the short-circuit assertions no longer depend on the current default implementation of LeafNode.computeStats.

override def output: Seq[Attribute] = right.output
override def computeStats(): Statistics =
throw new IllegalStateException("statistics should not be read")
}
val nullAwareAntiJoinWithoutStats = nullAwareAntiJoin(ThrowingStatsPlan())

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get).isEmpty)
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") {
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoinWithoutStats, SQLConf.get) === Some(BuildRight))
}

withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "10MB") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin.copy(right = largeRight), SQLConf.get).isEmpty)
withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoinWithoutStats, SQLConf.get).isEmpty)
}
}

test("NAAJ broadcast threshold rejects unknown sizes and respects the optimization flag") {
withSQLConf(

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MEDIUM

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.

@cloud-fan cloud-fan Sep 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Commit 0f5f063d366 adds both missing cases: a positive 10 MB dedicated threshold admits the normal right side while automatic broadcasting is disabled, and an 8 MB right side is admitted when the dedicated threshold is 5 MB and the automatic threshold is 10 MB. The latter specifically pins the larger-of-the-two behavior.

SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "10MB") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get) === Some(BuildRight))
assert(getBroadcastHashJoinBuildSide(
nullAwareAntiJoin.copy(right = negativeSizeRight), SQLConf.get).isEmpty)
nullAwareAntiJoin(right.copy(size = Some(-1))), SQLConf.get).isEmpty)
}

withSQLConf(
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin(), SQLConf.get).isEmpty)
}
}
}
43 changes: 43 additions & 0 deletions ...rc/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.catalyst.statsEstimation.StatsTestPlan
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.IntegerType

Expand Down Expand Up @@ -142,6 +143,48 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Aggregate: NAAJ pushdown follows the effective broadcast threshold") {

@LuciferYang LuciferYang Sep 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LOW

testRelation and testRelation1 are empty LocalRelations, so computeStats gives sizeInBytes = getSizePerRow(output) * data.length, which is 0 (LocalRelation.scala:107). 0 <= 10MB then holds trivially, and the first block pins only that automatic broadcasting is enabled, not any size comparison: an implementation that admits whenever the automatic threshold is positive, ignoring the size, passes it. The second block is sound, and the analyzed plan really is a fixpoint of this suite's batch once the pushdown is refused.

A third block with a StatsTestPlan right side of 20MB, auto = 10MB and dedicated 0, expecting no pushdown, would pin the size dimension too.

@cloud-fan cloud-fan Sep 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Commit 47fceb0e550 adds a StatsTestPlan right side of 20MB with automatic threshold 10MB and dedicated threshold 0, and asserts that the join remains above the aggregate. This covers the size-rejection dimension in the pushdown suite.

val aggregate = testRelation.groupBy($"b")($"b")
val equality = $"b" === $"d"
val originalQuery = aggregate.join(
testRelation1,
joinType = LeftAnti,
condition = Some(equality || IsNull(equality)))
val pushedDownQuery = testRelation
.join(
testRelation1,
joinType = LeftAnti,
condition = Some(equality || IsNull(equality)))
.groupBy($"b")($"b")
val largeRight = StatsTestPlan(
outputList = testRelation1.output,
rowCount = 20 * 1024 * 1024,
attributeStats = AttributeMap.empty,
size = Some(20 * 1024 * 1024))
val largeRightQuery = aggregate.join(
largeRight,
joinType = LeftAnti,
condition = Some(equality || IsNull(equality)))

withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
comparePlans(Optimize.execute(originalQuery.analyze), pushedDownQuery.analyze)
}

withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
}

withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
comparePlans(Optimize.execute(largeRightQuery.analyze), largeRightQuery.analyze)
}
}

test("Aggregate: LeftSemi join no pushdown") {
val originalQuery = testRelation
.groupBy($"b")($"b", sum($"c").as("sum"))
Expand Down
51 changes: 36 additions & 15 deletions sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1308,34 +1308,55 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper
}
}

test("SPARK-36082: left-broadcast NAAJ fallback uses nested-loop join") {
test("SPARK-36082: NAAJ hash eligibility takes precedence over a left broadcast hint") {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true") {
withTempView("naajHintedLeft", "naajHintedRight") {
Seq[java.lang.Double](-0.0d, 2.0d, null).toDF("key")
.createOrReplaceTempView("naajHintedLeft")
Seq[java.lang.Double](0.0d, 1.0d).toDF("key")
.createOrReplaceTempView("naajHintedRight")

val result = sql(
val query =
"select /*+ BROADCAST(naajHintedLeft) */ naajHintedLeft.* " +
"from naajHintedLeft left anti join naajHintedRight on " +
"naajHintedLeft.key = naajHintedRight.key or " +
"isnull(naajHintedLeft.key = naajHintedRight.key)")
val plan = result.queryExecution.sparkPlan
val nestedLoopJoins = plan.collect {
case join: BroadcastNestedLoopJoinExec => join
"isnull(naajHintedLeft.key = naajHintedRight.key)"

withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString,
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
val result = sql(query)
val plan = result.queryExecution.sparkPlan
val nestedLoopJoins = plan.collect {
case join: BroadcastNestedLoopJoinExec => join
}
val nullAwareHashJoins = plan.collect {
case join: BroadcastHashJoinExec if join.isNullAwareAntiJoin => join
}
assert(nestedLoopJoins.isEmpty)
assert(nullAwareHashJoins.size === 1)
assert(nullAwareHashJoins.head.buildSide === BuildRight)
checkAnswer(result, Row(2.0d))
}
val nullAwareHashJoins = plan.collect {
case join: BroadcastHashJoinExec if join.isNullAwareAntiJoin => join

withSQLConf(
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") {
val result = sql(query)
val plan = result.queryExecution.sparkPlan
val nestedLoopJoins = plan.collect {
case join: BroadcastNestedLoopJoinExec => join
}
val nullAwareHashJoins = plan.collect {
case join: BroadcastHashJoinExec if join.isNullAwareAntiJoin => join
}
assert(nestedLoopJoins.size === 1)
assert(nestedLoopJoins.head.buildSide === BuildLeft)
assert(nullAwareHashJoins.isEmpty)
checkAnswer(result, Row(2.0d))
}
assert(nestedLoopJoins.size === 1)
assert(nestedLoopJoins.head.buildSide === BuildLeft)
assert(nullAwareHashJoins.isEmpty)
checkAnswer(result, Row(2.0d))
}
}
}
Expand Down