pan3793 commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3985776781
##########
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:
Fixed in 4959171c248. The hinted side is now checked for key-set stability
instead of expression determinism. `RuntimeFilterSourceAnalysis` walks the side
bottom-up and tracks output attributes whose values are unstable:
non-deterministic expressions, order-dependent aggregate and window results,
subqueries whose plans fail the same check, and expressions over any of those.
The side is rejected when its row set is unstable (a filter, join condition or
grouping consumes an unstable attribute, an inner generate has an unstable
generator, a sample is unseeded or not over a scan, a limit is not over a
proven total order, i.e. sort keys covering an aggregate's grouping keys) or
when an operator is not modeled, and it qualifies only if the join key
references no unstable attribute. Both the Bloom filter and the DPP path use
it. Tests cover `first`/`any_value` keys, tied `ORDER BY ... LIMIT` over a
table, window-derived keys, `HAVING first(...)`, generators, subquery keys, and
the DPP co
nsumer.
##########
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:
Fixed in 4959171c248. The hint is credited to a DPP filter only after
simulating its fate: `PushDownPredicates` runs to a fixed point on the pruned
side, and the filter (matched by `exprId`) must end inside a
`NodeWithOnlyDeterministicProjectAndFilter` chain over a leaf, which is what
`CleanupDynamicPruningFilters` keeps. Reusing the rule avoids duplicating its
reachability logic. Otherwise the Bloom filter path runs. Regressions on the
final optimized plan: a window whose spec excludes the key and a
non-deterministic filter each end with a Bloom filter and no DPP; a
non-deterministic branch the DPP filter is pushed past ends with the DPP filter
and no Bloom filter. The silent DPP drop on the unhinted path is pre-existing
and untouched here; I will file it separately.
##########
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:
Fixed in 4959171c248: a `Sample` qualifies only with a seed and over a scan,
through deterministic projections and filters that keep the row order. Added
the seedless `TABLESAMPLE` and `REPEATABLE (42)` cases, and a seedless-sample
negative for the DPP consumer in `DynamicPartitionPruningSuite`.
##########
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:
Fixed in 4959171c248, see the reply on the same line for the survival check.
The correlated-subquery early return now reports a hinted join instead of
returning silently. The hint also takes effect through exactly one mechanism:
when a DPP filter honors it on any key, no Bloom filter is added on any key of
that join.
--
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]
