diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala index 4628fc32ea344..189442ce8bc8c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala @@ -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) { Some(BuildRight) } else { None diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 2ec67312f0218..50bbac84be0af 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -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) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala index 71fa91f5c37e0..d3066ee6081bf 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/JoinSelectionHelperSuite.scala @@ -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 @@ -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))) @@ -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( + 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", + 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 { + 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( + 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) } } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala index ba7386f8c9b5e..fe6876d41a874 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala @@ -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 @@ -142,6 +143,48 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest { comparePlans(optimized, originalQuery.analyze) } + test("Aggregate: NAAJ pushdown follows the effective broadcast threshold") { + 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")) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 642fcfde3420c..c5a9ddc927d9c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -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)) } } }