cloud-fan commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3966575700
##########
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:
**Non-blocking (P2):** Determinism is not sufficient to prove this DPP
survives. For example, a deterministic `Window` whose partition columns exclude
the join key blocks predicate pushdown; `CleanupDynamicPruningFilters` then
replaces the stranded DPP with `true` after this branch has already credited
the hint. The final plan has neither DPP nor the Bloom fallback, and no warning
is emitted. Please use the same scan-reachability criterion as DPP cleanup
before setting `appliedHint`; otherwise continue to the Bloom-filter path, with
a deterministic barrier regression test that inspects the final optimized plan.
**Recommended change:** Credit an existing DPP predicate only when it can
reach a supported scan; otherwise continue through the existing Bloom-filter
fallback and warning logic.
**Why this works:** Reuse the scan-reachability condition enforced by DPP
pushdown and cleanup instead of application-subtree determinism when computing
prunedByDpp.
**Scope:** InjectRuntimeFilter's DPP decision plus a focused
InjectRuntimeFilterSuite regression using a deterministic pushdown barrier.
**Compatibility:** Plans whose DPP already survives are unchanged; plans
where cleanup would remove DPP now receive a Bloom filter when eligible or the
documented not-applied warning.
**Risks:** Duplicating scan-reachability logic could drift from
CleanupDynamicPruningFilters. The fallback may add a Bloom filter in plans that
previously ended with no runtime filter.
**Constraints:** Keep the survival check aligned with the actual pushdown
and cleanup rules. Preserve the existing Bloom-filter count, key-shape,
enablement, and warning gates.
**Success:** Every credited hinted DPP remains in the final optimized plan;
when it cannot survive, the final plan has the Bloom fallback or the user
receives the documented reason-bearing warning.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -579,6 +579,56 @@ 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)
+ }
+
+ /**
+ * Whether `plan` can serve as a runtime filter source, i.e. produces the
same rows each time it
+ * is evaluated. A runtime filter evaluates its source separately from the
join, so a source that
+ * can yield different rows on re-evaluation could prune rows the join
itself matches.
+ *
+ * `deterministic` covers expressions only. Some operators produce a row set
that depends on
+ * evaluation order even with deterministic expressions: an unordered LIMIT,
OFFSET or TAIL keeps
+ * whichever rows arrive first, and a SAMPLE above anything but a leaf sees
a different row order
+ * per run. An ordered LIMIT is accepted: it is repeatable up to ties at the
cutoff, which Spark
+ * leaves to the user wherever a top-n plan is evaluated more than once.
+ */
+ def isRepeatableRuntimeFilterSource(plan: LogicalPlan): Boolean = {
+ def isOrdered(p: LogicalPlan): Boolean = p match {
+ case Sort(_, true, _, _) => true
+ case _: Project | _: GlobalLimit | _: LocalLimit | _: Offset =>
isOrdered(p.children.head)
+ case _ => false
+ }
+ !plan.isStreaming && plan.deterministic && !plan.exists {
Review Comment:
**Blocking (P1):** Independent runtime-filter source execution already
exists, and DPP broadcast reuse is configuration-dependent; the new risk here
is that this hint admits the complete source plan after only this check.
`plan.deterministic` covers expression determinism, so it still accepts
order-sensitive `first`/`last`/`any_value` aggregates and top-N sources with
ties. Separate filter and join evaluations can then choose different keys,
causing the filter to discard a row the join would match. Please reject
row-order-dependent hinted sources unless their selected key set is proven
stable, and cover both an order-sensitive aggregate and a tied top-N case.
**Recommended change:** Conservatively reject hinted source plans whose
output depends on input order, including order-sensitive aggregates and
LIMIT/OFFSET whose ordering is not proven total.
**Why this works:** Extend isRepeatableRuntimeFilterSource to recognize
these logical and aggregate shapes and return false, allowing the existing
not-applied warning path to handle them.
**Scope:** JoinSelectionHelper's repeatability predicate and focused
runtime-filter tests for both Bloom-filter and DPP consumers.
**Compatibility:** This only narrows acceptance of hint shapes that are
currently unsafe; stable sources retain existing behavior, while rejected
sources keep correct join results and receive the existing warning.
**Risks:** A conservative predicate may decline some sources that happen to
be stable at runtime. Aggregate aliases and equivalent top-N plan shapes must
be classified consistently.
**Constraints:** Do not treat Expression.deterministic or user
responsibility for sort ties as a row-set repeatability proof. Apply the same
source eligibility decision to both Bloom-filter and DPP paths.
**Success:** No accepted source can yield a join-key set in the join
evaluation that was absent from the independent filter evaluation, with
regressions covering first/last/any_value and tied top-N inputs.
--
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]
