sunchao commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r3986522728
##########
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:
[P1] Track unstable row counts through outer generators
An outer generator preserves each input row's presence, but an unstable
array length still changes its multiplicity. Marking only `generatorOutput` as
unstable loses that information: a following `count(*)` references none of
those attributes and is accepted as a stable join key. For example, use a
nondeterministic Scala UDF returning an array of length 1–20 in `SELECT
count(*) AS k FROM range(1) LATERAL VIEW OUTER explode(random_array(id)) x AS
v`, then join this hinted source to `range(1, 21)` on `id = k`. On 4959171c248,
with AQE/broadcast and exchange/subquery reuse disabled, the plain query
returned one row in all six runs; the hinted query injected a Bloom filter and
returned zero rows in five of six runs. The two source evaluations counted
different array lengths, so matching rows were discarded. Please propagate
unstable multiplicity into aggregate/window results, or reject these source
shapes.
##########
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:
[P1] Analyze subqueries inside the join key itself
This entry point checks `key.deterministic` and attribute taint, but does
not call the recursive `isStable` check used for expressions inside the source
plan. A scalar subquery embedded directly in the join key therefore bypasses
the new repeatability analysis:
```sql
SELECT /*+ RUNTIME_FILTER(t) */ f.id
FROM range(11) f JOIN range(1) t
ON f.id = t.id + coalesce(
(SELECT max(id) FROM sampled_data TABLESAMPLE (50 PERCENT)), 0L)
```
With `sampled_data` containing 0–9, this must return one row for any sample.
On 4959171c248, with AQE/broadcast and exchange/subquery reuse disabled, the
unhinted control returned one row in all eight runs; the hinted query injected
a Bloom filter and returned zero rows in six of eight runs. The filter and join
evaluate different samples. Please apply `isStable(key, t.unstable)` here too,
so the same unsafe scalar subquery is rejected whether projected by the source
or written directly in the join condition.
##########
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:
[P2] Match the scan types accepted by DPP cleanup
`CleanupDynamicPruningFilters` preserves DPP only over file, Hive, or V2
scan relations, whereas this check accepts every `LeafNode`. This is reachable
through a union whose first branch is a partitioned scan behind a pushdown
barrier:
```sql
SELECT /*+ RUNTIME_FILTER(s) */ f.p
FROM (
SELECT p FROM (SELECT p FROM partitioned_fact LIMIT 2)
UNION ALL SELECT id AS p FROM range(3)
) f JOIN range(2) s ON f.p = s.id
```
On 4959171c248, with `partitioned_fact` partitioned by `p`, DPP and Bloom
enabled and broadcasting disabled, the final optimized plan has neither filter
and emits no not-applied warning. DPP cannot pass the limit, but its copy on
`Range` satisfies this check and suppresses the Bloom fallback; cleanup
subsequently removes both copies. Removing the union's `Range` branch correctly
produces a Bloom filter. Please share cleanup's supported-scan predicate so a
filter on an unsupported leaf cannot count as applying the hint.
##########
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:
[P2] Guard Python UDFs before accepting a hinted DPP source
A deterministic Python UDF can pass this repeatability check, but the DPP
build plan is captured before `ExtractPythonUDFs` and later planned without
that extraction. For a Parquet fact table partitioned by `p` and a Python
identity UDF returning `long`, this is enough:
```sql
SELECT /*+ RUNTIME_FILTER(s) */ f.id, s.k
FROM partitioned_fact f
JOIN (SELECT python_identity(id) AS k FROM range(2)) s ON f.p = s.k
```
On 4959171c248 with broadcasting disabled and the default exchange/subquery
reuse settings, the unhinted query succeeds; the hinted query fails with
`[INTERNAL_ERROR] Cannot generate code for expression: python_identity(...)`.
This reproduces with AQE both off and on. The main join contains
`BatchEvalPython`, while the DPP subquery still contains the unevaluable
`PythonUDF`. Bloom's explicit Python-UDF guard does not help because the
surviving DPP has already been credited. Please reject an unextracted Python
UDF on this DPP path, or ensure the copied build plan receives the required
extraction before physical planning.
--
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]
