cloud-fan commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3995080333
##########
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:
**Non-blocking (P2):** This survival forecast differs from the optimizer
pipeline that will determine the DPP's actual fate in two ways: it invokes
`PushDownPredicates` even when that rule is configured in
`spark.sql.optimizer.excludedRules`, and the final check accepts every
`LeafNode` although `CleanupDynamicPruningFilters` preserves DPP only over its
supported file, Hive, and V2 scan relations. In either case this method can
credit DPP, suppress Bloom and the not-applied warning, and then have cleanup
remove the filter. Please base hint credit on the effective pushdown/cleanup
result, or otherwise make both the configured-rule behavior and supported-scan
predicate exactly match the downstream pipeline.
**Recommended change:** Move the DPP credit and Bloom-fallback decision to a
point after the real DPP pushdown and CleanupDynamicPruningFilters batches, so
InjectRuntimeFilter observes only predicates the configured pipeline retained.
Remove the private PushDownPredicates simulation and LeafNode approximation,
and preserve any required subquery-merge/column-pruning preparation for newly
injected Bloom filters at the new position. Add unsupported-Range-union and
excluded-PushDownPredicates regressions against the final optimized plan.
**Why this works:** The decision will inspect actual surviving
DynamicPruningSubquery nodes rather than predict them, eliminating both
configuration divergence and scan-type drift before it suppresses Bloom or a
warning.
**Scope:**
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer,
sql/core/src/main/scala/org/apache/spark/sql/execution,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** DPP remains preferred when it truly survives, and the
hint still never falls back to filtering in the opposite direction.
**Risks:** Moving injection later can bypass optimizer work that current
Bloom subqueries or filters rely on. Batch reordering can affect subquery
merging and fixed-point assumptions.
**Constraints:** Do not make catalyst depend directly on sql/core scan
implementation classes. Respect spark.sql.optimizer.excludedRules exactly as
the real batch does. A surviving DPP still takes effect as the sole mechanism
for the hinted join; otherwise Bloom and warning gates remain unchanged.
**Success:** A DPP copy on Range or another cleanup-unsupported leaf cannot
credit the hint. Excluding PushDownPredicates cannot make the forecast differ
from the final optimized plan. Every credited DPP survives in the final plan;
otherwise a Bloom filter is present or the hint has a reason-bearing warning.
##########
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:
**Blocking (P1):** `LeafNode` is not a repeatability guarantee. In
particular, `SparkSession.createDataFrame(rdd, ...)` produces a `LogicalRDD`,
and an uncheckpointed RDD can have opaque `mapPartitions` lineage that returns
a different subset on each computation even though all Catalyst attributes
appear stable. This accepts that source, after which the runtime-filter
subquery and join can recompute it and observe different key sets. Please fail
closed on opaque leaves and admit only sources with an explicit
repeatability/materialization guarantee, such as supported scans, local
materialization, or an actually checkpointed `LogicalRDD`; the seeded-sample
check should use the same guarantee.
**Recommended change:** Replace blanket LeafNode acceptance with an explicit
set of source classes whose repeated evaluation is stable, including local
materialization and supported scans, and accept LogicalRDD only when its input
is actually checkpointed. Reuse the same source predicate for seeded Sample's
over-scan check. Add uncheckpointed/checkpointed LogicalRDD controls for both
Bloom and DPP.
**Why this works:** Opaque sources will fail closed unless they expose a
concrete materialization or scan guarantee, preventing independent filter and
join evaluations from reading different key sets.
**Scope:**
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer,
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** Preserve behavior outside the stated repair boundary.
**Risks:** A source-class allowlist can drift as new scan leaf types are
added. Cross-module source classes may require a shared capability predicate
rather than catalyst depending on sql/core implementations.
**Constraints:** Do not equate LeafNode with stable ordering or
repeatability. A lazy checkpoint that has not materialized must remain
ineligible. Seeded Sample must use the same positive source guarantee.
**Success:** An uncheckpointed LogicalRDD with nondeterministic
mapPartitions lineage cannot source Bloom or hinted DPP. An actually
checkpointed LogicalRDD and supported catalog scan retain intended eligibility.
A seeded Sample over an opaque leaf is rejected rather than inheriting blanket
leaf acceptance.
##########
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:
**Nit (P3):** This says the hinted side must produce the same rows on every
evaluation, but the implementation intentionally accepts unstable values
carried only in non-key columns, for example a stable `GROUP BY id` key
alongside `first(name)`. Please describe the actual eligibility contract
instead: row membership and the selected join-key values must be repeatable;
unrelated carried output values need not be identical.
##########
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:
**Blocking (P1):** An unstable outer generator does not necessarily preserve
multiplicity: it can emit a different positive number of rows on each
evaluation while still preserving every input row. This branch records only
unstable `generatorOutput`, so a later `count(*)` (or an aggregate over an
otherwise stable child value) references no tainted attribute and is accepted
as a join key. The filter build and join can then compute different keys and
the runtime filter can drop the matching row. Please propagate a
row-multiplicity instability through this analysis and make downstream
multiplicity-dependent aggregates/windows reject it, while retaining the safe
case where the selected key is independent of that multiplicity.
**Recommended change:** Add an unstable-multiplicity component to Taint. Set
it for an unstable outer Generate (and any other modeled operator whose
repeatable membership can still change duplicate counts), propagate it through
row-preserving operators, and make multiplicity-sensitive Aggregate and Window
outputs unstable or reject the source. Add Bloom and DPP regressions with
count(*) above a nondeterministic outer generator.
**Why this works:** The analysis will no longer infer a stable aggregate
merely because its expression references no generatorOutput attribute; it will
explicitly account for the number of input rows contributing to each
aggregate/window result.
**Scope:**
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** Continue accepting outer generators whose instability is
confined to unused generator outputs and multiplicity-independent selected keys.
**Risks:** An incomplete transfer table could clear multiplicity taint
through another row-expanding or row-collapsing operator. Over-propagation
could conservatively decline safe hints.
**Constraints:** Use the same eligibility result for Bloom and DPP. Preserve
acceptance of an outer generator when the selected key and row membership do
not depend on its varying multiplicity.
**Success:** count(*) and equivalent multiplicity-sensitive keys above an
unstable outer generator are rejected. A stable child key carried through the
same outer generator remains eligible. The hinted and unhinted query return the
same rows for the regression shape.
##########
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:
**Non-blocking (P2):** `CollectMetrics` is row-preserving, but copying this
subtree into the runtime-filter subquery duplicates a user-visible side effect.
A `Dataset.observe` source then produces two collectors with the same
observation name; `QueryExecution.observedMetrics` combines metrics from the
main plan and subqueries with map overwrite semantics, so the injected
evaluation can replace the join-side value. Please reject a source containing
`CollectMetrics` (with the normal reason-bearing warning), or otherwise ensure
the observation is neither duplicated nor replaced.
##########
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:
**Blocking (P1):** This inherited allowlist treats `avg(int)` and
`avg(long)` as order-independent based on their integral input type, but
`Average` accumulates non-decimal inputs in a `Double` sum buffer.
Floating-point addition is not associative, so different input or
partial-buffer merge orders can produce different filter-side and join-side
averages; the filter can then remove a matching row. Please classify `Average`
using its actual accumulation type for this correctness check, conservatively
rejecting integral averages whose buffer is floating point without incidentally
broadening `EliminateSorts` behavior.
--
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]
