cloud-fan commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3995082746
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,55 +445,136 @@ 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)) {
+ hintErrorHandler.joinHintNotSupported(HintInfo(runtimeFilterSource =
true),
+ "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) match {
+ case Some(filterCreationSide) =>
+ injectFilter(applicationSideKey, currentApplicationSide,
filterCreationSide)
+ .fold(reason => blocked(reason), Some(_))
+ case None =>
+ blocked("the hinted side may produce different rows when
evaluated again")
+ }
+ }
+ }
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)) {
+ // A DPP filter on the key already prunes the application side, by
whole partitions
+ // rather than by rows, so no Bloom filter is added. That also
honors the hint, if any,
+ // provided the DPP predicate survives:
`CleanupDynamicPruningFilters` drops it when
+ // `PushDownPredicates` cannot carry it to the scan, which a
non-deterministic operator
+ // on the pruned side prevents. A Bloom filter needs no pushdown, so
one is still added
+ // for the hint in that case.
+ val prunedByDpp = hasDynamicPruningSubquery(left, right, l, r) &&
+ (!hinted || (if (injectLeftHinted) left else right).deterministic)
Review Comment:
The window barrier is fixed, but the survival predicate still does not match
cleanup: it accepts every LeafNode, while CleanupDynamicPruningFilters
preserves only file, Hive, and V2 scan relations. A Range branch can therefore
credit DPP that cleanup later removes, suppressing both Bloom and the warning.
Could you share the exact supported-scan predicate here?
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3966575700","thread_id":"inline:3966575700","verdict_sha256":"7420fc2de59b7ada011c053eea2a54ada04cf904d9431a35824b4dcb466fbd51"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
