pan3793 commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r4218005049
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -306,10 +315,22 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
* meet the following requirements:
* (1) it can not be a stream
* (2) it needs to contain a selective predicate or a cheaply-recomputable
materialized input
+ *
+ * (2) is evidence that pruning pays off, which a [[RuntimeFilterHint]] on
the filtering side
+ * (`hinted`) supplies directly. A hinted side only has to be a repeatable
source of the
+ * filtering key, see `JoinSelectionHelper.isRepeatableRuntimeFilterSource`,
since DPP
+ * re-evaluates it.
*/
- private def hasPartitionPruningFilter(plan: LogicalPlan): Boolean = {
- !plan.isStreaming &&
- (hasSelectivePredicate(plan) ||
isCheaplyRecomputableMaterializedPlan(plan))
+ private def hasPartitionPruningFilter(
+ plan: LogicalPlan,
+ hinted: Boolean,
+ filteringKey: Expression): Boolean = {
+ if (hinted) {
+ isRepeatableRuntimeFilterSource(plan, filteringKey)
Review Comment:
Fixed in 00d7051f73a: a hinted filtering side containing a Python UDF is not
used for DPP, since the standalone subquery is planned without the Python UDF
extraction. The same failure exists for unhinted standalone DPP with
`reuseBroadcastOnly=false`; the hint made it reachable by default. Test with
`TestPythonUDF` in `DynamicPartitionPruningSuite`.
##########
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)
Review Comment:
Done as recommended in 00d7051f73a: `InjectRuntimeFilter` moved after the
DPP pushdown and cleanup batches, followed by its own pushdown batch for the
injected Bloom filters. The simulation and the `LeafNode` approximation are
gone, so `excludedRules` and the cleanup's supported scans are respected by
construction. Regressions on the final optimized plan: window barrier,
non-deterministic filter, and union with `Range`. As a side effect, an unhinted
DPP filter that cleanup drops no longer suppresses the Bloom filter on that key.
##########
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) {
Review Comment:
Fixed in 00d7051f73a, see the reply to the same point on `joins.scala:807`.
The same eligibility result gates Bloom and DPP.
##########
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))
Review Comment:
Fixed in 00d7051f73a. `LeafNode` gains `isOutputRepeatable`, false by
default, so unknown and third-party leaves fail closed. It is true for batch
`LocalRelation`, `Range`, `OneRowRelation`, `EmptyRelation`,
`HiveTableRelation`, `DataSourceV2ScanRelation`, `LogicalRelation` and
`CommandResult`, for `InMemoryRelation` through its existing
materialized-and-repeatable definition, and for `LogicalRDD` only when actually
checkpointed; streaming leaves are never repeatable. The seeded-sample check
uses the same predicate. Tests with checkpointed and uncheckpointed RDD sources.
##########
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)))
+ } else {
+ Left(NotRepeatable)
+ }
+ }
+
+ case e: Expand => analyze(e.child).map { t =>
+ val unstable = e.output.zipWithIndex.collect {
+ case (attr, i) if e.projections.exists(p => !isStable(p(i),
t.unstable)) => attr
+ }
+ Taint(AttributeSet(unstable))
+ }
+
+ case s: Sort => analyze(s.child).map { t =>
+ val stableOrder = s.order.forall(o => isStable(o.child, t.unstable))
+ Taint(t.unstable, totallyOrdered = s.global && stableOrder &&
+ sortedOnUniqueKey(s.child, s.order.map(_.child)))
+ }
+
+ // A limit keeps whichever rows arrive first unless the order is total.
+ case l @ (_: GlobalLimit | _: LocalLimit | _: Offset | _: Tail) =>
+ analyze(l.children.head).flatMap { t =>
+ if (t.totallyOrdered) Right(t) else Left(NotRepeatable)
+ }
+
+ // A sample draws a fresh seed per evaluation unless one is given, and
depends on the input
+ // row order even then: only a seeded sample over a scan, through
projections and filters
+ // that keep the row order, is repeatable.
+ case s: Sample =>
+ val overScan = NodeWithOnlyDeterministicProjectAndFilter.unapply(s.child)
+ .exists(_.isInstanceOf[LeafNode])
+ if (s.seed.isDefined && overScan) analyze(s.child) else
Left(NotRepeatable)
+
+ // The rows and their values are unchanged; a shuffle loses the order.
+ case _: Distinct | _: SubqueryAlias | _: Repartition | _:
RepartitionByExpression |
+ _: RebalancePartitions =>
+ analyze(plan.children.head).map(t => Taint(t.unstable))
+
+ // Observed metrics do not touch the rows.
+ case c: CollectMetrics => analyze(c.child)
Review Comment:
Fixed in 00d7051f73a: a source containing `CollectMetrics` is rejected with
its own reason. Test with a `Dataset.observe` source.
##########
docs/sql-ref-syntax-qry-select-hints.md:
##########
@@ -172,6 +172,55 @@ SELECT /*+ SHUFFLE_REPLICATE_NL(t1) */ * FROM t1 INNER
JOIN t2 ON t1.key = t2.ke
SELECT /*+ BROADCAST(t1), MERGE(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key =
t2.key;
```
+### Runtime Filter Hints
+
+A runtime filter prunes one side of a join using the join key values found on
the other side, so
+rows that cannot match are discarded early. Spark decides on its own whether
such a filter is worth
+building, based on estimates of how much data it would save. Runtime filter
hints let users make
+that decision instead, for the cases where the estimates are unavailable or
wrong.
+
+#### Runtime Filter Hints Types
+
+* **RUNTIME_FILTER**
+
+ Suggests that Spark build a runtime filter from the hinted relation and
use it to prune the
+ other side of the join. Use it when the hinted side is known to match only
a small fraction
+ of the other side, but Spark does not choose a runtime filter on its own,
typically because
+ table statistics are missing or misleading. The hinted side may be any
relation or subquery,
+ and is never itself pruned. The hint does not choose how the pruning is
done; Spark picks the
+ mechanism. `RUNTIME_FILTER` can be combined with a join strategy hint.
+
+The hint overrides Spark's cost estimates, but not the requirements that make
a runtime filter
+correct, so Spark is not guaranteed to follow it. A side that join semantics
forbid pruning is
+never pruned, e.g. the left side of a `LEFT OUTER` join, whose rows must all
appear in the output.
+The hinted side must produce the same rows each time it is evaluated, since
building the filter
Review Comment:
Fixed in 00d7051f73a: the docs now say the hinted side must produce the same
rows and join keys whenever it is evaluated. The section was also cut back to
the contract; the eligibility details live in the scaladoc.
##########
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)))
+ } else {
+ Left(NotRepeatable)
+ }
+ }
+
+ case e: Expand => analyze(e.child).map { t =>
+ val unstable = e.output.zipWithIndex.collect {
+ case (attr, i) if e.projections.exists(p => !isStable(p(i),
t.unstable)) => attr
+ }
+ Taint(AttributeSet(unstable))
+ }
+
+ case s: Sort => analyze(s.child).map { t =>
+ val stableOrder = s.order.forall(o => isStable(o.child, t.unstable))
+ Taint(t.unstable, totallyOrdered = s.global && stableOrder &&
+ sortedOnUniqueKey(s.child, s.order.map(_.child)))
+ }
+
+ // A limit keeps whichever rows arrive first unless the order is total.
+ case l @ (_: GlobalLimit | _: LocalLimit | _: Offset | _: Tail) =>
+ analyze(l.children.head).flatMap { t =>
+ if (t.totallyOrdered) Right(t) else Left(NotRepeatable)
+ }
+
+ // A sample draws a fresh seed per evaluation unless one is given, and
depends on the input
+ // row order even then: only a seeded sample over a scan, through
projections and filters
+ // that keep the row order, is repeatable.
+ case s: Sample =>
+ val overScan = NodeWithOnlyDeterministicProjectAndFilter.unapply(s.child)
+ .exists(_.isInstanceOf[LeafNode])
+ if (s.seed.isDefined && overScan) analyze(s.child) else
Left(NotRepeatable)
+
+ // The rows and their values are unchanged; a shuffle loses the order.
+ case _: Distinct | _: SubqueryAlias | _: Repartition | _:
RepartitionByExpression |
+ _: RebalancePartitions =>
+ analyze(plan.children.head).map(t => Taint(t.unstable))
+
+ // Observed metrics do not touch the rows.
+ case c: CollectMetrics => analyze(c.child)
+
+ // Anything else, e.g. a typed operator or a script transformation, is not
analyzed.
+ case _ =>
+ Left(s"the hinted side contains ${plan.nodeName}, which cannot be
checked for repeatability")
+ }
+
+ /**
+ * Whether an aggregate expression's value is independent of the input
order. Spark's own
+ * allowlist covers the SQL functions; a Bloom filter aggregate, which this
rule injects for an
+ * inner join, merges commutatively.
+ */
+ private def isOrderIrrelevantAggregate(e: NamedExpression): Boolean = e
match {
+ case Alias(AggregateExpression(_: BloomFilterAggregate, _, _, _, _), _) =>
true
+ case _ => EliminateSorts.isOrderIrrelevantAggs(Seq(e))
Review Comment:
Fixed in 00d7051f73a. The analysis has its own classification instead of
reusing `EliminateSorts`, which is untouched: min, max, count, bit aggregates,
sum over a non-floating input, and average only with a decimal or interval
accumulator. `avg(int)` and the central moments are now order-dependent. Test
with an `avg` key.
--
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]
