diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index ea8f77498e5d5..7f91596faeedd 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -10217,7 +10217,7 @@ }, "_LEGACY_ERROR_TEMP_1046" : { "message" : [ - "Join strategy hint parameter should be an identifier or string but was ()." + "Join hint parameter should be an identifier or string but was ()." ] }, "_LEGACY_ERROR_TEMP_1047" : { diff --git a/docs/sql-ref-syntax-qry-select-hints.md b/docs/sql-ref-syntax-qry-select-hints.md index c9c591d6728e3..c942e3498104a 100644 --- a/docs/sql-ref-syntax-qry-select-hints.md +++ b/docs/sql-ref-syntax-qry-select-hints.md @@ -172,6 +172,55 @@ SELECT /*+ SHUFFLE_REPLICATE_NL(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.ke SELECT /*+ BROADCAST(t1), MERGE(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key; ``` +### Runtime Filter Hints + +A runtime filter prunes one side of a join using the join key values found on the other side, so +rows that cannot match are discarded early. Spark decides on its own whether such a filter is worth +building, based on estimates of how much data it would save. Runtime filter hints let users make +that decision instead, for the cases where the estimates are unavailable or wrong. + +#### Runtime Filter Hints Types + +* **RUNTIME_FILTER** + + Suggests that Spark build a runtime filter from the hinted relation and use it to prune the + other side of the join. Use it when the hinted side is known to match only a small fraction + of the other side, but Spark does not choose a runtime filter on its own, typically because + table statistics are missing or misleading. The hinted side may be any relation or subquery, + and is never itself pruned. The hint does not choose how the pruning is done; Spark picks the + mechanism. `RUNTIME_FILTER` can be combined with a join strategy hint. + +The hint overrides Spark's cost estimates, but not the requirements that make a runtime filter +correct, so Spark is not guaranteed to follow it. A side that join semantics forbid pruning is +never pruned, e.g. the left side of a `LEFT OUTER` join, whose rows must all appear in the output. +The hinted side must produce the same rows each time it is evaluated, since building the filter +may evaluate it separately from the join; a side whose rows or join keys depend on evaluation +order, such as a `LIMIT` without a unique ordering, a `TABLESAMPLE` without `REPEATABLE`, or a +key computed by `first` or `last`, does not qualify. Building the filter may evaluate the hinted +side once more, which is the cost the hint asks Spark to spend. A hint that cannot be applied does +not make Spark build a filter in the opposite direction instead. + +Spark issues a warning with the reason when it cannot apply the hint. Hinting both sides of a join +is ambiguous, since each side would then have to be built from the other; Spark warns and ignores +the hint. + +#### Examples + +```sql +-- Build a runtime filter from t2 and use it to prune t1. +SELECT /*+ RUNTIME_FILTER(t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key; + +-- The hinted side may be any relation or subquery, not only a table. +SELECT /*+ RUNTIME_FILTER(t2) */ * +FROM t1 INNER JOIN (SELECT DISTINCT key FROM t3) t2 ON t1.key = t2.key; + +-- A runtime filter hint can be combined with a join strategy hint. +SELECT /*+ MERGE(t1, t2), RUNTIME_FILTER(t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key; + +-- Hinting both sides is ambiguous, so Spark issues a warning and ignores the hint. +SELECT /*+ RUNTIME_FILTER(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key; +``` + ### Related Statements * [JOIN](sql-ref-syntax-qry-select-join.html) diff --git a/python/pyspark/sql/connect/proto/relations_pb2.pyi b/python/pyspark/sql/connect/proto/relations_pb2.pyi index 2d17e88446d60..80338e3511b87 100644 --- a/python/pyspark/sql/connect/proto/relations_pb2.pyi +++ b/python/pyspark/sql/connect/proto/relations_pb2.pyi @@ -3448,7 +3448,8 @@ class Hint(google.protobuf.message.Message): name: builtins.str """(Required) Hint name. - Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL. + Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL, + RUNTIME_FILTER. Supported partitioning hints include COALESCE, REPARTITION, REPARTITION_BY_RANGE. """ diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 35b9052686dcf..1a3ea43f9f7e8 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -586,7 +586,7 @@ class Analyzer( Batch("Disable Hints", Once, new ResolveHints.DisableHints), Batch("Hints", fixedPoint, - Seq(ResolveHints.ResolveJoinStrategyHints, + Seq(ResolveHints.ResolveJoinHints, ResolveHints.ResolveCoalesceHints) ++ hintResolutionRules: _*), Batch("Simple Sanity Check", Once, diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala index 2998dd7e4f682..adad083a81757 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala @@ -31,7 +31,8 @@ import org.apache.spark.sql.internal.SQLConf /** - * Collection of rules related to hints. The only hint currently available is join strategy hint. + * Collection of rules related to hints: join strategy hints, the runtime filter hint, and + * partitioning hints. * * Note that this is separately into two rules because in the future we might introduce new hint * rules that have different ordering requirements from join strategies. @@ -41,7 +42,7 @@ object ResolveHints { /** * Checks if the given multi-part identifiers are matched with each other. * - * The [[ResolveJoinStrategyHints]] rule is applied before the resolution batch in the analyzer + * The [[ResolveJoinHints]] rule is applied before the resolution batch in the analyzer * and we cannot semantically compare them at this stage. Therefore, we follow a simple rule; * they match if an identifier in a hint is a tail of an identifier in a relation. This process * is independent of a session catalog (`currentDb` in [[SessionCatalog]]) and it just compares @@ -77,22 +78,28 @@ object ResolveHints { * is not aliased differently), subquery, or common table expression that match the specified * name. * + * [[RuntimeFilterHint]] is resolved here too. It takes the same per-relation form, e.g. + * "RUNTIME_FILTER(a)", and applies to a join side, so it shares this rule's relation matching + * and lands on the same [[HintInfo]] -- which is what lets it accompany a join strategy hint on + * a relation instead of displacing one. + * * The hint resolution works by recursively traversing down the query plan to find a relation or * subquery that matches one of the specified relation aliases. The traversal does not go past * beyond any view reference, with clause or subquery alias. * * This rule must happen before common table expressions. */ - object ResolveJoinStrategyHints extends Rule[LogicalPlan] { + object ResolveJoinHints extends Rule[LogicalPlan] { private def hintErrorHandler = conf.hintErrorHandler def resolver: Resolver = conf.resolver private def createHintInfo(hintName: String): HintInfo = { - HintInfo(strategy = - JoinStrategyHint.strategies.find( + HintInfo( + strategy = JoinStrategyHint.strategies.find( _.hintAliases.map( - _.toUpperCase(Locale.ROOT)).contains(hintName.toUpperCase(Locale.ROOT)))) + _.toUpperCase(Locale.ROOT)).contains(hintName.toUpperCase(Locale.ROOT))), + runtimeFilterSource = RuntimeFilterHint.isRuntimeFilterHintName(hintName)) } private def matchedIdentifier(identInHint: Seq[String], identInQuery: Seq[String]): Boolean = @@ -157,7 +164,8 @@ object ResolveHints { def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning( _.containsPattern(UNRESOLVED_HINT), ruleId) { - case h: UnresolvedHint if JoinStrategyHint.isJoinStrategyHintName(h.name) => + case h: UnresolvedHint if JoinStrategyHint.isJoinStrategyHintName(h.name) || + RuntimeFilterHint.isRuntimeFilterHintName(h.name) => if (h.parameters.isEmpty) { // If there is no table alias specified, apply the hint on the entire subtree. ResolvedHint(h.child, createHintInfo(h.name)) @@ -167,7 +175,7 @@ object ResolveHints { case StringLiteral(tableName) => UnresolvedAttribute.parseAttributeName(tableName) case tableId: UnresolvedAttribute => tableId.nameParts case unsupported => - throw QueryCompilationErrors.joinStrategyHintParameterNotSupportedError(unsupported) + throw QueryCompilationErrors.joinHintParameterNotSupportedError(unsupported) }.toSet val relationsInHintWithMatch = new mutable.HashSet[Seq[String]] val applied = applyJoinStrategyHint( diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala index aef521e8d0a7d..a8c71632b896d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala @@ -17,11 +17,14 @@ package org.apache.spark.sql.catalyst.optimizer +import java.util.Locale + import scala.annotation.tailrec +import scala.util.{Left, Right} import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.expressions.aggregate.BloomFilterAggregate -import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys +import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, NodeWithOnlyDeterministicProjectAndFilter} import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.trees.TreePattern.{INVOKE, JSON_TO_STRUCT, LIKE_FAMLIY, PYTHON_UDF, REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE, SCALA_UDF} @@ -34,20 +37,35 @@ import org.apache.spark.sql.internal.SQLConf * the creation side is a table scan with a selective filter. * The runtime filter is logically an IN subquery with the join keys. Currently it's always * bloom filter but we may add other physical implementations in the future. + * + * A [[RuntimeFilterHint]] on a join side ("RUNTIME_FILTER(dim)") requests that side be used as + * the creation side, whatever its shape, and waives the checks that only estimate whether a filter + * pays off: the user has asserted the benefit they try to predict. It waives no correctness + * requirement (see `JoinSelectionHelper.isRepeatableRuntimeFilterSource`), and it does not lift + * the limits on the number of filters and on a filter's size. A hinted side is never itself + * filtered, and a hint takes effect through one mechanism: when a DPP filter honors it on any + * join key, no Bloom filter is added for that join. When the hint is not applied, the reason is + * reported through the hint error handler. */ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with JoinSelectionHelper { + private def hintErrorHandler = conf.hintErrorHandler + private case class FilterCreationSide( key: Expression, plan: LogicalPlan, useMaterializedThreshold: Boolean, materializedRowCount: Option[BigInt] = None, - materializedSizeInBytes: Option[BigInt] = None) + materializedSizeInBytes: Option[BigInt] = None, + hinted: Boolean = false) + /** + * Returns the application side with a runtime filter on its key, or the reason none was built. + */ private def injectFilter( filterApplicationSideKey: Expression, filterApplicationSidePlan: LogicalPlan, - filterCreationSide: FilterCreationSide): LogicalPlan = { + filterCreationSide: FilterCreationSide): Either[String, LogicalPlan] = { injectBloomFilter( filterApplicationSideKey, filterApplicationSidePlan, @@ -58,18 +76,23 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J private def injectBloomFilter( filterApplicationSideKey: Expression, filterApplicationSidePlan: LogicalPlan, - filterCreationSide: FilterCreationSide): LogicalPlan = { + filterCreationSide: FilterCreationSide): Either[String, LogicalPlan] = { val filterCreationSideKey = filterCreationSide.key val filterCreationSidePlan = filterCreationSide.plan - val creationSideThreshold = if (filterCreationSide.useMaterializedThreshold) { - conf.runtimeFilterMaterializedCreationSideThreshold + val creationSideThresholdConf = if (filterCreationSide.useMaterializedThreshold) { + SQLConf.RUNTIME_BLOOM_FILTER_MATERIALIZED_CREATION_SIDE_THRESHOLD } else { - conf.runtimeFilterCreationSideThreshold + SQLConf.RUNTIME_BLOOM_FILTER_CREATION_SIDE_THRESHOLD } - // Skip if the filter creation side is too big - if (filterCreationSide.materializedSizeInBytes - .getOrElse(filterCreationSidePlan.stats.sizeInBytes) > creationSideThreshold) { - return filterApplicationSidePlan + val creationSideThreshold = conf.getConf(creationSideThresholdConf) + val creationSideSize = filterCreationSide.materializedSizeInBytes + .getOrElse(filterCreationSidePlan.stats.sizeInBytes) + // Skip if the filter creation side is too big. This estimates whether the filter is worth its + // cost, which a hint asserts, and the filter's size is bounded by the max number of bits + // regardless, so a hinted creation side is not subject to it. + if (!filterCreationSide.hinted && creationSideSize > creationSideThreshold) { + return Left(s"the creation side ($creationSideSize bytes) exceeds " + + s"${creationSideThresholdConf.key} ($creationSideThreshold bytes)") } val rowCount = filterCreationSide.materializedRowCount .orElse(filterCreationSidePlan.stats.rowCount) @@ -82,16 +105,30 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J val alias = Alias(bloomFilterAgg.toAggregateExpression(), "bloomFilter")() val aggregate = - ConstantFolding(ColumnPruning(Aggregate(Nil, Seq(alias), filterCreationSidePlan))) + ConstantFolding(pruneColumns(Aggregate(Nil, Seq(alias), filterCreationSidePlan))) // Runtime filters are introduced after subquery optimization, so Python UDFs in a new // creation-side subquery cannot be extracted into a Python evaluation operator. if (aggregate.containsPattern(PYTHON_UDF)) { - return filterApplicationSidePlan + return Left("the creation side contains a Python UDF") } val bloomFilterSubquery = ScalarSubquery(aggregate, Nil) val filter = BloomFilterMightContain(bloomFilterSubquery, new XxHash64(Seq(filterApplicationSideKey))) - Filter(filter, filterApplicationSidePlan) + Right(Filter(filter, filterApplicationSidePlan)) + } + + // Prunes the columns of the new subquery to a fixed point. A hinted creation side can have any + // shape, e.g. an aggregate or a window whose extra columns take more than one pass to remove. + private def pruneColumns(plan: LogicalPlan): LogicalPlan = { + var current = plan + var pruned = ColumnPruning(current) + var iteration = 1 + while (!pruned.fastEquals(current) && iteration < conf.optimizerMaxIterations) { + current = pruned + pruned = ColumnPruning(current) + iteration += 1 + } + pruned } /** @@ -302,13 +339,26 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J * - The filter creation side has a selective predicate, or its exact materialized row count * is smaller than the application side's distinct join-key count * - The max filterApplicationSide scan size is greater than a configurable threshold + * + * All three predict whether a filter pays off, so a [[RuntimeFilterHint]] on the creation side + * (`hinted`) waives them: the creation side is used as it is, whatever its shape, as long as it + * is a repeatable source, since the filter subquery re-executes it. Returns the reason when no + * creation side is extracted; only a hinted one is reported. */ private def extractBeneficialFilterCreatePlan( filterApplicationSide: LogicalPlan, filterCreationSide: LogicalPlan, filterApplicationSideKey: Expression, - filterCreationSideKey: Expression): Option[FilterCreationSide] = { - if (findExpressionAndTrackLineageDown( + filterCreationSideKey: Expression, + hinted: Boolean): Either[String, FilterCreationSide] = { + if (hinted) { + runtimeFilterSourceRejection(filterCreationSide, filterCreationSideKey).toLeft( + FilterCreationSide( + filterCreationSideKey, + filterCreationSide, + useMaterializedThreshold = false, + hinted = true)) + } else if (findExpressionAndTrackLineageDown( filterApplicationSideKey, filterApplicationSide).isDefined && satisfyByteSizeRequirement(filterApplicationSide)) { val allowMaterializedCache = UnsafeRowUtils.isBinaryStable(filterCreationSideKey.dataType) && @@ -346,7 +396,7 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J currentDistinctCount } } - if (allowMaterializedCache) { + val extracted = if (allowMaterializedCache) { var sawMaterializedLeaf = false val selectiveCreationSide = extractSelectiveFilterOverScan( filterCreationSide, @@ -374,24 +424,45 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J allowMaterializedCache = false, applicationDistinctCount = None) } + extracted.toRight("no selective creation side") } else { - None + Left("the application side does not qualify") } } - // This checks if there is already a DPP filter, as this rule is called just after DPP. + // Returns the DPP filter on `key` at the top of `plan`, as this rule is called just after DPP. @tailrec - private def hasDynamicPruningSubquery( - left: LogicalPlan, - right: LogicalPlan, - leftKey: Expression, - rightKey: Expression): Boolean = { - (left, right) match { - case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _), plan), _) => - pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan, right, leftKey, rightKey) - case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _), plan)) => - pruningKey.fastEquals(rightKey) || - hasDynamicPruningSubquery(left, plan, leftKey, rightKey) + private def findDynamicPruning( + plan: LogicalPlan, + key: Expression): Option[DynamicPruningSubquery] = plan match { + case Filter(dpp @ DynamicPruningSubquery(pruningKey, _, _, _, _, _, _), child) => + if (pruningKey.fastEquals(key)) Some(dpp) else findDynamicPruning(child, key) + case _ => None + } + + /** + * Whether the DPP filter `exprId` at the top of `prunedSide` reaches the scan. It is not final + * here: `PushDownPredicates` carries it towards the scan later, and + * `CleanupDynamicPruningFilters` then keeps it only in a chain of deterministic projections and + * filters directly over the scan. Simulate that with the same pushdown rule rather than + * predicting what it can push through. The cleanup also folds a filter into an equality on the + * same key already sitting on the scan, which prunes at least as much. + */ + private def dynamicPruningReachesScan(prunedSide: LogicalPlan, exprId: ExprId): Boolean = { + var plan = prunedSide + var pushed = PushDownPredicates(plan) + var iteration = 1 + while (!pushed.fastEquals(plan) && iteration < conf.optimizerMaxIterations) { + plan = pushed + pushed = PushDownPredicates(plan) + iteration += 1 + } + pushed.exists { + case f @ Filter(condition, _) if condition.exists { + case dpp: DynamicPruningSubquery => dpp.exprId == exprId + case _ => false + } => + NodeWithOnlyDeterministicProjectAndFilter.unapply(f).exists(_.isInstanceOf[LeafNode]) case _ => false } } @@ -411,55 +482,169 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = { var filterCounter = 0 val numFilterThreshold = conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD) + val bloomFilterEnabled = conf.runtimeFilterBloomFilterEnabled plan transformUp { case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _, left, right, hint) => var newLeft = left var newRight = right + // A side hinted as the runtime filter source is the creation side, so the filter is + // applied to the other side. An ambiguous hint is reported and otherwise ignored, leaving + // the heuristics to decide. + val hintedSource = runtimeFilterSourceSide(hint) + val hinted = hintedSource.isDefined + val injectLeftHinted = hintedSource.contains(BuildRight) + val injectRightHinted = hintedSource.contains(BuildLeft) + if (isRuntimeFilterHintAmbiguous(hint)) { + reportHintNotApplied(join, + "the runtime filter source is ambiguous as both join sides are hinted") + } + var appliedHint = false + // The first reason the hint could not be applied on a key. The hinted side is the same + // for every key, so the first reason is as representative as any. + var notAppliedReason: Option[String] = None + def hintBlocked(reason: => String): Unit = { + if (notAppliedReason.isEmpty) notAppliedReason = Some(reason) + } + lazy val hasShuffle = isProbablyShuffleJoin(left, right, hint) + // Tries to filter `applicationSide` with a filter built from `creationSide`. Returns the + // filtered side, recording the reason when this direction is the hinted one and no filter + // was added. Requirements: + // 1. The join type supports pruning the application side + // 2. The application side is not the hinted source, which is never itself filtered + // 3. The join is a shuffle join, or a broadcast join with a shuffle below it -- an + // estimate of whether the filter pays off, so a hint waives it + // 4. There is no Bloom filter on the application side's key yet + def tryInject( + applicationSide: LogicalPlan, + currentApplicationSide: LogicalPlan, + applicationSideKey: Expression, + creationSide: LogicalPlan, + creationSideKey: Expression, + canPrune: Boolean, + applicationHinted: Boolean, + creationHinted: Boolean, + sideName: String): Option[LogicalPlan] = { + def blocked(reason: => String): Option[LogicalPlan] = { + if (applicationHinted) hintBlocked(reason) + None + } + if (!canPrune) { + blocked(s"the $sideName side of a " + + s"${joinType.sql.toLowerCase(Locale.ROOT)} join cannot be pruned") + } else if (creationHinted || + !(applicationHinted || hasShuffle || probablyHasShuffle(applicationSide))) { + None + } else if (hasBloomFilter(currentApplicationSide, applicationSideKey)) { + blocked("a runtime filter on the join key already exists") + } else { + extractBeneficialFilterCreatePlan(applicationSide, creationSide, + applicationSideKey, creationSideKey, applicationHinted) + .flatMap(injectFilter(applicationSideKey, currentApplicationSide, _)) + .fold(reason => blocked(reason), Some(_)) + } + } + // A DPP filter prunes by whole partitions rather than by rows, so it is preferred. The + // hint is honored by one that prunes the application side and survives to the physical + // plan, and then takes effect through that mechanism alone: no Bloom filter is added on + // any key of the join. A Bloom filter needs no pushdown, so one is added instead when no + // DPP filter survives. + val dppHonorsHint = hinted && leftKeys.lazyZip(rightKeys).exists { (l, r) => + val (applicationSide, applicationSideKey) = + if (injectLeftHinted) (left, l) else (right, r) + findDynamicPruning(applicationSide, applicationSideKey) + .exists(dpp => dynamicPruningReachesScan(applicationSide, dpp.exprId)) + } + appliedHint = dppHonorsHint leftKeys.lazyZip(rightKeys).foreach((l, r) => { - // Check if: - // 1. There is already a DPP filter on the key - // 2. The keys are simple cheap expressions - if (filterCounter < numFilterThreshold && - !hasDynamicPruningSubquery(left, right, l, r) && - isSimpleExpression(l) && isSimpleExpression(r)) { + val dppOnLeft = findDynamicPruning(left, l) + val dppOnRight = findDynamicPruning(right, r) + // A DPP filter that prunes the hinted side is a runtime filter on the key too, and only + // one is built per key. + val dppOnHintedSide = + hinted && (if (injectLeftHinted) dppOnRight else dppOnLeft).isDefined + if (dppHonorsHint || (!hinted && (dppOnLeft.isDefined || dppOnRight.isDefined))) { + // Already pruned by partition pruning: no Bloom filter for the key. + } else if (dppOnHintedSide) { + hintBlocked("a dynamic partition pruning filter on the join key already prunes " + + "the hinted side") + } else if (!bloomFilterEnabled) { + hintBlocked(s"${SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key} is false") + } else if (filterCounter >= numFilterThreshold) { + hintBlocked( + s"${SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD.key} ($numFilterThreshold) is reached") + } else if (!isSimpleExpression(l) || !isSimpleExpression(r)) { + // The keys become the filter's input and must be cheap to evaluate. + hintBlocked("the join key is not a simple expression") + } else { val oldLeft = newLeft val oldRight = newRight - // Check if: - // 1. The current join type supports prune the left side with runtime filter - // 2. The current join is a shuffle join or a broadcast join that - // has a shuffle below it - // 3. There is no bloom filter on the left key yet - val hasShuffle = isProbablyShuffleJoin(left, right, hint) - if (canPruneLeft(joinType) && (hasShuffle || probablyHasShuffle(left)) && - !hasBloomFilter(newLeft, l)) { - extractBeneficialFilterCreatePlan(left, right, l, r).foreach { creationSide => - newLeft = injectFilter(l, newLeft, creationSide) - } - } + tryInject(left, newLeft, l, right, r, canPruneLeft(joinType), + injectLeftHinted, injectRightHinted, "left").foreach(newLeft = _) // Did we actually inject on the left? If not, try on the right - // Check if: - // 1. The current join type supports prune the right side with runtime filter - // 2. The current join is a shuffle join or a broadcast join that - // has a shuffle below it - // 3. There is no bloom filter on the right key yet - if (newLeft.fastEquals(oldLeft) && canPruneRight(joinType) && - (hasShuffle || probablyHasShuffle(right)) && !hasBloomFilter(newRight, r)) { - extractBeneficialFilterCreatePlan(right, left, r, l).foreach { creationSide => - newRight = injectFilter(r, newRight, creationSide) - } + if (newLeft.fastEquals(oldLeft)) { + tryInject(right, newRight, r, left, l, canPruneRight(joinType), + injectRightHinted, injectLeftHinted, "right").foreach(newRight = _) } if (!newLeft.fastEquals(oldLeft) || !newRight.fastEquals(oldRight)) { filterCounter = filterCounter + 1 + appliedHint = appliedHint || hinted } } }) + if (hinted && !appliedHint) { + reportHintNotApplied(join, + notAppliedReason.getOrElse("no runtime filter could be built from the hinted side")) + } join.withNewChildren(Seq(newLeft, newRight)) + case join @ Join(_, _, _, _, hint) + if hintToRuntimeFilterSourceLeft(hint) || hintToRuntimeFilterSourceRight(hint) => + // A runtime filter is built from the join keys, so a join without equi-join keys has + // nothing to build from. + reportHintNotApplied(join, if (isRuntimeFilterHintAmbiguous(hint)) { + "the runtime filter source is ambiguous as both join sides are hinted" + } else { + "no equi-join keys" + }) + join } } + private def hasRuntimeFilterHint(hint: JoinHint): Boolean = { + hintToRuntimeFilterSourceLeft(hint) || hintToRuntimeFilterSourceRight(hint) + } + + // A HintInfo carries no relation name, so the join is identified by the hinted side and its + // condition. Only the runtime filter facet is reported. + private def reportHintNotApplied(join: Join, reason: String): Unit = { + val side = (hintToRuntimeFilterSourceLeft(join.hint), + hintToRuntimeFilterSourceRight(join.hint)) match { + case (true, true) => "both" + case (true, false) => "left" + case _ => "right" + } + val condition = join.condition.map(_.toString).getOrElse("none") + hintErrorHandler.joinHintNotSupported(HintInfo(runtimeFilterSource = true), + s"$reason (hinted side: $side, join condition: $condition)") + } + + private def hasRuntimeFilterHint(plan: LogicalPlan): Boolean = plan.exists { + case Join(_, _, _, _, hint) => hasRuntimeFilterHint(hint) + case _ => false + } + + // With Bloom filters disabled the rule still runs over a plan with a runtime filter hint, so the + // hint is credited to a DPP filter or reported as not applied. override def apply(plan: LogicalPlan): LogicalPlan = plan match { - case s: Subquery if s.correlated => plan - case _ if !conf.runtimeFilterBloomFilterEnabled => plan + case s: Subquery if s.correlated => + // Runtime filters are not injected inside a correlated subquery, so a hint there is + // reported rather than dropped silently. + s.foreach { + case join: Join if hasRuntimeFilterHint(join.hint) => + reportHintNotApplied(join, "the join is inside a correlated subquery") + case _ => + } + plan + case _ if !conf.runtimeFilterBloomFilterEnabled && !hasRuntimeFilterHint(plan) => plan case _ => tryInjectRuntimeFilter(plan) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index a1b650fd315d4..c2fd8925bf723 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -2168,7 +2168,7 @@ object EliminateSorts extends Rule[LogicalPlan] { case _ => false } - private def isOrderIrrelevantAggs(aggs: Seq[NamedExpression]): Boolean = { + private[optimizer] def isOrderIrrelevantAggs(aggs: Seq[NamedExpression]): Boolean = { def isOrderIrrelevantAggFunction(func: AggregateFunction): Boolean = func match { case _: Min | _: Max | _: Count | _: BitAggregate => true // Arithmetic operations for floating-point values are order-sensitive 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..0d9758b5bd396 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 @@ -18,13 +18,14 @@ package org.apache.spark.sql.catalyst.optimizer import scala.annotation.tailrec +import scala.util.{Left, Right} import scala.util.control.NonFatal import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys.{HASH_JOIN_KEYS, JOIN_CONDITION} import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression -import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, ExtractFiltersAndInnerJoins, ExtractSingleColumnNullAwareAntiJoin} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate} +import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, ExtractFiltersAndInnerJoins, ExtractSingleColumnNullAwareAntiJoin, NodeWithOnlyDeterministicProjectAndFilter} import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules._ @@ -579,6 +580,44 @@ trait JoinSelectionHelper extends Logging { hint.rightHint.exists(_.strategy.contains(NO_BROADCAST_AND_REPLICATION)) } + def hintToRuntimeFilterSourceLeft(hint: JoinHint): Boolean = { + hint.leftHint.exists(_.runtimeFilterSource) + } + + def hintToRuntimeFilterSourceRight(hint: JoinHint): Boolean = { + hint.rightHint.exists(_.runtimeFilterSource) + } + + /** + * The join side a [[RuntimeFilterHint]] names as the runtime filter source, i.e. the side a + * runtime filter is built from to prune the other side. `None` when neither side is hinted, and + * also when both are: each side would then have to be the other's source, so the hint is + * ambiguous and ignored, see [[isRuntimeFilterHintAmbiguous]]. + */ + def runtimeFilterSourceSide(hint: JoinHint): Option[BuildSide] = { + (hintToRuntimeFilterSourceLeft(hint), hintToRuntimeFilterSourceRight(hint)) match { + case (true, false) => Some(BuildLeft) + case (false, true) => Some(BuildRight) + case _ => None + } + } + + def isRuntimeFilterHintAmbiguous(hint: JoinHint): Boolean = { + hintToRuntimeFilterSourceLeft(hint) && hintToRuntimeFilterSourceRight(hint) + } + + /** + * Why `plan` cannot serve as the source of a runtime filter on its join key `key`, or None when + * it can, see [[RuntimeFilterSourceAnalysis]]. + */ + def runtimeFilterSourceRejection(plan: LogicalPlan, key: Expression): Option[String] = { + RuntimeFilterSourceAnalysis.rejection(plan, key) + } + + def isRepeatableRuntimeFilterSource(plan: LogicalPlan, key: Expression): Boolean = { + runtimeFilterSourceRejection(plan, key).isEmpty + } + private def getBuildSide( canBuildLeft: Boolean, canBuildRight: Boolean, @@ -644,3 +683,214 @@ private[sql] object NullAwareAntiJoinPlanning extends JoinSelectionHelper { } } } + +/** + * Decides whether a plan can serve as the source of a runtime filter on a join key. A runtime + * filter evaluates its source separately from the join, so the key values the source produces + * must be the same in both evaluations, or the filter could prune rows the join itself matches. + * `deterministic` is not enough for that: Spark flags order-dependent computations such as + * first, last, row_number or an unordered LIMIT as deterministic. + * + * The plan is walked bottom-up, tracking the output attributes whose values are unstable: they + * come from a non-deterministic expression, an order-dependent aggregate or window function, or + * an expression over such an attribute. The plan is rejected outright when its row set is + * unstable: a filter, join condition or grouping consumes an unstable attribute, an inner + * generate uses an unstable generator, a sample is unseeded or over anything but a scan, or a + * limit is over anything but a total order; and, with its own reason, when an operator's effect + * on the rows is not analyzed. The source qualifies when the key references no unstable + * attribute. Values that are unstable but only carried to the output (a `first(name)` next to a + * `GROUP BY id`, a row number next to the key) do not disqualify it. + */ +private[optimizer] object RuntimeFilterSourceAnalysis extends AliasHelper { + + /** + * @param unstable output attributes whose values depend on evaluation order or on chance. + * @param totallyOrdered whether the rows are in a total order on stable keys, so that a limit + * over them keeps the same rows every time. + */ + private case class Taint(unstable: AttributeSet, totallyOrdered: Boolean = false) + + private val NotRepeatable = + "the hinted side may produce different rows or join keys when evaluated again" + + /** Why `plan` is not a repeatable source of `key`, or None when it is. */ + def rejection(plan: LogicalPlan, key: Expression): Option[String] = { + if (plan.isStreaming) { + Some("the hinted side is a stream") + } else if (!key.deterministic) { + Some(NotRepeatable) + } else { + analyze(plan) match { + case Left(reason) => Some(reason) + case Right(t) if key.references.intersect(t.unstable).nonEmpty => Some(NotRepeatable) + case _ => None + } + } + } + + /** + * Whether `e` yields the same value on every evaluation. A subquery counts as deterministic + * when its plan is, which is the very check this analysis replaces, so its plan is analyzed + * too. + */ + private def isStable(e: Expression, unstable: AttributeSet): Boolean = { + e.deterministic && e.references.intersect(unstable).isEmpty && !e.exists { + case s: SubqueryExpression => s.plan.isStreaming || + analyze(s.plan).forall(t => s.plan.outputSet.intersect(t.unstable).nonEmpty) + case _ => false + } + } + + /** Returns the taint of `plan`'s output, or the reason its row set is not repeatable. */ + private def analyze(plan: LogicalPlan): Either[String, Taint] = plan match { + case _: LeafNode => Right(Taint(AttributeSet.empty)) + + case p: Project => analyze(p.child).map { t => + Taint( + AttributeSet(p.projectList.filterNot(isStable(_, t.unstable)).map(_.toAttribute)), + t.totallyOrdered) + } + + case f: Filter => analyze(f.child).flatMap { t => + if (isStable(f.condition, t.unstable)) Right(t) else Left(NotRepeatable) + } + + case j: Join => analyze(j.left).flatMap { l => + analyze(j.right).flatMap { r => + val unstable = l.unstable ++ r.unstable + if (j.condition.forall(isStable(_, unstable))) { + Right(Taint(unstable)) + } else { + Left(NotRepeatable) + } + } + } + + case a: Aggregate => analyze(a.child).map { t => + // Grouping on an unstable value changes which rows form a group, so every aggregate result + // then depends on it; a grouping expression's own value is as stable as its input. + val stableGroups = a.groupingExpressions.forall(isStable(_, t.unstable)) + val unstable = a.aggregateExpressions.filter { e => + !isStable(e, t.unstable) || + (e.exists(_.isInstanceOf[AggregateExpression]) && + (!stableGroups || !isOrderIrrelevantAggregate(e))) + } + Taint(AttributeSet(unstable.map(_.toAttribute))) + } + + case w: Window => analyze(w.child).map { t => + val stablePartitions = w.partitionSpec.forall(isStable(_, t.unstable)) + val unstable = w.windowExpressions.filter { e => + !stablePartitions || !isStable(e, t.unstable) || !isOrderIrrelevantWindow(e) + } + Taint(t.unstable ++ AttributeSet(unstable.map(_.toAttribute))) + } + + case u: Union => + val taints = u.children.map(analyze) + taints.collectFirst { case Left(reason) => Left(reason) }.getOrElse { + val unstable = u.output.zipWithIndex.collect { + case (attr, i) if u.children.zip(taints).exists { + case (child, Right(taint)) => taint.unstable.contains(child.output(i)) + case _ => false + } => attr + } + Right(Taint(AttributeSet(unstable))) + } + + // An inner generate drops the rows for which the generator yields nothing, so an unstable + // generator changes the row set; an outer generate keeps them. + case g: Generate => analyze(g.child).flatMap { t => + if (isStable(g.generator, t.unstable)) { + Right(t) + } else if (g.outer) { + Right(Taint(t.unstable ++ AttributeSet(g.generatorOutput))) + } else { + Left(NotRepeatable) + } + } + + case e: Expand => analyze(e.child).map { t => + val unstable = e.output.zipWithIndex.collect { + case (attr, i) if e.projections.exists(p => !isStable(p(i), t.unstable)) => attr + } + Taint(AttributeSet(unstable)) + } + + case s: Sort => analyze(s.child).map { t => + val stableOrder = s.order.forall(o => isStable(o.child, t.unstable)) + Taint(t.unstable, totallyOrdered = s.global && stableOrder && + sortedOnUniqueKey(s.child, s.order.map(_.child))) + } + + // A limit keeps whichever rows arrive first unless the order is total. + case l @ (_: GlobalLimit | _: LocalLimit | _: Offset | _: Tail) => + analyze(l.children.head).flatMap { t => + if (t.totallyOrdered) Right(t) else Left(NotRepeatable) + } + + // A sample draws a fresh seed per evaluation unless one is given, and depends on the input + // row order even then: only a seeded sample over a scan, through projections and filters + // that keep the row order, is repeatable. + case s: Sample => + val overScan = NodeWithOnlyDeterministicProjectAndFilter.unapply(s.child) + .exists(_.isInstanceOf[LeafNode]) + if (s.seed.isDefined && overScan) analyze(s.child) else Left(NotRepeatable) + + // The rows and their values are unchanged; a shuffle loses the order. + case _: Distinct | _: SubqueryAlias | _: Repartition | _: RepartitionByExpression | + _: RebalancePartitions => + analyze(plan.children.head).map(t => Taint(t.unstable)) + + // Observed metrics do not touch the rows. + case c: CollectMetrics => analyze(c.child) + + // Anything else, e.g. a typed operator or a script transformation, is not analyzed. + case _ => + Left(s"the hinted side contains ${plan.nodeName}, which cannot be checked for repeatability") + } + + /** + * Whether an aggregate expression's value is independent of the input order. Spark's own + * allowlist covers the SQL functions; a Bloom filter aggregate, which this rule injects for an + * inner join, merges commutatively. + */ + private def isOrderIrrelevantAggregate(e: NamedExpression): Boolean = e match { + case Alias(AggregateExpression(_: BloomFilterAggregate, _, _, _, _), _) => true + case _ => EliminateSorts.isOrderIrrelevantAggs(Seq(e)) + } + + /** + * A window function's value depends on the row order within its frame, unless the frame is + * the whole partition and the function is order-irrelevant. + */ + private def isOrderIrrelevantWindow(e: NamedExpression): Boolean = e match { + case Alias(WindowExpression(_: AggregateExpression, spec), _) => + val wholePartition = spec.orderSpec.isEmpty && (spec.frameSpecification match { + case UnspecifiedFrame => true + case SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing) => true + case _ => false + }) + wholePartition && EliminateSorts.isOrderIrrelevantAggs(Seq(e)) + case _ => false + } + + /** + * Whether `sortKeys` cover a key of `plan` that is proven unique, so that sorting on them is a + * total order. The only uniqueness Catalyst can establish is an aggregate's grouping keys. + */ + private def sortedOnUniqueKey(plan: LogicalPlan, sortKeys: Seq[Expression]): Boolean = { + plan match { + case p: Project => + val aliases = getAliasMap(p) + sortedOnUniqueKey(p.child, sortKeys.map(replaceAlias(_, aliases))) + case Filter(_, child) => sortedOnUniqueKey(child, sortKeys) + case a: Aggregate => + val aliases = getAliasMap(a) + val keys = sortKeys.map(replaceAlias(_, aliases)) + a.groupingExpressions.nonEmpty && + a.groupingExpressions.forall(g => keys.exists(_.semanticEquals(g))) + case _ => false + } + } +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala index dc8c4cf3dfa85..b5f587e753614 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala @@ -82,8 +82,13 @@ object JoinHint { * The hint attributes to be applied on a specific node. * * @param strategy The preferred join strategy. + * @param runtimeFilterSource Whether this side was hinted as a runtime filter source, i.e. the + * side a runtime filter should be built from to prune the other side + * of the join. */ -case class HintInfo(strategy: Option[JoinStrategyHint] = None) { +case class HintInfo( + strategy: Option[JoinStrategyHint] = None, + runtimeFilterSource: Boolean = false) { /** * Combine this [[HintInfo]] with another [[HintInfo]] and return the new [[HintInfo]]. @@ -95,17 +100,27 @@ case class HintInfo(strategy: Option[JoinStrategyHint] = None) { * [[HintInfo]] if defined, otherwise the strategy in the other [[HintInfo]]. The * `hintOverriddenCallback` will be called if this [[HintInfo]] and the other [[HintInfo]] * both have a strategy defined but the join strategies are different. + * + * A runtime filter hint is not a join strategy, so it composes with one rather than overriding + * it: the merged [[HintInfo]] is a runtime filter source if either side is. Only the strategy + * is reported as overridden, since only the strategy is. */ def merge(other: HintInfo, hintErrorHandler: HintErrorHandler): HintInfo = { if (this.strategy.isDefined && other.strategy.isDefined && this.strategy.get != other.strategy.get) { - hintErrorHandler.hintOverridden(other) + hintErrorHandler.hintOverridden(other.copy(runtimeFilterSource = false)) } - HintInfo(strategy = this.strategy.orElse(other.strategy)) + HintInfo( + strategy = this.strategy.orElse(other.strategy), + runtimeFilterSource = this.runtimeFilterSource || other.runtimeFilterSource) } - override def toString: String = strategy.map(s => s"(strategy=$s)").getOrElse("none") + override def toString: String = { + val fields = strategy.map(s => s"strategy=$s").toSeq ++ + Option.when(runtimeFilterSource)("runtime_filter_source") + if (fields.isEmpty) "none" else fields.mkString("(", ", ", ")") + } } sealed abstract class JoinStrategyHint { @@ -204,6 +219,21 @@ abstract class WindowHint abstract class SortHint +/** + * The relation-level hint that marks a join side as a runtime filter source, i.e. the side a + * runtime filter is built from to prune the other side of the join, e.g. + * "RUNTIME_FILTER(dim)". Unlike a [[JoinStrategyHint]] this does not choose a join + * implementation, so it composes with one; it also does not name a filtering mechanism, leaving + * Spark free to pick one. + */ +object RuntimeFilterHint { + val hintName: String = "RUNTIME_FILTER" + + def isRuntimeFilterHintName(name: String): Boolean = { + name.toUpperCase(Locale.ROOT) == hintName + } +} + /** * The callback for implementing customized strategies of handling hint errors. */ diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala index cd1d14e1bd6c1..7a58f2a9e5950 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala @@ -93,7 +93,7 @@ object RuleIdCollection { "org.apache.spark.sql.catalyst.analysis.ResolveGroupByAll" :: "org.apache.spark.sql.catalyst.analysis.ResolveHigherOrderFunctions" :: "org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveCoalesceHints" :: - "org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveJoinStrategyHints" :: + "org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveJoinHints" :: "org.apache.spark.sql.catalyst.analysis.ResolveInlineTables" :: "org.apache.spark.sql.catalyst.analysis.ResolveLambdaVariables" :: "org.apache.spark.sql.catalyst.analysis.ResolveLateralColumnAliasReference" :: diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala index 28f496ab1bd8b..149dc39e08b60 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala @@ -1148,7 +1148,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat messageParameters = Map.empty) } - def joinStrategyHintParameterNotSupportedError(unsupported: Expression): Throwable = { + def joinHintParameterNotSupportedError(unsupported: Expression): Throwable = { new AnalysisException( errorClass = "_LEGACY_ERROR_TEMP_1046", messageParameters = Map( diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveHintsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveHintsSuite.scala index 54b0827717c96..8b4f96f0dd0b2 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveHintsSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveHintsSuite.scala @@ -307,6 +307,39 @@ class ResolveHintsSuite extends AnalysisTest { } } + test("runtime filter hint resolution") { + // Resolves per relation, like a join strategy hint. + checkAnalysisWithoutViewWrapper( + UnresolvedHint("RUNTIME_FILTER", Seq("table2"), + table("TaBlE").join(table("TaBlE2"))), + Join( + testRelation, + ResolvedHint(testRelation2, HintInfo(runtimeFilterSource = true)), + Inner, + None, + JoinHint.NONE), + caseSensitive = false) + + // Composes with a join strategy hint on the same relation rather than overriding it. + checkAnalysisWithoutViewWrapper( + UnresolvedHint("RUNTIME_FILTER", Seq("table2"), + UnresolvedHint("MERGEJOIN", Seq("table2"), + table("TaBlE").join(table("TaBlE2")))), + Join( + testRelation, + ResolvedHint(testRelation2, + HintInfo(strategy = Some(SHUFFLE_MERGE), runtimeFilterSource = true)), + Inner, + None, + JoinHint.NONE), + caseSensitive = false) + + // Without parameters it applies to the whole subtree, like a join strategy hint. + checkAnalysisWithoutViewWrapper( + UnresolvedHint("RUNTIME_FILTER", Seq(), table("TaBlE")), + ResolvedHint(testRelation, HintInfo(runtimeFilterSource = true))) + } + test("SPARK-35786: Support optimize rebalance by expression in AQE") { checkAnalysisWithoutViewWrapper( UnresolvedHint("REBALANCE", Seq(UnresolvedAttribute("a")), table("TaBlE")), diff --git a/sql/connect/common/src/main/protobuf/spark/connect/relations.proto b/sql/connect/common/src/main/protobuf/spark/connect/relations.proto index 1517197437032..8fb0cea75a698 100644 --- a/sql/connect/common/src/main/protobuf/spark/connect/relations.proto +++ b/sql/connect/common/src/main/protobuf/spark/connect/relations.proto @@ -960,7 +960,8 @@ message Hint { // (Required) Hint name. // - // Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL. + // Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL, + // RUNTIME_FILTER. // // Supported partitioning hints include COALESCE, REPARTITION, REPARTITION_BY_RANGE. string name = 2; 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..04edec78f5d3b 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 @@ -200,8 +200,9 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { joinType: JoinType, hint: JoinHint, isBroadcast: Boolean): Unit = { + // Only the strategy is rejected here, so report only that facet of the hint. def invalidBuildSideInHint(hintInfo: HintInfo, buildSide: String): Unit = { - hintErrorHandler.joinHintNotSupported(hintInfo, + hintErrorHandler.joinHintNotSupported(hintInfo.copy(runtimeFilterSource = false), s"build $buildSide for ${joinType.sql.toLowerCase(Locale.ROOT)} join") } @@ -220,8 +221,9 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { private def checkHintNonEquiJoin(hint: JoinHint): Unit = { if (hintToShuffleHashJoin(hint) || hintToSortMergeJoin(hint)) { - assert(hint.leftHint.orElse(hint.rightHint).isDefined) - hintErrorHandler.joinHintNotSupported(hint.leftHint.orElse(hint.rightHint).get, + val strategyHint = hint.leftHint.filter(_.strategy.isDefined).orElse(hint.rightHint) + assert(strategyHint.isDefined) + hintErrorHandler.joinHintNotSupported(strategyHint.get.copy(runtimeFilterSource = false), "no equi-join keys") } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala index c0747cbfb1130..273d5b4c9c775 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.dynamicpruning import org.apache.spark.sql.catalyst.catalog.HiveTableRelation import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.optimizer.{JoinSelectionHelper, ReusableBroadcastValueProjection} +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, JoinSelectionHelper, ReusableBroadcastValueProjection} import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule @@ -103,7 +103,8 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join filteringKeys: Seq[Expression], filteringPlan: LogicalPlan, joinKeys: Seq[Expression], - partScan: LogicalPlan): LogicalPlan = { + partScan: LogicalPlan, + hinted: Boolean): LogicalPlan = { val reuseEnabled = conf.exchangeReuseEnabled require(filteringKeys.size == 1, "DPP Filters should only have a single broadcasting key " + "since there are no usage for multiple broadcasting keys at the moment.") @@ -113,9 +114,17 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join } else { None } - lazy val hasBenefit = pruningHasBenefit( + // A [[RuntimeFilterHint]] on the filtering side asserts the benefit `pruningHasBenefit` + // estimates, so take it as given rather than guessing from statistics and filter ratios. + lazy val hasBenefit = hinted || pruningHasBenefit( pruningKey, partScan, filteringKeys.head, filteringPlan, hasSelectivePredicate(filteringPlan)) if (reuseEnabled || hasBenefit) { + // `reuseBroadcastOnly` keeps DPP from re-executing the filtering side unless a broadcast + // can be reused, i.e. unless pruning is free. That is a cost bound, and the cost is what + // a [[RuntimeFilterHint]] asks to spend: a hinted filter is applied whether a broadcast + // turns up, rather than degrading to `true` in [[PlanDynamicPruningFilters]]. + val onlyInBroadcast = !hinted && + (conf.dynamicPartitionPruningReuseBroadcastOnly || !hasBenefit) // insert a DynamicPruning wrapper to identify the subquery during query planning Filter( DynamicPruningSubquery( @@ -123,7 +132,7 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join filteringPlan, joinKeys, indices, - conf.dynamicPartitionPruningReuseBroadcastOnly || !hasBenefit)(broadcastValueProjection), + onlyInBroadcast)(broadcastValueProjection), pruningPlan) } else { // abort dynamic partition pruning @@ -306,10 +315,22 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join * meet the following requirements: * (1) it can not be a stream * (2) it needs to contain a selective predicate or a cheaply-recomputable materialized input + * + * (2) is evidence that pruning pays off, which a [[RuntimeFilterHint]] on the filtering side + * (`hinted`) supplies directly. A hinted side only has to be a repeatable source of the + * filtering key, see `JoinSelectionHelper.isRepeatableRuntimeFilterSource`, since DPP + * re-evaluates it. */ - private def hasPartitionPruningFilter(plan: LogicalPlan): Boolean = { - !plan.isStreaming && - (hasSelectivePredicate(plan) || isCheaplyRecomputableMaterializedPlan(plan)) + private def hasPartitionPruningFilter( + plan: LogicalPlan, + hinted: Boolean, + filteringKey: Expression): Boolean = { + if (hinted) { + isRepeatableRuntimeFilterSource(plan, filteringKey) + } else { + !plan.isStreaming && + (hasSelectivePredicate(plan) || isCheaplyRecomputableMaterializedPlan(plan)) + } } private def prune(plan: LogicalPlan): LogicalPlan = { @@ -321,6 +342,13 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join var newLeft = left var newRight = right + // A side hinted as the runtime filter source is the filtering side, so the other side is + // the one pruned, and the hinted side is never itself pruned. An ambiguous hint is ignored + // here and reported by [[InjectRuntimeFilter]], the single place that warns about it. + val hintedSource = runtimeFilterSourceSide(hint) + val pruneLeftHinted = hintedSource.contains(BuildRight) + val pruneRightHinted = hintedSource.contains(BuildLeft) + // extract the left and right keys of the join condition val (leftKeys, rightKeys) = j match { case ExtractEquiJoinKeys(_, lkeys, rkeys, _, _, _, _, _) => (lkeys, rkeys) @@ -348,14 +376,16 @@ object PartitionPruning extends Rule[LogicalPlan] with PredicateHelper with Join // there should be a partitioned table and a filter on the dimension table, // otherwise the pruning will not trigger var filterableScan = getFilterableTableScan(l, left) - if (filterableScan.isDefined && canPruneLeft(joinType) && - hasPartitionPruningFilter(right)) { - newLeft = insertPredicate(l, newLeft, Seq(r), right, rightKeys, filterableScan.get) + if (filterableScan.isDefined && canPruneLeft(joinType) && !pruneRightHinted && + hasPartitionPruningFilter(right, pruneLeftHinted, r)) { + newLeft = insertPredicate( + l, newLeft, Seq(r), right, rightKeys, filterableScan.get, pruneLeftHinted) } else { filterableScan = getFilterableTableScan(r, right) - if (filterableScan.isDefined && canPruneRight(joinType) && - hasPartitionPruningFilter(left) ) { - newRight = insertPredicate(r, newRight, Seq(l), left, leftKeys, filterableScan.get) + if (filterableScan.isDefined && canPruneRight(joinType) && !pruneLeftHinted && + hasPartitionPruningFilter(left, pruneRightHinted, l)) { + newRight = insertPredicate( + r, newRight, Seq(l), left, leftKeys, filterableScan.get, pruneRightHinted) } } case _ => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala index 525e485f0afae..74e5a639ae9a9 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala @@ -513,8 +513,8 @@ class DataFrameJoinSuite extends SharedSparkSession val statement = s"SELECT /*+ BROADCASTJOIN(t) */ * FROM $db1Name.t, $db2Name.t " + s"WHERE $db1Name.t.id = $db2Name.t.id" sql(statement).queryExecution.optimizedPlan match { - case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST))), - Some(HintInfo(Some(BROADCAST))))) => + case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST), _)), + Some(HintInfo(Some(BROADCAST), _)))) => case _ => fail("broadcast hint not found in both tables") } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala index e09a170734bbb..b402471d4605e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala @@ -2151,6 +2151,86 @@ abstract class DynamicPartitionPruningV1Suite extends DynamicPartitionPruningDat } } + test("RUNTIME_FILTER hint injects a standalone DPP subquery without an estimated benefit") { + // `dim_store` carries no selective predicate, so DPP finds no evidence of a pruning benefit + // and, with no broadcast to reuse, injects nothing. The hint asserts the benefit, so the same + // join gets a standalone DPP subquery -- partition pruning, not a row-level filter -- with + // `reuseBroadcastOnly` at its default. + withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + val unhinted = sql( + "SELECT f.date_id, f.store_id FROM fact_sk f JOIN dim_store s ON f.store_id = s.store_id") + checkPartitionPruningPredicate(unhinted, withSubquery = false, withBroadcast = false) + + val hinted = sql( + """SELECT /*+ RUNTIME_FILTER(s) */ f.date_id, f.store_id + |FROM fact_sk f JOIN dim_store s ON f.store_id = s.store_id""".stripMargin) + checkPartitionPruningPredicate(hinted, withSubquery = true, withBroadcast = false) + checkAnswer(hinted, unhinted.collect().toSeq) + } + } + + test("RUNTIME_FILTER hint still reuses a broadcast when one is available") { + // Overriding `reuseBroadcastOnly` must not stop the cheaper option being taken: with a + // broadcast hash join the DPP filter reuses that broadcast rather than duplicating the + // filtering side into a standalone subquery. + withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true") { + val hinted = sql( + """SELECT /*+ RUNTIME_FILTER(s), BROADCAST(s) */ f.date_id, f.store_id + |FROM fact_sk f JOIN dim_store s ON f.store_id = s.store_id""".stripMargin) + checkPartitionPruningPredicate(hinted, withSubquery = false, withBroadcast = true) + checkAnswer(hinted, + sql("SELECT date_id, store_id FROM fact_sk WHERE store_id IN (SELECT store_id FROM " + + "dim_store)").collect().toSeq) + } + } + + test("RUNTIME_FILTER hint keeps DPP's partition-key requirement") { + withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // `fact_np` is not partitioned, so partition pruning cannot apply however strongly the user + // asserts a benefit. The hint must not manufacture a DPP filter here. + val hinted = sql( + """SELECT /*+ RUNTIME_FILTER(s) */ f.date_id, f.store_id + |FROM fact_np f JOIN dim_store s ON f.store_id = s.store_id""".stripMargin) + checkPartitionPruningPredicate(hinted, withSubquery = false, withBroadcast = false) + checkAnswer(hinted, + sql("SELECT date_id, store_id FROM fact_np WHERE store_id IN (SELECT store_id FROM " + + "dim_store)").collect().toSeq) + } + } + + test("RUNTIME_FILTER hint keeps DPP's repeatable-source requirement") { + withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // DPP re-evaluates the filtering side, so a sample without a seed, which draws different + // rows per evaluation, is not accepted as a source however strongly the user asserts a + // benefit. The same predicate gates the row-level filter. + val hinted = sql( + """SELECT /*+ RUNTIME_FILTER(s) */ f.date_id, f.store_id + |FROM fact_sk f JOIN (SELECT store_id FROM dim_store TABLESAMPLE (50 PERCENT)) s + |ON f.store_id = s.store_id""".stripMargin) + checkPartitionPruningPredicate(hinted, withSubquery = false, withBroadcast = false) + } + } + + test("RUNTIME_FILTER hint on both join sides leaves DPP to its own estimates") { + // The confs and query of "simple inner join triggers DPP with mock-up tables": an ambiguous + // hint is ignored, so the selective predicate on `dim_store` still gets DPP as without a hint, + // rather than each hinted side blocking the other. + withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false", + SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") { + val hinted = sql( + """ + |SELECT /*+ RUNTIME_FILTER(f, s) */ f.date_id, f.store_id FROM fact_sk f + |JOIN dim_store s ON f.store_id = s.store_id AND s.country = 'NL' + """.stripMargin) + checkPartitionPruningPredicate(hinted, withSubquery = true, withBroadcast = false) + checkAnswer(hinted, Row(1000, 1) :: Row(1010, 2) :: Row(1020, 2) :: Nil) + } + } + test("SPARK-54593: a materialized filtering side keeps statistics-backed standalone DPP") { // A checkpoint-derived LogicalRDD has no Filter but can retain column statistics. When those // statistics establish a pruning benefit, DPP must still be injected as a standalone subquery diff --git a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala index beabadd178295..17077eadf3915 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala @@ -19,14 +19,17 @@ package org.apache.spark.sql import java.io.File -import org.apache.spark.sql.catalyst.expressions.{Alias, BloomFilterMightContain, Literal, ScalarSubquery} +import org.apache.logging.log4j.Level + +import org.apache.spark.sql.catalyst.expressions.{Alias, BloomFilterMightContain, DynamicPruningSubquery, Literal, NamedExpression, ScalarSubquery, XxHash64} import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate} -import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan} +import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join, LogicalPlan, SHUFFLE_MERGE} import org.apache.spark.sql.columnar.CachedBatch import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec} import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, AQEPropagateEmptyRelation} import org.apache.spark.sql.execution.columnar.InMemoryRelation import org.apache.spark.sql.execution.planmerging.MergeSubplans +import org.apache.spark.sql.functions.udf import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types.{IntegerType, StructType} @@ -256,19 +259,31 @@ class InjectRuntimeFilterSuite extends SharedSparkSession } def getNumBloomFilters(plan: LogicalPlan): Integer = { + // Counts the Bloom filter aggregates at the root of each Bloom filter subquery, where every + // aggregate expression must be one. Aggregates below the root belong to a hinted creation + // side (e.g. a `SELECT DISTINCT` one) and are not visited; a subquery whose root aggregate + // has no Bloom filter aggregate is the query's own. + def isBloomFilterAgg(e: NamedExpression): Boolean = e.exists { + case AggregateExpression(_: BloomFilterAggregate, _, _, _, _) => true + case _ => false + } + def countBloomFilterAggs(plan: LogicalPlan): Int = plan match { + case Aggregate(_, aggregateExpressions, _, _) + if aggregateExpressions.exists(isBloomFilterAgg) => + aggregateExpressions.map { + case Alias(AggregateExpression(bfAgg : BloomFilterAggregate, _, _, _, _), + _) => + assert(bfAgg.estimatedNumItemsExpression.isInstanceOf[Literal]) + assert(bfAgg.numBitsExpression.isInstanceOf[Literal]) + 1 + }.sum + case _: Aggregate => 0 + case other => other.children.map(countBloomFilterAggs).sum + } val numBloomFilterAggs = plan.collect { case Filter(condition, _) => condition.collect { - case subquery: org.apache.spark.sql.catalyst.expressions.ScalarSubquery - => subquery.plan.collect { - case Aggregate(_, aggregateExpressions, _, _) => - aggregateExpressions.map { - case Alias(AggregateExpression(bfAgg : BloomFilterAggregate, _, _, _, _), - _) => - assert(bfAgg.estimatedNumItemsExpression.isInstanceOf[Literal]) - assert(bfAgg.numBitsExpression.isInstanceOf[Literal]) - 1 - }.sum - }.sum + case subquery: org.apache.spark.sql.catalyst.expressions.ScalarSubquery => + countBloomFilterAggs(subquery.plan) }.sum }.sum val numMightContains = plan.collect { @@ -302,6 +317,46 @@ class InjectRuntimeFilterSuite extends SharedSparkSession checkWithAndWithoutFeatureEnabled(query, shouldReplace = false) } + /** + * Returns the join keys the query's runtime Bloom filters are applied on. The attribute names in + * this suite's tables are unique per table (`c1` in `bf1`, `c2` in `bf2`, ...), so this + * identifies which side a filter was applied to. + */ + def bloomFilterApplicationSideKeys(query: String): Set[String] = { + sql(query).queryExecution.optimizedPlan.collect { + case Filter(condition, _) => condition.collect { + case BloomFilterMightContain(_, XxHash64(Seq(key), _)) => key.references.map(_.name) + }.flatten + }.flatten.toSet + } + + /** Optimizes `query` and returns the warnings the hint error handler logged while doing so. */ + def hintWarnings(query: String): Seq[String] = { + val logAppender = new LogAppender("runtime filter hint") + withLogAppender(logAppender, level = Some(Level.WARN)) { + sql(query).queryExecution.optimizedPlan + } + logAppender.loggingEvents.map(_.getMessage.getFormattedMessage).toSeq + } + + def assertHintNotApplied(query: String, reason: String): Unit = { + val warnings = hintWarnings(query) + assert(warnings.exists(_.contains(s"is not supported in the query: $reason")), + s"expected a warning mentioning '$reason', got: $warnings") + } + + /** Asserts `query` gets no runtime Bloom filter, without executing it. */ + def assertNoBloomFilters(query: String): Unit = { + assert(getNumBloomFilters(sql(query).queryExecution.optimizedPlan) == 0) + } + + def hasDynamicPruning(query: String): Boolean = { + sql(query).queryExecution.optimizedPlan.exists { + case Filter(condition, _) => condition.exists(_.isInstanceOf[DynamicPruningSubquery]) + case _ => false + } + } + test("SPARK-58272: safely use fully materialized selectively filtered caches") { val cacheName = "cached_bloom_filter_keys" val query = s"SELECT * FROM bf1 JOIN $cacheName ON bf1.c1 = $cacheName.c2" @@ -1270,6 +1325,433 @@ class InjectRuntimeFilterSuite extends SharedSparkSession } } + test("RUNTIME_FILTER hint waives the creation side's selective predicate requirement") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + val query = "select * from bf1 join bf2 on bf1.c1 = bf2.c2" + // Without a selective predicate on either side there is no evidence a filter pays off. + assertDidNotRewriteWithBloomFilter(query) + // The hint supplies that evidence, and picks which side to build from. + assertRewroteWithBloomFilter( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") + assert(bloomFilterApplicationSideKeys( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") == Set("c1")) + assert(bloomFilterApplicationSideKeys( + "select /*+ RUNTIME_FILTER(bf1) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") == Set("c2")) + } + } + + test("RUNTIME_FILTER hint waives the application side's scan size threshold") { + // A threshold above the application side's scan size, so only the hint can trigger a rewrite. + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1GB", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + assertDidNotRewriteWithBloomFilter( + "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62") + assertRewroteWithBloomFilter( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") + } + } + + test("RUNTIME_FILTER hint waives the creation side's size threshold") { + // A threshold below any creation side's size, so only the hint can trigger a rewrite. The + // threshold estimates whether the filter is worth building; the filter's own size is bounded + // by the max number of bits regardless. + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.RUNTIME_BLOOM_FILTER_CREATION_SIDE_THRESHOLD.key -> "1", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + assertDidNotRewriteWithBloomFilter( + "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62") + assertRewroteWithBloomFilter( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") + } + } + + test("RUNTIME_FILTER hint waives the shuffle requirement") { + // `bf2` is below the broadcast threshold, so this is a broadcast join with no shuffle below + // the application side, where a filter is not expected to pay off. + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000") { + assertDidNotRewriteWithBloomFilter( + "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62") + assertRewroteWithBloomFilter( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2") + } + } + + test("RUNTIME_FILTER hint waives the application side's single-leaf lineage requirement") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // The application side's key is an aggregate result, which cannot be traced to a scan. The + // heuristics use that lineage to predict that the filter reaches a scan; the filter above + // the aggregate is valid regardless, and prunes the aggregate's output before the join. + val aggregatedApplicationSide = + """ + |FROM (SELECT max(c1) AS c1 FROM bf1 GROUP BY b1) t + | JOIN bf2 + | ON t.c1 = bf2.c2 + """.stripMargin + assertDidNotRewriteWithBloomFilter(s"SELECT * $aggregatedApplicationSide WHERE bf2.a2 = 62") + assertRewroteWithBloomFilter( + s"SELECT /*+ RUNTIME_FILTER(bf2) */ * $aggregatedApplicationSide") + } + } + + test("RUNTIME_FILTER hint accepts a creation side the heuristics reject") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // Left to itself the rule looks for a selective predicate over a scan, a search an + // `Aggregate` stops, so this is not rewritten despite the selective predicate below it. The + // hint designates the creation side instead of searching for one, so the shape stops + // mattering. + val distinctSource = + """ + |FROM bf1 + | JOIN (SELECT DISTINCT c2 FROM bf2 WHERE a2 = 62) t + | ON bf1.c1 = t.c2 + """.stripMargin + assertDidNotRewriteWithBloomFilter(s"SELECT * $distinctSource") + assertRewroteWithBloomFilter(s"SELECT /*+ RUNTIME_FILTER(t) */ * $distinctSource") + // Likewise for a `Limit` the search does not descend through, here a top-n of grouping + // keys, whose order is total because the keys are unique. + assertRewroteWithBloomFilter( + """ + |SELECT /*+ RUNTIME_FILTER(t) */ * + |FROM bf1 + | JOIN (SELECT c2 FROM bf2 WHERE a2 = 62 GROUP BY c2 ORDER BY c2 LIMIT 5) t + | ON bf1.c1 = t.c2 + """.stripMargin) + } + } + + test("RUNTIME_FILTER hint builds a filter from a computed join key") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // The filter is built from the hinted side's output, so a join key the creation side computes + // rather than reads from a scan -- here an aggregate result -- works like any other. + assertRewroteWithBloomFilter( + """ + |SELECT /*+ RUNTIME_FILTER(t) */ * + |FROM bf1 + | JOIN (SELECT max(c2) AS c2 FROM bf2 GROUP BY b2) t + | ON bf1.c1 = t.c2 + """.stripMargin) + } + } + + test("RUNTIME_FILTER hint takes effect through one mechanism") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") { + // `bf5part` is partitioned by `f5`, and the selective `bf2.a2 = 62` makes DPP prune it on + // that key. The hint is honored by that DPP filter, so no Bloom filter is added: not on the + // key DPP already prunes, matching the unhinted behavior asserted above, and not on the + // other join key either, where the heuristics would add one ("add bloom filter if dpp + // filter exists on a different column" above). + val oneKey = "select /*+ RUNTIME_FILTER(bf2) */ * from bf5part join bf2 " + + "on bf5part.f5 = bf2.c2 where bf2.a2 = 62" + assertDidNotRewriteWithBloomFilter(oneKey) + assert(hasDynamicPruning(oneKey)) + val twoKeys = "select /*+ RUNTIME_FILTER(bf2) */ * from bf5part join bf2 " + + "on bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62" + assertDidNotRewriteWithBloomFilter(twoKeys) + assert(hasDynamicPruning(twoKeys)) + } + } + + test("RUNTIME_FILTER hint is honored by a DPP filter only when that filter reaches the scan") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + spark.udf.register("bf_nondeterministic_true", udf((_: Int) => true).asNondeterministic()) + // The hint gets a DPP filter on `bf5part`'s partition key, but a barrier on the pruned side + // keeps it from being pushed to the scan, so it is dropped later: a non-deterministic + // filter, or a window whose partitioning does not cover the key. The Bloom filter needs no + // pushdown, so the hint is honored with one instead, and exactly one of the two remains. + Seq( + "SELECT f5 FROM bf5part WHERE bf_nondeterministic_true(f5)", + "SELECT f5 FROM (SELECT f5, row_number() OVER (PARTITION BY c5 ORDER BY f5) rn " + + "FROM bf5part) WHERE rn = 1" + ).foreach { prunedSide => + val query = "SELECT /*+ RUNTIME_FILTER(bf2) */ f.f5, bf2.c2 " + + s"FROM ($prunedSide) f JOIN bf2 ON f.f5 = bf2.c2" + assertRewroteWithBloomFilter(query) + assert(!hasDynamicPruning(query)) + } + // A DPP filter that does reach the scan honors the hint on its own, even when the pruned + // side is not deterministic elsewhere (here on the other side of a join the DPP filter is + // pushed past), so no Bloom filter is added on top of it. + val reaching = "SELECT /*+ RUNTIME_FILTER(bf2) */ f.f5, f.nd, bf2.c2 FROM " + + "(SELECT p.f5, b.nd FROM bf5part p JOIN " + + "(SELECT c3, bf_nondeterministic_true(a3) AS nd FROM bf3) b ON p.c5 = b.c3) f " + + "JOIN bf2 ON f.f5 = bf2.c2" + assertDidNotRewriteWithBloomFilter(reaching) + assert(hasDynamicPruning(reaching)) + } + } + + test("RUNTIME_FILTER hint is ignored for a creation side that is not repeatable") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // The filter subquery re-executes the creation side, so one that can produce different rows + // or keys on re-evaluation could yield keys the join itself never sees and prune rows that + // do match. Catalyst flags most of these as deterministic. + val reason = "the hinted side may produce different rows or join keys when evaluated again" + spark.udf.register("bf_nondeterministic_array", + udf((i: Int) => Seq(i)).asNondeterministic()) + def query(source: String): String = + s"SELECT /*+ RUNTIME_FILTER(t) */ * FROM bf1 JOIN ($source) t ON bf1.c1 = t.c2" + Seq( + // A non-deterministic expression. + "SELECT c2 FROM bf2 WHERE rand() < 0.5", + // A limit keeps whichever rows arrive first: unordered, or ordered on a key that is not + // unique, so that ties at the cutoff can be resolved differently. + "SELECT c2 FROM bf2 WHERE a2 = 62 LIMIT 5", + "SELECT c2 FROM bf2 WHERE a2 = 62 ORDER BY c2 LIMIT 5", + // A sample without a seed draws a different sample per evaluation. + "SELECT c2 FROM bf2 TABLESAMPLE (50 PERCENT)", + // A key computed by an order-dependent aggregate function. + "SELECT first(c2) AS c2 FROM bf2 GROUP BY b2", + "SELECT any_value(c2) AS c2 FROM bf2 GROUP BY b2", + // A key computed by a window function. + "SELECT row_number() OVER (ORDER BY c2) AS c2 FROM bf2", + // An inner generate drops the rows for which an unstable generator yields nothing. + "SELECT c2 FROM bf2 LATERAL VIEW explode(bf_nondeterministic_array(a2)) x AS v", + // A key computed by a subquery whose own plan is not repeatable. + "SELECT (SELECT max(c2) FROM bf2 TABLESAMPLE (50 PERCENT)) AS c2 FROM bf2" + ).foreach { source => + assertNoBloomFilters(query(source)) + assertHintNotApplied(query(source), reason) + } + // A seeded sample over a scan is repeatable. + assertRewroteWithBloomFilter( + query("SELECT c2 FROM bf2 TABLESAMPLE (50 PERCENT) REPEATABLE (42)")) + } + } + + test("RUNTIME_FILTER hint accepts a creation side whose unstable values do not reach the key") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // Only the key values and the row set have to be repeatable. An order-dependent value that + // is merely carried to the output, next to a stable key, does not disqualify the side, and + // neither does a filter on a window aggregate whose frame is the whole partition. + def query(source: String, key: String): String = + s"SELECT /*+ RUNTIME_FILTER(t) */ * FROM bf1 JOIN ($source) t ON bf1.c1 = t.$key" + spark.udf.register("bf_nondeterministic_array", + udf((i: Int) => Seq(i)).asNondeterministic()) + Seq( + ("SELECT b2, first(c2) AS c2 FROM bf2 GROUP BY b2", "b2"), + ("SELECT c2, v FROM bf2 LATERAL VIEW OUTER explode(bf_nondeterministic_array(a2)) x AS v", + "c2"), + ("SELECT (SELECT max(c2) FROM bf2) AS c2, b2 FROM bf2", "c2"), + ("SELECT c2, row_number() OVER (ORDER BY a2) AS rn FROM bf2", "c2"), + ("SELECT c2, sum(a2) OVER (PARTITION BY b2) AS s FROM bf2", "c2"), + ("SELECT c2 FROM (SELECT c2, sum(a2) OVER (PARTITION BY b2) AS s FROM bf2) WHERE s > 0", + "c2") + ).foreach { case (source, key) => + assertRewroteWithBloomFilter(query(source, key)) + } + // Once an unstable value decides which rows survive, the row set is not repeatable. + val reason = "the hinted side may produce different rows or join keys when evaluated again" + Seq( + ("SELECT c2 FROM (SELECT c2, row_number() OVER (PARTITION BY b2 ORDER BY a2) AS rn " + + "FROM bf2) WHERE rn = 1", "c2"), + ("SELECT b2 FROM bf2 GROUP BY b2 HAVING first(c2) > 0", "b2") + ).foreach { case (source, key) => + assertNoBloomFilters(query(source, key)) + assertHintNotApplied(query(source, key), reason) + } + } + } + + test("RUNTIME_FILTER hint keeps correctness requirements") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // A left outer join may not have its left side pruned: dropping a left row would drop the + // null-extended output row it is entitled to. + val leftOuter = + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 left outer join bf2 on bf1.c1 = bf2.c2" + assertDidNotRewriteWithBloomFilter(leftOuter) + assertHintNotApplied(leftOuter, "the left side of a left outer join cannot be pruned") + // The mirror case is allowed: the right side of a left outer join is prunable. + assertRewroteWithBloomFilter( + "select /*+ RUNTIME_FILTER(bf1) */ * from bf1 left outer join bf2 on bf1.c1 = bf2.c2") + // Non-simple expressions stay excluded, as the filter subquery would recompute them. + spark.udf.register("bf_square", (s: Long) => s * s) + val udfKey = + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf_square(bf2.c2)" + assertDidNotRewriteWithBloomFilter(udfKey) + assertHintNotApplied(udfKey, "the join key is not a simple expression") + } + } + + test("RUNTIME_FILTER hint accepts a creation side that carries a runtime filter of its own") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // The inner join gets a Bloom filter on `bf1.c1` from the selective `bf2` on its own. The + // hinted outer join uses that join as its source; the filter it carries is a stable value, + // so a second filter is built from the source for `bf3`. + val query = "select /*+ RUNTIME_FILTER(t) */ * from bf3 join " + + "(select bf1.c1 from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 5) t " + + "on bf3.c3 = t.c1" + assertRewroteWithBloomFilter(query, 2) + assert(bloomFilterApplicationSideKeys(query) == Set("c1", "c3")) + } + } + + test("RUNTIME_FILTER hint reports a creation side it cannot analyze") { + withTempView("bf2_typed") { + spark.table("bf2").filter((_: Row) => true).createOrReplaceTempView("bf2_typed") + assertHintNotApplied( + "select /*+ RUNTIME_FILTER(bf2_typed) */ * from bf1 join bf2_typed " + + "on bf1.c1 = bf2_typed.c2", + "the hinted side contains TypedFilter, which cannot be checked for repeatability") + } + } + + test("RUNTIME_FILTER hint that cannot be applied does not fall back to the other direction") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // Left to itself the rule builds a filter from the selective `bf2` and applies it to `t`. + val join = "FROM (SELECT c1 FROM bf1 ORDER BY c1 LIMIT 5) t JOIN bf2 ON t.c1 = bf2.c2 " + + "WHERE bf2.a2 = 62" + assertRewroteWithBloomFilter(s"SELECT * $join") + // Hinting `t` as the source designates the opposite direction. `t` is not a repeatable + // source, so the hint fails, and the filter the heuristics would have built is not built + // either: the hint decides the direction rather than adding to the heuristics. + val hinted = s"SELECT /*+ RUNTIME_FILTER(t) */ * $join" + assertDidNotRewriteWithBloomFilter(hinted) + assertHintNotApplied(hinted, + "the hinted side may produce different rows or join keys when evaluated again") + } + } + + test("RUNTIME_FILTER hint inside a correlated subquery is reported") { + assertHintNotApplied( + "select * from bf1 where exists (select /*+ RUNTIME_FILTER(bf3) */ 1 from bf2 join bf3 " + + "on bf2.c2 = bf3.c3 where bf2.a2 = bf1.a1)", + "the join is inside a correlated subquery") + } + + test("RUNTIME_FILTER hint never filters the hinted side") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") { + // Left to itself the rule builds a filter from the selective `bf2` and applies it to `bf1`. + // Hinting `bf1` as the source designates the opposite direction, and the hinted side is not + // filtered even though it would have been without the hint. + val query = "select /*+ RUNTIME_FILTER(bf1) */ * from bf1 join bf2 on bf1.c1 = bf2.c2 " + + "where bf2.a2 = 62" + assertRewroteWithBloomFilter(query) + assert(bloomFilterApplicationSideKeys(query) == Set("c2")) + } + } + + test("RUNTIME_FILTER hint composes with a join strategy hint") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + val query = + "select /*+ MERGE(bf1, bf2), RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2" + assertRewroteWithBloomFilter(query) + val plan = sql(query).queryExecution.optimizedPlan + val hints = plan.collect { case Join(_, _, _, _, hint) => hint } + assert(hints.size == 1) + // Both hints survive on the same side: the strategy hint is not overridden by the runtime + // filter hint, nor the other way around. + assert(hints.head.leftHint.exists(_.strategy.contains(SHUFFLE_MERGE))) + assert(hints.head.rightHint.exists(_.strategy.contains(SHUFFLE_MERGE))) + assert(hints.head.rightHint.exists(_.runtimeFilterSource)) + assert(!hints.head.leftHint.exists(_.runtimeFilterSource)) + // A warning about one hint does not name the other: overriding the strategy leaves the + // runtime filter hint in effect, and rejecting the runtime filter hint leaves the strategy. + val warnings = hintWarnings("select /*+ BROADCAST(bf2), MERGE(bf2), RUNTIME_FILTER(bf2) */ " + + "* from bf1 left outer join bf2 on bf1.c1 = bf2.c2") + assert(warnings.exists(_.contains("Hint (strategy=merge) is overridden"))) + assert(warnings.exists(_.contains("Hint (runtime_filter_source) is not supported"))) + assert(!warnings.exists(_.contains("(strategy=merge, runtime_filter_source)"))) + } + } + + test("RUNTIME_FILTER hint on both join sides is ambiguous and ignored") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") { + // Each side would have to be the other's creation side, so the hint is dropped and the + // heuristics decide: the selective `bf2` still yields a filter on `bf1`, as without a hint. + val query = "select /*+ RUNTIME_FILTER(bf1, bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2 " + + "where bf2.a2 = 62" + assertRewroteWithBloomFilter(query) + assert(bloomFilterApplicationSideKeys(query) == Set("c1")) + assertHintNotApplied(query, "the runtime filter source is ambiguous") + // The same on a join without equi-join keys. + withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "true") { + assertHintNotApplied( + "select /*+ RUNTIME_FILTER(bf1, bf2) */ * from bf1 join bf2 on bf1.c1 > bf2.c2", + "the runtime filter source is ambiguous") + } + } + } + + test("RUNTIME_FILTER hint warns with the reason when it cannot be applied") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") { + // A filter on the key was already injected at the inner join, so the hinted one is not + // built; the warning says so rather than blaming the hinted side. + val existing = "select /*+ RUNTIME_FILTER(bf3) */ * from bf1 join bf2 on bf1.c1 = bf2.c2 " + + "join bf3 on bf3.c3 = bf1.c1 where bf2.a2 = 5" + assertRewroteWithBloomFilter(existing) + assertHintNotApplied(existing, "a runtime filter on the join key already exists") + // The filter count limit applies to hinted filters too. + withSQLConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD.key -> "1") { + assertRewroteWithBloomFilter(existing) + assertHintNotApplied(existing, + s"${SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD.key} (1) is reached") + } + // With Bloom filters disabled the hint is still reported rather than silently dropped. + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "false") { + assertHintNotApplied( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 = bf2.c2", + s"${SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key} is false") + } + } + } + + test("RUNTIME_FILTER hint on a relation that is not part of a join is reported") { + val logAppender = new LogAppender("runtime filter hint") + withLogAppender(logAppender, level = Some(Level.WARN)) { + sql("select /*+ RUNTIME_FILTER(bf1) */ * from bf1").queryExecution.optimizedPlan + } + assert(logAppender.loggingEvents.exists(_.getMessage.getFormattedMessage.contains( + "runtime_filter_source) is specified but it is not part of a join relation"))) + } + + test("RUNTIME_FILTER hint on a join without equi-join keys is reported") { + withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.CROSS_JOINS_ENABLED.key -> "true") { + assertHintNotApplied( + "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 > bf2.c2", + "no equi-join keys") + } + } + + test("RUNTIME_FILTER hint inside a subquery applies to the join it is rewritten into") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + // An IN subquery becomes a left semi join, whose left side can be pruned. + assertRewroteWithBloomFilter( + "select * from bf1 where bf1.c1 in (select /*+ RUNTIME_FILTER(bf2) */ c2 from bf2)") + // A NOT EXISTS subquery becomes a left anti join, whose left side cannot. + val notExists = "select * from bf1 where not exists " + + "(select /*+ RUNTIME_FILTER(bf2) */ 1 from bf2 where bf2.c2 = bf1.c1)" + assertDidNotRewriteWithBloomFilter(notExists) + assertHintNotApplied(notExists, "the left side of a left anti join cannot be pruned") + } + } + + test("RUNTIME_FILTER hint without parameters applies to the subtree below it") { + withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") { + val query = "select * from bf1 join (select /*+ RUNTIME_FILTER */ c2 from bf2) t " + + "on bf1.c1 = t.c2" + assertRewroteWithBloomFilter(query) + assert(bloomFilterApplicationSideKeys(query) == Set("c1")) + } + } + test("Merge runtime bloom filters") { withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000", SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000", diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/GlobalTempViewSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/GlobalTempViewSuite.scala index 13035814dc800..98c7b1c3136a0 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/GlobalTempViewSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/GlobalTempViewSuite.scala @@ -187,7 +187,7 @@ class GlobalTempViewSuite extends SharedSparkSession { "SELECT /*+ MAPJOIN(global_temp.v1) */ * FROM global_temp.v1, v2 WHERE v1.id = v2.id" ).foreach { statement => sql(statement).queryExecution.optimizedPlan match { - case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST))), None)) => + case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST), _)), None)) => case _ => fail("broadcast hint not found in a left-side table") } }