ulysses-you commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3966889599
##########
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) &&
Review Comment:
F2: the credit tests only `(application side).deterministic`, but
`CleanupDynamicPruningFilters` keeps a DPP filter only inside a
`NodeWithOnlyDeterministicProjectAndFilter(scan)` chain. A deterministic
operator that blocks `PushDownPredicates` (a `Window` whose spec excludes the
pruning key, an `Aggregate`, ...) also causes the drop to true, so the hint is
credited to a filter that then dies: no bloom, and no warning, violating "the
hint is never silently dropped".
Observed on this head (DPP on, `REUSE_BROADCAST_ONLY=false`):
```sql
SELECT /*+ RUNTIME_FILTER(t) */ w.k FROM
(SELECT k, v, row_number() OVER (PARTITION BY v ORDER BY v) rn FROM ppart2)
w
JOIN pd t ON w.k = t.k WHERE w.rn = 1 AND t.v = 3
```
No bloom, no DPP in the final optimized plan, `warnings=` empty. Control:
`PARTITION BY k` (key inside the spec) survives to the executed plan. Without
the hint the skip is silent today too (pre-existing-made-visible); the breach
is of this PR's own guarantee. Suggested fix: credit only when the DPP can
reach a filterable scan (mirror the cleanup rule's shape test); same class
applies to a hint inside a correlated subquery, which both rules early-return
without reporting.
##########
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 {
+ case l @ (_: GlobalLimit | _: LocalLimit | _: Offset | _: Tail) =>
!isOrdered(l.children.head)
+ case Sample(_, _, _, _, child, _) => !child.isInstanceOf[LeafNode]
Review Comment:
F1, blocking: a `Sample` over a leaf is accepted regardless of `seed`.
`Sample(seed = None)` is documented to generate a random seed at execution
time, and `SampleExec.resolvedSeed` (`basicPhysicalOperators.scala:525`)
resolves it per physical operator instance, so the join's copy and the
runtime-filter subquery's copy of the hinted side draw different samples. The
filter then prunes rows that the join itself matched: wrong results.
Repro on this head (`pf` 200 keys, `pd` 8 rows,
`AUTO_BROADCASTJOIN_THRESHOLD=-1`, scan threshold 1):
```sql
SELECT /*+ RUNTIME_FILTER(t) */ j.k, t.k, t.v FROM pf j
JOIN (SELECT * FROM pd TABLESAMPLE (50 PERCENT)) t ON j.k = t.k
```
Bloom injected, zero warnings; over 30 runs hinted avg 2.067 vs plain avg
4.000, i.e. the expected size of an intersection of two independent 50 percent
draws (8*0.25 = 2) rather than one draw (8*0.5 = 4). `REPEATABLE (42)` control:
hinted stable across 12 runs, row-equal to plain. `Sample` has no
`deterministic` override, so `plan.deterministic` above doesn't catch it
either. The unordered-LIMIT test guards the same axis; a seedless SAMPLE has no
test.
Fix shape: `case s @ Sample(_, _, _, _, child, _) =>
!child.isInstanceOf[LeafNode] || s.seed.isEmpty`, plus seedless-TABLESAMPLE and
REPEATABLE cases in "RUNTIME_FILTER hint is ignored for a creation side that is
not repeatable". The same helper gates the hinted DPP path, so this covers both.
##########
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 ||
Review Comment:
F3 (plan quality): for a hinted join the opposite direction returns `None`
unconditionally, and `PartitionPruning` mirrors it (`!pruneLeftHinted` /
`!pruneRightHinted`). So a hint replaces the heuristics' direction rather than
just pinning the source side: a join where the heuristics would otherwise build
a filter toward the hinted side loses it even when the hinted direction fails.
Intended, but the doc only promises "the hinted side is never itself pruned";
it should also say a failed hint does not fall back to the opposite direction,
plus one plan-shape test pinning that.
##########
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),
Review Comment:
Nit: all three `joinHintNotSupported(HintInfo(runtimeFilterSource = true),
...)` sites (here, :549, :562) log `Hint (runtime_filter_source) is not
supported...` with no relation name, so with multiple hinted joins the user
cannot tell which was rejected. The actual side `HintInfo` (and a side label in
the reason) is at hand here; passing it through instead of the synthetic value
would keep these messages consistent with the strategy-hint ones.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -260,8 +263,10 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
case Filter(condition, _) => condition.collect {
case subquery: org.apache.spark.sql.catalyst.expressions.ScalarSubquery
=> subquery.plan.collect {
+ // A hinted creation side can carry its own `Aggregate` (e.g. a
`SELECT DISTINCT` one),
+ // so count the Bloom filter aggregates rather than assuming every
aggregate is one.
case Aggregate(_, aggregateExpressions, _, _) =>
- aggregateExpressions.map {
+ aggregateExpressions.collect {
Review Comment:
Nit: the map-to-collect switch is needed for hinted sources that carry their
own aggregate, but it also relaxes every existing query: a non-Bloom aggregate
inside a filter subquery no longer fails the assertion, and the
`numBloomFilterAggs == numMightContains` cross-check only compensates when the
BloomFilterMightContain count changes, not for a stray subquery containing
neither. A hinted-only branch (or an expected-aggregate-count parameter) would
keep the old teeth for non-hinted plans.
--
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]
