pan3793 commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r4218003803
##########
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:
Addressed in 00d7051f73a by observing the surviving DPP filters instead of
predicting them, see the reply on `InjectRuntimeFilter.scala:453`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -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)))
Review Comment:
Fixed in 00d7051f73a. The analysis now tracks unstable row multiplicity: an
outer generate with an unstable generator sets it, row-preserving operators and
joins propagate it, every aggregate and window result over such rows becomes
unstable, and grouping or `DISTINCT` clears it. Your `count(*)` shape is
rejected; a stable key carried through the same generate still qualifies. Tests
cover both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -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)
Review Comment:
Fixed in 00d7051f73a and 2ea84990cd0. Both join keys go through the same
stability check as expressions inside the source, so a scalar subquery over an
unseeded sample is rejected whether it sits in the source, the creation-side
key, the application-side key, or the pruning key of a hinted DPP filter. Tests
cover each placement, including a partitioned table with subquery reuse
disabled.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -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])
Review Comment:
Fixed in 00d7051f73a by removing the prediction. `InjectRuntimeFilter` now
runs after the DPP pushdown and `CleanupDynamicPruningFilters` batches, so it
only sees DPP filters that actually survived, and credits the hint to one whose
pruning key and build key match the join keys. Your union-with-`Range` shape
now gets a Bloom filter and no DPP; test added.
--
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]
