diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala index 1d25788fdb6c2..47ed2830d868c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala @@ -18,9 +18,9 @@ package org.apache.spark.sql.catalyst.optimizer import org.apache.spark.SparkException -import org.apache.spark.sql.catalyst.expressions.{Alias, And, ArrayDistinct, ArrayExcept, ArrayIntersect, ArraysOverlap, ArrayTransform, ArrayUnion, CaseWhen, Coalesce, CreateArray, CreateMap, CreateNamedStruct, EqualTo, ExpectsInputTypes, Expression, GetStructField, If, IsNull, KnownFloatingPointNormalized, LambdaFunction, Literal, NamedLambdaVariable, TransformValues, UnaryExpression} +import org.apache.spark.sql.catalyst.expressions.{Alias, And, ArrayDistinct, ArrayExcept, ArrayIntersect, ArraysOverlap, ArrayTransform, ArrayUnion, CaseWhen, Coalesce, CreateArray, CreateMap, CreateNamedStruct, EqualTo, ExpectsInputTypes, Expression, GetStructField, If, IsNull, KnownFloatingPointNormalized, LambdaFunction, Literal, NamedLambdaVariable, Or, TransformValues, UnaryExpression} import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode} -import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys +import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, ExtractSingleColumnNullAwareAntiJoin} import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Window} import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.trees.TreePattern._ @@ -94,6 +94,12 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] { } ++ condition j.copy(condition = Some(newConditions.reduce(And))) + // The specialized NAAJ is a hash join, but its OR condition is not an equi-join shape. + case j @ ExtractSingleColumnNullAwareAntiJoin(leftKeys, rightKeys) + if leftKeys.exists(needNormalize) => + val equality = EqualTo(normalize(leftKeys.head), normalize(rightKeys.head)) + j.copy(condition = Some(Or(equality, IsNull(equality)))) + // TODO: ideally Aggregate should also be handled here, but its grouping expressions are // mixed in its aggregate expressions. It's unreliable to change the grouping expressions // here. For now we normalize grouping expressions during planning. See Case 2 in the diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala index d6612bca4e9ed..af7b69107065e 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala @@ -18,7 +18,6 @@ package org.apache.spark.sql.catalyst.optimizer import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.planning.ExtractSingleColumnNullAwareAntiJoin import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule @@ -61,11 +60,12 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] } } - // LeftSemi/LeftAnti over Aggregate + // LeftSemi/LeftAnti over Aggregate, only push down if join can be planned as broadcast join. case join @ Join(agg: Aggregate, rightOp, LeftSemiOrAnti(_), joinCond, _) if agg.aggregateExpressions.forall(_.deterministic) && agg.groupingExpressions.nonEmpty && !agg.aggregateExpressions.exists(ScalarSubquery.hasCorrelatedScalarSubquery) && - canPushThroughCondition(agg.children, joinCond, rightOp) => + canPushThroughCondition(agg.children, joinCond, rightOp) && + canPlanAsBroadcastHashJoin(join, conf) => val aliasMap = getAliasMap(agg) val canPushDownPredicate = (predicate: Expression) => { val replaced = replaceAlias(predicate, aliasMap) @@ -75,20 +75,7 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] val makeJoinCondition = (predicates: Seq[Expression]) => { replaceAlias(predicates.reduce(And), aliasMap) } - val canPushDownJoin = if (ExtractSingleColumnNullAwareAntiJoin.extract(join).isDefined) { - val originalIsBroadcastHash = - NullAwareAntiJoinPlanning.decide(join, conf) == NullAwareAntiJoinPlanning.BroadcastHash - (pushedJoin: Join) => originalIsBroadcastHash && - NullAwareAntiJoinPlanning.decide(pushedJoin, conf) == - NullAwareAntiJoinPlanning.BroadcastHash - } else { - (_: Join) => canPlanAsBroadcastHashJoin(join, conf) - } - pushDownJoin( - join, - canPushDownPredicate, - makeJoinCondition, - canPushDownJoin) + pushDownJoin(join, canPushDownPredicate, makeJoinCondition) // LeftSemi/LeftAnti over Window case join @ Join(w: Window, rightOp, LeftSemiOrAnti(_), _, _) @@ -146,17 +133,11 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] private def pushDownJoin( join: Join, canPushDownPredicate: Expression => Boolean, - makeJoinCondition: Seq[Expression] => Expression, - canPushDownJoin: Join => Boolean = _ => true): LogicalPlan = { + makeJoinCondition: Seq[Expression] => Expression): LogicalPlan = { assert(join.left.children.length == 1) if (join.condition.isEmpty) { - val pushedJoin = join.copy(left = join.left.children.head) - if (canPushDownJoin(pushedJoin)) { - join.left.withNewChildren(Seq(pushedJoin)) - } else { - join - } + join.left.withNewChildren(Seq(join.copy(left = join.left.children.head))) } else { val (pushDown, stayUp) = splitConjunctivePredicates(join.condition.get) .partition(canPushDownPredicate) @@ -170,25 +151,19 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan] if (pushDown.isEmpty || referRightSideCols) { join } else { - val pushedJoin = join.copy( - left = join.left.children.head, condition = Some(makeJoinCondition(pushDown))) - if (!canPushDownJoin(pushedJoin)) { - join + val newPlan = join.left.withNewChildren(Seq(join.copy( + left = join.left.children.head, condition = Some(makeJoinCondition(pushDown))))) + // If there is no more filter to stay up, return the new plan that has join pushed down. + if (stayUp.isEmpty) { + newPlan } else { - val newPlan = join.left.withNewChildren(Seq(pushedJoin)) - // If no predicates remain above the join, return the plan with the join pushed down. - if (stayUp.isEmpty) { - newPlan - } else { - join.joinType match { - // For a left semi join, the non-pushable part of the condition is kept as a Filter - // above the join. - case LeftSemi => Filter(stayUp.reduce(And), newPlan) - // In the case of a left anti join, the join is pushed down only when the entire join - // condition is eligible to be pushed down to preserve the semantics of the left anti - // join. - case _ => join - } + join.joinType match { + // In case of Left semi join, the part of the join condition which does not refer to + // to attributes of the grandchild are kept as a Filter above. + case LeftSemi => Filter(stayUp.reduce(And), newPlan) + // In case of left-anti join, the join is pushed down only when the entire join + // condition is eligible to be pushed down to preserve the semantics of left-anti join. + case _ => join } } } 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 6a4b332b35cd5..4628fc32ea344 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 @@ -353,66 +353,6 @@ trait JoinSelectionHelper extends Logging { } } - def getBroadcastNestedLoopJoinDesiredBuildSide(join: Join): BuildSide = { - if (join.joinType.isInstanceOf[InnerLike] || join.joinType == FullOuter) { - getSmallerSide(join.left, join.right) - } else { - // For perf reasons, BroadcastNestedLoopJoinExec prefers to broadcast the left side for a - // right join and the right side for a left join. If one side is much smaller, revisiting - // that preference may be worthwhile. - if (canBuildBroadcastLeft(join.joinType)) BuildLeft else BuildRight - } - } - - def getBroadcastNestedLoopJoinBuildSide( - join: Join, - hintOnly: Boolean, - conf: SQLConf): Option[BuildSide] = { - lazy val buildLeft = if (hintOnly) { - hintToBroadcastLeft(join.hint) - } else { - canBroadcastBySize(join.left, conf) && - !hintToNotBroadcastAndReplicateLeft(join.hint) - } - lazy val buildRight = if (hintOnly) { - hintToBroadcastRight(join.hint) - } else { - canBroadcastBySize(join.right, conf) && - !hintToNotBroadcastAndReplicateRight(join.hint) - } - - if (join.joinType.isInstanceOf[InnerLike] || join.joinType == FullOuter) { - if (buildLeft && buildRight) { - Some(getBroadcastNestedLoopJoinDesiredBuildSide(join)) - } else if (buildLeft) { - Some(BuildLeft) - } else if (buildRight) { - Some(BuildRight) - } else { - None - } - } else { - getBroadcastNestedLoopJoinDesiredBuildSide(join) match { - case BuildLeft => - if (buildLeft) Some(BuildLeft) else if (buildRight) Some(BuildRight) else None - case BuildRight => - if (buildRight) Some(BuildRight) else if (buildLeft) Some(BuildLeft) else None - } - } - } - - def getBroadcastNestedLoopJoinBuildSide(join: Join, conf: SQLConf): BuildSide = { - val hintedBuildSide = if (join.hint.isEmpty) { - None - } else { - getBroadcastNestedLoopJoinBuildSide(join, hintOnly = true, conf) - } - hintedBuildSide - .orElse(getBroadcastNestedLoopJoinBuildSide(join, hintOnly = false, conf)) - .orElse(getBroadcastNestedLoopJoinBuildSide(join.hint, join.joinType)) - .getOrElse(getBroadcastNestedLoopJoinDesiredBuildSide(join)) - } - def getSmallerSide(left: LogicalPlan, right: LogicalPlan): BuildSide = { if (right.stats.sizeInBytes <= left.stats.sizeInBytes) BuildRight else BuildLeft } @@ -498,9 +438,12 @@ trait JoinSelectionHelper extends Logging { getBroadcastBuildSide(join, hintOnly = true, conf).orElse { if (noShufflePlannedBefore) getBroadcastBuildSide(join, hintOnly = false, conf) else None } - case j if ExtractSingleColumnNullAwareAntiJoin.extract(j).isDefined => - if (NullAwareAntiJoinPlanning.decide(j, conf) == - NullAwareAntiJoinPlanning.BroadcastHash) { + // `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. + case j @ ExtractSingleColumnNullAwareAntiJoin(_, _) => + val threshold = conf.nullAwareAntiJoinBroadcastThreshold + val rightSize = j.right.stats.sizeInBytes + if (threshold < 0 || (threshold > 0 && rightSize >= 0 && rightSize <= threshold)) { Some(BuildRight) } else { None @@ -628,19 +571,3 @@ trait JoinSelectionHelper extends Logging { conf.getConfString("spark.sql.join.forceApplyShuffledHashJoin", "false") == "true" } } - -private[sql] object NullAwareAntiJoinPlanning extends JoinSelectionHelper { - sealed trait Decision - case object BroadcastHash extends Decision - case object BroadcastNestedLoop extends Decision - - def decide(join: Join, conf: SQLConf): Decision = { - if (conf.optimizeNullAwareAntiJoin && - canBroadcastBySize(join.right, conf) && - getBroadcastNestedLoopJoinBuildSide(join, conf) == BuildRight) { - BroadcastHash - } else { - BroadcastNestedLoop - } - } -} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala index 5bc5280450d5a..c4af18fc8739a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala @@ -404,11 +404,12 @@ object ExtractSingleColumnNullAwareAntiJoin extends JoinSelectionHelper with Pre * But if it's a single column case O(M*N) calculation could be optimized into O(M) * using hash lookup instead of loop lookup. */ - private[sql] def extract(join: Join): Option[ReturnType] = join match { + def unapply(join: Join): Option[ReturnType] = join match { case Join(left, right, LeftAnti, Some(Or(e @ EqualTo(leftAttr: Expression, rightAttr: Expression), IsNull(e2 @ EqualTo(_, _)))), _) - if e.semanticEquals(e2) => + if SQLConf.get.optimizeNullAwareAntiJoin && + e.semanticEquals(e2) => if (canEvaluate(leftAttr, left) && canEvaluate(rightAttr, right)) { Some(Seq(leftAttr), Seq(rightAttr)) } else if (canEvaluate(leftAttr, right) && canEvaluate(rightAttr, left)) { @@ -418,10 +419,6 @@ object ExtractSingleColumnNullAwareAntiJoin extends JoinSelectionHelper with Pre } case _ => None } - - def unapply(join: Join): Option[ReturnType] = { - if (SQLConf.get.optimizeNullAwareAntiJoin) extract(join) 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 0b52c0e34d7a5..9eb0277d61f01 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 @@ -7366,15 +7366,34 @@ object SQLConf { val OPTIMIZE_NULL_AWARE_ANTI_JOIN = buildConf("spark.sql.optimizeNullAwareAntiJoin") .internal() - .doc("When true, NULL-aware anti join execution will be planed into " + + .doc("When true, NULL-aware anti join execution can be planned as " + "BroadcastHashJoinExec with flag isNullAwareAntiJoin enabled, " + "optimized from O(M*N) calculation into O(M) calculation " + "using Hash lookup instead of Looping lookup. " + - "Only support for singleColumn NAAJ for now.") + "Only support for singleColumn NAAJ for now. The optimization is also controlled by " + + "spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold.") .version("3.1.0") .booleanConf .createWithDefault(true) + val NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD = + buildConf("spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold") + .internal() + .doc("Configures the maximum estimated size in bytes of the right side of a " + + "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. " + + "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.") + .version("4.2.1") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .bytesConf(ByteUnit.BYTE) + .createWithDefault(-1) + val LEGACY_DUPLICATE_BETWEEN_INPUT = buildConf("spark.sql.legacy.duplicateBetweenInput") .internal() @@ -9743,6 +9762,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def optimizeNullAwareAntiJoin: Boolean = getConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN) + def nullAwareAntiJoinBroadcastThreshold: Long = + getConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD) + def legacyDuplicateBetweenInput: Boolean = getConf(SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT) 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 7a593e1a4aafb..71fa91f5c37e0 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 @@ -17,26 +17,15 @@ package org.apache.spark.sql.catalyst.optimizer -import java.util.concurrent.atomic.AtomicBoolean - import org.apache.spark.sql.catalyst.dsl.expressions._ -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, EqualTo, IsNull, Or} -import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, PlanTest, RightOuter} -import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, HintInfo, Join, JoinHint, LeafNode, NO_BROADCAST_HASH, SHUFFLE_HASH, Statistics} +import org.apache.spark.sql.catalyst.expressions.{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.statsEstimation.StatsTestPlan import org.apache.spark.sql.internal.SQLConf class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper { - private case class TrackingStatsTestPlan( - override val output: Seq[Attribute], - statsAccessed: AtomicBoolean) extends LeafNode { - override def computeStats(): Statistics = { - statsAccessed.set(true) - Statistics(sizeInBytes = 20000000) - } - } - private val left = StatsTestPlan( outputList = Seq($"a".int, $"b".int, $"c".int), rowCount = 20000000, @@ -160,22 +149,6 @@ class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper { assert(getSmallerSide(left, right) === BuildRight) } - test("getBroadcastNestedLoopJoinBuildSide checks the fixed desired side first") { - val leftStatsAccessed = new AtomicBoolean(false) - val uncachedLeft = TrackingStatsTestPlan(Seq($"uncachedLeft".int), leftStatsAccessed) - val leftAntiJoin = Join(uncachedLeft, right, LeftAnti, None, JoinHint.NONE) - val rightStatsAccessed = new AtomicBoolean(false) - val uncachedRight = TrackingStatsTestPlan(Seq($"uncachedRight".int), rightStatsAccessed) - val rightOuterJoin = Join(right, uncachedRight, RightOuter, None, JoinHint.NONE) - - withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") { - assert(getBroadcastNestedLoopJoinBuildSide(leftAntiJoin, SQLConf.get) === BuildRight) - assert(!leftStatsAccessed.get()) - assert(getBroadcastNestedLoopJoinBuildSide(rightOuterJoin, SQLConf.get) === BuildLeft) - assert(!rightStatsAccessed.get()) - } - } - test("canBroadcastBySize should return true if the plan size is less than 10MB") { withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") { assert(canBroadcastBySize(left, SQLConf.get) === false) @@ -183,51 +156,6 @@ class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper { } } - test("canPlanAsBroadcastHashJoin should respect NAAJ size and nested-loop build side") { - 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) - val smallLeft = left.copy(rowCount = 1000, size = Some(1000)) - val largeRight = right.copy(rowCount = 20000000, size = Some(20000000)) - - withSQLConf( - SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") { - assert(canPlanAsBroadcastHashJoin(nullAwareAntiJoin, SQLConf.get)) - assert(!canPlanAsBroadcastHashJoin( - nullAwareAntiJoin.copy(right = largeRight.copy(rowCount = 1)), SQLConf.get)) - assert(!canPlanAsBroadcastHashJoin( - nullAwareAntiJoin.copy(left = smallLeft, right = largeRight), SQLConf.get)) - assert(!canPlanAsBroadcastHashJoin( - nullAwareAntiJoin.copy(hint = JoinHint(hintBroadcast, None)), SQLConf.get)) - } - - withSQLConf( - SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") { - assert(!canPlanAsBroadcastHashJoin(nullAwareAntiJoin, SQLConf.get)) - } - } - - test("canPlanAsBroadcastHashJoin checks the NAAJ right-size short circuit") { - val leftStatsAccessed = new AtomicBoolean(false) - val uncachedLeft = TrackingStatsTestPlan(Seq($"uncachedLeft".int), leftStatsAccessed) - val largeRight = right.copy(rowCount = 20000000, size = Some(20000000)) - val leftKey = uncachedLeft.output.head - val rightKey = largeRight.output.head - val condition = Or(EqualTo(leftKey, rightKey), IsNull(EqualTo(leftKey, rightKey))) - val nullAwareAntiJoin = Join( - uncachedLeft, largeRight, LeftAnti, Some(condition), JoinHint.NONE) - - withSQLConf( - SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") { - assert(!canPlanAsBroadcastHashJoin(nullAwareAntiJoin, SQLConf.get)) - assert(!leftStatsAccessed.get()) - } - } - test("getBroadcastHashJoinBuildSide returns the hinted side") { val equiJoin = join.copy(condition = Some(EqualTo(left.output.head, right.output.head))) withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { @@ -267,17 +195,48 @@ class JoinSelectionHelperSuite extends PlanTest with JoinSelectionHelper { } } - test("getBroadcastHashJoinBuildSide builds from the right for a null-aware anti join") { + 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) + val largeRight = right.copy(rowCount = 20000000, size = Some(20000000)) + val negativeSizeRight = right.copy(size = Some(-1)) + 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)) + assert(getBroadcastHashJoinBuildSide( + nullAwareAntiJoin.copy(right = largeRight), SQLConf.get) === Some(BuildRight)) + assert(getBroadcastHashJoinBuildSide( + nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight)) + } + + withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-2") { + assert(getBroadcastHashJoinBuildSide( + nullAwareAntiJoin.copy(right = overLongMaxRight), SQLConf.get) === Some(BuildRight)) + } + + withSQLConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") { + assert(getBroadcastHashJoinBuildSide(nullAwareAntiJoin, 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) + } + + 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) + assert(getBroadcastHashJoinBuildSide( + nullAwareAntiJoin.copy(right = negativeSizeRight), 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 2b89f752a9974..ba7386f8c9b5e 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,7 +24,6 @@ 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 @@ -110,66 +109,6 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest { comparePlans(optimized, correctAnswer) } - test("Aggregate: NAAJ pushdown when original and rewritten joins can build right") { - val condition = Or($"b" === $"d", IsNull($"b" === $"d")) - val originalQuery = testRelation - .groupBy($"b")($"b", sum($"c")) - .join(testRelation1, joinType = LeftAnti, condition = Some(condition)) - val correctAnswer = testRelation - .join(testRelation1, joinType = LeftAnti, condition = Some(condition)) - .groupBy($"b")($"b", sum($"c")) - - withSQLConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true") { - val analyzedOriginal = originalQuery.analyze - val analyzedCorrectAnswer = correctAnswer.analyze - val originalJoin = analyzedOriginal.asInstanceOf[Join] - val pushedJoin = analyzedCorrectAnswer.asInstanceOf[Aggregate].child.asInstanceOf[Join] - assert(PushDownLeftSemiAntiJoin.canPlanAsBroadcastHashJoin(originalJoin, SQLConf.get)) - assert(PushDownLeftSemiAntiJoin.canPlanAsBroadcastHashJoin(pushedJoin, SQLConf.get)) - comparePlans(Optimize.execute(analyzedOriginal), analyzedCorrectAnswer) - } - } - - test("Aggregate: NAAJ no pushdown when the original join would build left") { - val leftKey = $"leftKey".int - val leftKeyStats = ColumnStat( - distinctCount = Some(1), - min = Some(0), - max = Some(0), - nullCount = Some(0), - avgLen = Some(4), - maxLen = Some(4)) - val child = StatsTestPlan( - Seq(leftKey), 1000, AttributeMap(Seq(leftKey -> leftKeyStats)), Some(1000)) - val aggregate = Aggregate(Seq(leftKey), Seq(leftKey), child) - val right = StatsTestPlan( - Seq($"rightKey".int), 1000, AttributeMap(Seq()), Some(1000)) - val condition = Or( - EqualTo(aggregate.output.head, right.output.head), - IsNull(EqualTo(aggregate.output.head, right.output.head))) - val originalQuery = Join( - aggregate, right, LeftAnti, Some(condition), JoinHint.NONE) - - withSQLConf( - SQLConf.CBO_ENABLED.key -> "true", - SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "100") { - assert(aggregate.stats.sizeInBytes <= SQLConf.get.autoBroadcastJoinThreshold) - assert(child.stats.sizeInBytes > SQLConf.get.autoBroadcastJoinThreshold) - assert(right.stats.sizeInBytes > SQLConf.get.autoBroadcastJoinThreshold) - comparePlans(Optimize.execute(originalQuery), originalQuery) - } - } - - test("Aggregate: ordinary LeftSemi join no pushdown - empty join condition") { - val originalQuery = testRelation - .groupBy($"b")($"b", sum($"c")) - .join(testRelation1, joinType = LeftSemi, condition = None) - val correctAnswer = originalQuery.analyze - - comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer) - } - test("Aggregate: LeftSemi join no pushdown - non-deterministic aggr expressions") { val originalQuery = testRelation .groupBy($"b")($"b", Rand(10).as("c")) @@ -521,7 +460,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest { } Seq(LeftSemi, LeftAnti).foreach { jt => - test(s"SPARK-34081: ordinary $jt only pushes down when broadcast-eligible") { + test(s"SPARK-34081: $jt only push down if join can be planned as broadcast join") { Seq(-1, 100000).foreach { threshold => withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold.toString) { val originalQuery = testRelation diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingPointNumbersSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingPointNumbersSuite.scala index a0a9c8ec32243..a710aebaff0d2 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingPointNumbersSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingPointNumbersSuite.scala @@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.optimizer import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.dsl.plans._ import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayExcept, ArrayIntersect, ArraysOverlap, ArrayTransform, ArrayUnion, CaseWhen, Expression, If, IsNull, KnownFloatingPointNormalized, LambdaFunction, NamedLambdaVariable} -import org.apache.spark.sql.catalyst.plans.PlanTest +import org.apache.spark.sql.catalyst.plans.{LeftAnti, PlanTest} import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.RuleExecutor import org.apache.spark.sql.types.DoubleType @@ -92,6 +92,23 @@ class NormalizeFloatingPointNumbersSuite extends PlanTest { comparePlans(doubleOptimized, correctAnswer) } + test("normalize floating points in null-aware anti join keys") { + val equality = a === b + val query = testRelation1.join( + testRelation2, joinType = LeftAnti, condition = Some(equality || IsNull(equality))) + + val optimized = Optimize.execute(query) + val doubleOptimized = Optimize.execute(optimized) + val normalizedEquality = KnownFloatingPointNormalized(NormalizeNaNAndZero(a)) === + KnownFloatingPointNormalized(NormalizeNaNAndZero(b)) + val correctAnswer = testRelation1.join( + testRelation2, + joinType = LeftAnti, + condition = Some(normalizedEquality || IsNull(normalizedEquality))) + + comparePlans(doubleOptimized, correctAnswer) + } + test("normalize floating points in join keys (equal null safe) - idempotence") { val query = testRelation1.join(testRelation2, condition = Some(a <=> b)) @@ -249,4 +266,3 @@ class NormalizeFloatingPointNumbersSuite extends PlanTest { comparePlans(doubleOptimized, correctAnswer) } } - diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala index 85011386656c5..c25414a009c0d 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala @@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, NamedRelation} import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.optimizer.{BuildRight, BuildSide, JoinSelectionHelper, NormalizeFloatingNumbers, NullAwareAntiJoinPlanning} +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide, JoinSelectionHelper, NormalizeFloatingNumbers} import org.apache.spark.sql.catalyst.planning._ import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.logical._ @@ -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) + if canPlanAsBroadcastHashJoin(j, conf) => + Seq(joins.BroadcastHashJoinExec(leftKeys, rightKeys, LeftAnti, BuildRight, + None, planLater(j.left), planLater(j.right), isNullAwareAntiJoin = true)) // If it is not an equi-join, we first look at the join hints w.r.t. the following order: // 1. broadcast hint: pick broadcast nested loop join. If both sides have the broadcast @@ -369,12 +361,42 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { // 3. Pick broadcast nested loop join as the final solution. It may OOM but we don't have // other choice. It broadcasts the smaller side for inner and full joins, broadcasts the // left side for right join, and broadcasts right side for left join. - case j @ logical.Join(left, right, joinType, condition, hint) => + case logical.Join(left, right, joinType, condition, hint) => checkHintNonEquiJoin(hint) - val desiredBuildSide = getBroadcastNestedLoopJoinDesiredBuildSide(j) + val desiredBuildSide = if (joinType.isInstanceOf[InnerLike] || joinType == FullOuter) { + getSmallerSide(left, right) + } else { + // For perf reasons, `BroadcastNestedLoopJoinExec` prefers to broadcast left side if + // it's a right join, and broadcast right side if it's a left join. + // TODO: revisit it. If left side is much smaller than the right side, it may be better + // to broadcast the left side even if it's a left join. + if (canBuildBroadcastLeft(joinType)) BuildLeft else BuildRight + } def createBroadcastNLJoin(onlyLookingAtHint: Boolean) = { - getBroadcastNestedLoopJoinBuildSide(j, onlyLookingAtHint, conf).map { buildSide => + val buildLeft = if (onlyLookingAtHint) { + hintToBroadcastLeft(hint) + } else { + canBroadcastBySize(left, conf) && !hintToNotBroadcastAndReplicateLeft(hint) + } + + val buildRight = if (onlyLookingAtHint) { + hintToBroadcastRight(hint) + } else { + canBroadcastBySize(right, conf) && !hintToNotBroadcastAndReplicateRight(hint) + } + + val maybeBuildSide = if (buildLeft && buildRight) { + Some(desiredBuildSide) + } else if (buildLeft) { + Some(BuildLeft) + } else if (buildRight) { + Some(BuildRight) + } else { + None + } + + maybeBuildSide.map { buildSide => Seq(joins.BroadcastNestedLoopJoinExec( planLater(left), planLater(right), buildSide, joinType, condition)) } 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 d950357ec2dd6..642fcfde3420c 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 @@ -1282,58 +1282,10 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper } } - test("SPARK-54972: Improve not in subqueries with non-nullable columns") { - withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) { - // testData.key nullable false - // testData2.* nullable false - - val joinExec = assertJoin(( - "select * from testData where key not in (select a from testData2)", - classOf[BroadcastHashJoinExec])) - assert(!joinExec.asInstanceOf[BroadcastHashJoinExec].isNullAwareAntiJoin) - - val joinExec2 = assertJoin(( - "select * from testData where (key, key + 1) not in (select * from testData2)", - classOf[BroadcastHashJoinExec])) - assert(!joinExec2.asInstanceOf[BroadcastHashJoinExec].isNullAwareAntiJoin) - } - } - - test("SPARK-36082: keep in-threshold NAAJ hash join when the fallback builds right") { - withSQLConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) { - val joinExec = assertJoin(( - "select * from testData where key not in (select b from testData3)", - classOf[BroadcastHashJoinExec])) - assert(joinExec.asInstanceOf[BroadcastHashJoinExec].isNullAwareAntiJoin) - } - } - - test("SPARK-36082: disabled NAAJ hash optimization uses nested-loop fallback") { - withSQLConf( - SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "false", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) { - val joinExec = assertJoin(( - "select * from testData where key not in (select b from testData3)", - classOf[BroadcastNestedLoopJoinExec])) - assert(joinExec.asInstanceOf[BroadcastNestedLoopJoinExec].buildSide === BuildRight) - } - } - - test("SPARK-36082: keep over-threshold NAAJ nested-loop join when fallback builds right") { - withSQLConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0") { - val joinExec = assertJoin(( - "select * from testData where key not in (select b from testData3)", - classOf[BroadcastNestedLoopJoinExec])) - assert(joinExec.asInstanceOf[BroadcastNestedLoopJoinExec].buildSide === BuildRight) - } - } - - test("SPARK-36082: keep NAAJ nested-loop join when the fallback builds left") { + test("SPARK-36082: NAAJ broadcast threshold enables a build-left fallback") { val smallLeft = spark.range(1).selectExpr("id AS key") val largeRight = spark.range(100) - .selectExpr("IF(id = 0, CAST(NULL AS BIGINT), id) AS b") + .selectExpr("IF(id = 0, CAST(NULL AS BIGINT), id) AS key") val threshold = statisticSizeInByte(smallLeft) assert(threshold < statisticSizeInByte(largeRight)) @@ -1342,11 +1294,12 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper largeRight.createOrReplaceTempView("naajLargeRight") withSQLConf( SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold.toString) { + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold.toString, + SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> threshold.toString) { val result = sql( - "select * from naajSmallLeft where key not in (select b from naajLargeRight)") + "SELECT * FROM naajSmallLeft WHERE key NOT IN (SELECT key FROM naajLargeRight)") val joinExec = result.queryExecution.sparkPlan.collect { - case j: BroadcastNestedLoopJoinExec => j + case join: BroadcastNestedLoopJoinExec => join } assert(joinExec.size === 1) assert(joinExec.head.buildSide === BuildLeft) @@ -1359,7 +1312,8 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) { + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString, + SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") { withTempView("naajHintedLeft", "naajHintedRight") { Seq[java.lang.Double](-0.0d, 2.0d, null).toDF("key") .createOrReplaceTempView("naajHintedLeft") @@ -1390,7 +1344,8 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0") { + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0", + SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "0") { withTempView("naajFloatingLeft", "naajFloatingRight") { Seq[java.lang.Double](-0.0d, 2.0d, null).toDF("key") .createOrReplaceTempView("naajFloatingLeft") @@ -1410,6 +1365,50 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper } } + test("SPARK-36082: NAAJ hash join preserves floating-point equality") { + Seq(false, true).foreach { adaptiveEnabled => + Seq(false, true).foreach { codegenEnabled => + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptiveEnabled.toString, + SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegenEnabled.toString, + SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "0") { + withTempView("naajHashLeft", "naajHashRight") { + Seq[java.lang.Double](-0.0d, 2.0d, null).toDF("key") + .createOrReplaceTempView("naajHashLeft") + Seq[java.lang.Double](0.0d, 1.0d).toDF("key") + .createOrReplaceTempView("naajHashRight") + + val result = sql( + "select * from naajHashLeft where key not in (select key from naajHashRight)") + checkAnswer(result, Row(2.0d)) + val joinExec = collect(result.queryExecution.executedPlan) { + case join: BroadcastHashJoinExec if join.isNullAwareAntiJoin => join + } + assert(joinExec.size === 1) + } + } + } + } + } + + test("SPARK-54972: Improve not in subqueries with non-nullable columns") { + withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> Long.MaxValue.toString) { + // testData.key nullable false + // testData2.* nullable false + + val joinExec = assertJoin(( + "select * from testData where key not in (select a from testData2)", + classOf[BroadcastHashJoinExec])) + assert(!joinExec.asInstanceOf[BroadcastHashJoinExec].isNullAwareAntiJoin) + + val joinExec2 = assertJoin(( + "select * from testData where (key, key + 1) not in (select * from testData2)", + classOf[BroadcastHashJoinExec])) + assert(!joinExec2.asInstanceOf[BroadcastHashJoinExec].isNullAwareAntiJoin) + } + } + test("SPARK-32399: Full outer shuffled hash join") { val inputDFs = Seq( // Test unique join key