peter-toth commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r4495128548
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
}
+ currentApplicationSide
+ } else if (applicationHinted) {
Review Comment:
**Finding 2.** This is
[SPARK-58486](https://issues.apache.org/jira/browse/SPARK-58486). The Bloom
filter hashes a collated string by its bytes, while the join compares it under
the collation. So for a collated key that isn't a plain column, `'abc'` and
`'ABC'` match in the join but not in the filter. Without the hint, the ticket's
repro needs `applicationSideScanSizeThreshold` lowered. With the hint, it fails
at the default settings. With the ticket's tables, this returns 0 instead of
5,000:
```sql
SELECT /*+ RUNTIME_FILTER(d) */ count(*) FROM fact f JOIN dim d ON
substr(f.k, 1, 3) = d.key
WHERE d.cat = 'x'
```
Without the hint, the same query returns 5,000.
Until the ticket is fixed, the hinted Bloom path could refuse such a key, as
`allowMaterializedCache` already does for materialized caches:
```scala
} else if (applicationHinted &&
!(UnsafeRowUtils.isBinaryStable(applicationSideKey.dataType) &&
UnsafeRowUtils.isBinaryStable(creationSideKey.dataType))) {
hintBlocked("the join key has a collation the Bloom filter hash
does not follow")
currentApplicationSide
} else if (applicationHinted) {
```
With it, the query returns 5,000. `f.k = d.key` still gets its filter, and
`InjectRuntimeFilterSuite` passes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala:
##########
@@ -53,6 +53,8 @@ case class LogicalRelation(
output = output.map(QueryPlan.normalizeExpressions(_, output)),
catalogTable = None)
+ override def isOutputRepeatable: Boolean = !isStreaming
Review Comment:
**Finding 4.** This trusts every non-streaming `BaseRelation`, but some of
them return different rows when read again:
- A V1 JDBC relation whose `dbtable` or `query` is itself a query, e.g. with
a `LIMIT`, `random()` or a sequence.
- A Kafka batch read whose `endingOffsets` is `latest`, which each scan
resolves again.
The filter subquery and the join each read the relation. I measured it with
an H2 sequence:
```scala
spark.read.jdbc(url, "(SELECT NEXT VALUE FOR s AS k) q", new Properties())
.createOrReplaceTempView("seq_side")
```
`SELECT /*+ RUNTIME_FILTER(t) */ count(*) FROM fact f JOIN seq_side t ON f.k
= t.k` returns 0. The same query without the hint returns 20,000. The filter is
built from one value of the sequence, and the join reads the next.
This is the V1 counterpart of the pushed DSv2 `LIMIT` from
[r4250017527](https://github.com/apache/spark/pull/58635#discussion_r4250017527),
which `repeatableScan` now rejects. The cache already has a check for the same
contract: `InMemoryRelation.hasRepeatableLogicalPlan` trusts only a
`HadoopFsRelation` with a known file format. A shape in line with it, which I
haven't run the suites with:
```scala
override def isOutputRepeatable: Boolean = !isStreaming &&
relation.isInstanceOf[HadoopFsRelation]
```
Other relations could then opt in, e.g. a JDBC relation over a plain table.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
}
+ currentApplicationSide
+ } else if (applicationHinted) {
+ // Unlike the heuristics below, a hinted side adds its own filter
even on a key that
+ // already has one.
+ runtimeFilterHintRejection(applicationSideKey, creationSide,
creationSideKey) match {
+ case Some(reason) =>
+ hintBlocked(reason)
+ currentApplicationSide
+ case None =>
+ injectFilter(applicationSideKey, currentApplicationSide,
FilterCreationSide(
Review Comment:
**Finding 1.** When the application side is a stream with a stateful
operator, `PushDownRuntimeFilters` moves this Bloom filter below that operator.
The filter is built again from the static side in every micro-batch. So rows it
drops in one batch never reach the operator's state, and a key the static side
gains later finds its earlier rows missing. The docs recommend this shape: "the
static side of a stream-static join can be hinted to prune the stream".
I ran this `StreamTest`, with `testcat` an `InMemoryTableCatalog` and
`autoBroadcastJoinThreshold = -1`:
```scala
sql("CREATE TABLE testcat.ns.dim (k INT, v STRING)")
sql("INSERT INTO testcat.ns.dim VALUES (1, 'a')")
val input = MemoryStream[Int]
val agg = input.toDF().groupBy($"value".as("k")).count()
val joined = agg.join(spark.table("testcat.ns.dim").hint("runtime_filter"),
"k")
testStream(joined, OutputMode.Update())(
AddData(input, 1, 2),
CheckNewAnswer((1, 1L, "a")),
Execute { _ => sql("INSERT INTO testcat.ns.dim VALUES (2, 'b')") },
AddData(input, 2),
CheckNewAnswer((2, 2L, "b")))
```
Results:
- With the hint, the head returns `[2,1,b]`. The base, which ignores the
hint, returns `[2,2,b]`.
- Without the hint, both return `[2,2,b]`.
- Without the hint, with `applicationSideScanSizeThreshold = 0` and a
selective static side (`where($"k" < 100)`), both return `[2,1,b]`. So the bug
exists already, and the hint makes it reachable with the default settings.
I tried skipping a streaming application side with a stateful operator at
the top of `tryInject`, for the hint and the heuristics alike:
```scala
if (applicationSide.isStreaming &&
applicationSide.exists(UnsupportedOperationChecker.isStatefulOperation)) {
if (applicationHinted) {
hintBlocked(s"the $sideName side is a stream with a stateful
operator")
}
currentApplicationSide
} else if (!canPrune) {
```
All three cases then return `[2,2,b]`. `InjectRuntimeFilterSuite` and
`streaming/available-now-runtime-filter-hint.sql` still pass. The guard covers
the heuristics too. If you'd rather keep this PR to the hint, and agree that
the unhinted path is a bug too, could you file a ticket for it? A stateful case
next to that golden file would pin the fix.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -578,3 +640,286 @@ trait JoinSelectionHelper extends Logging {
conf.getConfString("spark.sql.join.forceApplyShuffledHashJoin", "false")
== "true"
}
}
+
+/**
+ * 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 rows the
source produces and the
+ * key values they carry 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, and a leaf
+ * such as an RDD that is not checkpointed may return different rows on each
computation.
+ *
+ * The plan is walked bottom-up, tracking:
+ * 1. output attributes whose values are unstable: they come from a
non-deterministic expression,
+ * an order-dependent aggregate or window function, a subquery whose plan
fails this same
+ * analysis, or an expression over such an attribute;
+ * 2. whether the row multiplicity is unstable, i.e. every row is still
present but may be
+ * repeated a different number of times, which an outer generate with an
unstable generator,
+ * or an aggregate over an unstable value, causes and which makes every
aggregate over the
+ * rows unstable.
+ *
+ * The plan is rejected outright when its row set is unstable (a filter or
join condition consumes
+ * an unstable attribute, an inner generate uses an unstable generator, any
limit, offset or
+ * sample), when a leaf cannot promise the same rows again, and, each with its
own reason, when it
+ * calls a function the filter subquery cannot evaluate, when an operator
would duplicate an
+ * observation or 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 {
+
+ /**
+ * @param unstable output attributes whose values depend on evaluation order
or on chance.
+ * @param unstableMultiplicity whether rows may be repeated a different
number of times per
+ * evaluation, while each row stays present.
+ */
+ private case class Taint(unstable: AttributeSet, unstableMultiplicity:
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 (containsExtractedFunction(plan) ||
+ // A DPP filter evaluates the key in its own subquery, which no UDF
extraction rewrites.
+ key.containsAnyPattern(PYTHON_UDF, EXTERNAL_UDF)) {
+ Some("the hinted side contains a Python UDF, Python UDTF or external
UDF, which a runtime " +
+ "filter cannot evaluate")
+ } else {
+ analyze(plan) match {
+ case Left(reason) => Some(reason)
+ case Right(t) if !isStable(key, t.unstable) => Some(NotRepeatable)
+ case _ => None
+ }
+ }
+ }
+
+ def isRepeatableExpression(e: Expression): Boolean = isStable(e,
AttributeSet.empty)
+
+ def containsExtractedFunction(plan: LogicalPlan): Boolean = {
+ plan.containsAnyPattern(PYTHON_UDF, EXTERNAL_UDF) || plan.exists {
+ case g: Generate => g.generator.isInstanceOf[PythonUDTF]
+ case _ => false
+ }
+ }
+
+ /**
+ * 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 => !isRepeatableSubquery(s)
+ case _ => false
+ }
+ }
+
+ /**
+ * A subquery is repeatable when its row set is, and so is whatever of its
output the subquery
+ * exposes: the build keys of a DPP filter, nothing but existence for
EXISTS, the output columns
+ * otherwise. A scalar subquery fails on more than one row, so its
multiplicity counts too.
+ */
+ private def isRepeatableSubquery(s: SubqueryExpression): Boolean = {
+ !s.plan.isStreaming && analyze(s.plan).exists { t =>
+ s match {
+ case d: DynamicPruningSubquery =>
+ d.broadcastKeyIndices.forall(i => isStable(d.buildKeys(i),
t.unstable))
+ case _: Exists => true
+ case _: ScalarSubquery =>
+ !t.unstableMultiplicity &&
s.plan.outputSet.intersect(t.unstable).isEmpty
+ case _ => s.plan.outputSet.intersect(t.unstable).isEmpty
+ }
+ }
+ }
+
+ /** 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 {
+ // Whether a materialized leaf returns the same rows is known only once it
is materialized.
+ case m: MaterializedLeafNode if m.materializedMetadata.isEmpty =>
+ Left(s"the hinted side reads ${m.nodeName}, which is not materialized
yet")
+
+ // The source may have absorbed an operator from the hinted side, so name
that rather than the
+ // scan the user did not write.
+ case s: DataSourceV2ScanRelation if !s.isOutputRepeatable =>
+ Left(s"the hinted side reads ${s.name} with an aggregate, limit, offset,
top-N or sample " +
+ "pushed into the source, which may return different rows when
evaluated again")
+
+ case l: LeafNode =>
+ if (l.isOutputRepeatable) {
+ Right(Taint(AttributeSet.empty))
+ } else {
+ Left(s"the hinted side reads ${l.nodeName}, which may return different
rows when " +
+ "evaluated again")
+ }
+
+ case p: Project => analyze(p.child).map { t =>
+ t.copy(unstable =
+ AttributeSet(p.projectList.filterNot(isStable(_,
t.unstable)).map(_.toAttribute)))
+ }
+
+ 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, l.unstableMultiplicity ||
r.unstableMultiplicity))
+ } else {
+ Left(NotRepeatable)
+ }
+ }
+ }
+
+ case a: Aggregate => analyze(a.child).map { t =>
+ // Grouping on an unstable value changes which rows form a group, and
unstable multiplicity
+ // changes how many rows each group has, so every aggregate result then
depends on them; a
+ // grouping expression's own value is as stable as its input. With
stable grouping the
+ // groups occur once each, whatever the input multiplicity; with
unstable grouping, the
+ // number of groups that share the stable grouping values varies.
+ val stableGroups = a.groupingExpressions.forall(isStable(_, t.unstable))
+ // A group of values that are equal but not identical, e.g. under a
case-insensitive
+ // collation, outputs whichever of them arrives first.
+ val representatives = AttributeSet(a.groupingExpressions
+ .filterNot(e =>
UnsafeRowUtils.isBinaryStable(e.dataType)).flatMap(_.references))
+ val unstable = a.aggregateExpressions.filter { e =>
+ !isStable(e, t.unstable) ||
+ referencesOutsideAggregates(e).intersect(representatives).nonEmpty ||
+ (e.exists(_.isInstanceOf[AggregateExpression]) &&
+ (!stableGroups || t.unstableMultiplicity || !isOrderIrrelevant(e)))
+ }
+ Taint(AttributeSet(unstable.map(_.toAttribute)), unstableMultiplicity =
!stableGroups)
+ }
+
+ case w: Window => analyze(w.child).map { t =>
+ val stablePartitions = w.partitionSpec.forall(isStable(_, t.unstable))
+ val unstable = w.windowExpressions.filter { e =>
+ !stablePartitions || t.unstableMultiplicity || !isStable(e,
t.unstable) ||
+ !isOrderIrrelevantWindow(e)
+ }
+ t.copy(unstable = 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),
+ taints.exists(_.exists(_.unstableMultiplicity))))
+ }
+
+ // An inner generate drops the rows for which the generator yields
nothing, so an unstable
+ // generator changes the row set; an outer generate keeps every row but
repeats it as many
+ // times as the generator yields values, so an unstable one changes the
multiplicity.
+ 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),
unstableMultiplicity = true))
Review Comment:
**Finding 6.** Accepting this is right for the key set. But `ColumnPruning`
can't drop an outer `Generate`, so the filter's copy of the side keeps the
non-deterministic generator. `SubqueryExpression.deterministic` includes the
plan, so the filter itself becomes non-deterministic, and pushdown won't move
it past some operators. I measured it with a hinted `t` = `SELECT k, v FROM dim
LATERAL VIEW OUTER explode(nd_array(k)) x AS v`:
- Against `(SELECT k, count(*) c FROM fact GROUP BY k) f`, the Bloom filter
stays above the `Aggregate`. With a stable `t`, it reaches the `fact` scan.
- Against a table partitioned by the join key, the hinted DPP filter isn't
pushed, so cleanup drops it, and a Bloom filter replaces it. With a stable `t`,
the DPP filter stays.
In both cases the hint counts as applied, and nothing is logged. Neither the
Bloom filter aggregate nor DPP's distinct-key aggregate depends on how many
times a row repeats. So the filter subquery and the DPP build plan could drop
an outer `Generate` whose output they don't use, which keeps them deterministic.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
}
+ currentApplicationSide
+ } else if (applicationHinted) {
+ // Unlike the heuristics below, a hinted side adds its own filter
even on a key that
+ // already has one.
+ runtimeFilterHintRejection(applicationSideKey, creationSide,
creationSideKey) match {
Review Comment:
**Finding 5.** This calls the shared check, which rejects a hinted side with
a Python UDF, a Python UDTF or an external UDF anywhere in it.
[r4250017548](https://github.com/apache/spark/pull/58635#discussion_r4250017548)
made it one check for both paths. DPP needs it, since its standalone subquery
runs the whole side. The Bloom filter subquery is column-pruned first, though,
and `injectBloomFilter` already checks the pruned subquery (line 122). So a
side that only carries a UDF result, in a column the filter doesn't read, is
rejected for nothing. PySpark code hits this easily: a dim with
`withColumn("x", py_udf(...))`, hinted and joined on a plain key.
I measured it with this query:
```sql
SELECT /*+ RUNTIME_FILTER(t) */ f.v, t.x
FROM fact f JOIN (SELECT k - 100 AS k, py_identity(k) AS x FROM dim WHERE
attr = 'x') t
ON f.k = t.k
```
- With the hint, it gets no Bloom filter and the "contains a Python UDF"
warning.
- Without the hint, the heuristics build one, so adding the hint removes a
filter.
- With `k + 1 AS x` instead of the UDF, the hinted query gets its filter.
Here I tried skipping the whole-side check and reporting the same reason
when `injectFilter` returns the side unchanged:
```scala
runtimeFilterHintRejection(applicationSideKey, creationSide,
creationSideKey,
checkFunctions = false) match {
case Some(reason) =>
hintBlocked(reason)
currentApplicationSide
case None =>
val injected = injectFilter(applicationSideKey,
currentApplicationSide,
FilterCreationSide(creationSideKey, creationSide,
useMaterializedThreshold = false, hinted = true))
if (injected eq currentApplicationSide) {
hintBlocked("the hinted side contains a Python UDF, Python
UDTF or external " +
"UDF, which a runtime filter cannot evaluate")
}
injected
}
```
`checkFunctions = false` skips only the `containsExtractedFunction(plan)`
clause in `RuntimeFilterSourceAnalysis.rejection`, and the key check stays.
With this change:
- The query gets its Bloom filter and returns the same 200,000 rows.
- `InjectRuntimeFilterSuite` passes, the Python UDTF test included.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -578,3 +640,286 @@ trait JoinSelectionHelper extends Logging {
conf.getConfString("spark.sql.join.forceApplyShuffledHashJoin", "false")
== "true"
}
}
+
+/**
+ * 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 rows the
source produces and the
+ * key values they carry 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, and a leaf
+ * such as an RDD that is not checkpointed may return different rows on each
computation.
+ *
+ * The plan is walked bottom-up, tracking:
+ * 1. output attributes whose values are unstable: they come from a
non-deterministic expression,
+ * an order-dependent aggregate or window function, a subquery whose plan
fails this same
+ * analysis, or an expression over such an attribute;
+ * 2. whether the row multiplicity is unstable, i.e. every row is still
present but may be
+ * repeated a different number of times, which an outer generate with an
unstable generator,
+ * or an aggregate over an unstable value, causes and which makes every
aggregate over the
+ * rows unstable.
+ *
+ * The plan is rejected outright when its row set is unstable (a filter or
join condition consumes
+ * an unstable attribute, an inner generate uses an unstable generator, any
limit, offset or
+ * sample), when a leaf cannot promise the same rows again, and, each with its
own reason, when it
+ * calls a function the filter subquery cannot evaluate, when an operator
would duplicate an
+ * observation or 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 {
+
+ /**
+ * @param unstable output attributes whose values depend on evaluation order
or on chance.
+ * @param unstableMultiplicity whether rows may be repeated a different
number of times per
+ * evaluation, while each row stays present.
+ */
+ private case class Taint(unstable: AttributeSet, unstableMultiplicity:
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 (containsExtractedFunction(plan) ||
+ // A DPP filter evaluates the key in its own subquery, which no UDF
extraction rewrites.
+ key.containsAnyPattern(PYTHON_UDF, EXTERNAL_UDF)) {
+ Some("the hinted side contains a Python UDF, Python UDTF or external
UDF, which a runtime " +
+ "filter cannot evaluate")
+ } else {
+ analyze(plan) match {
+ case Left(reason) => Some(reason)
+ case Right(t) if !isStable(key, t.unstable) => Some(NotRepeatable)
+ case _ => None
+ }
+ }
+ }
+
+ def isRepeatableExpression(e: Expression): Boolean = isStable(e,
AttributeSet.empty)
+
+ def containsExtractedFunction(plan: LogicalPlan): Boolean = {
+ plan.containsAnyPattern(PYTHON_UDF, EXTERNAL_UDF) || plan.exists {
+ case g: Generate => g.generator.isInstanceOf[PythonUDTF]
+ case _ => false
+ }
+ }
+
+ /**
+ * 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 => !isRepeatableSubquery(s)
+ case _ => false
+ }
+ }
+
+ /**
+ * A subquery is repeatable when its row set is, and so is whatever of its
output the subquery
+ * exposes: the build keys of a DPP filter, nothing but existence for
EXISTS, the output columns
+ * otherwise. A scalar subquery fails on more than one row, so its
multiplicity counts too.
+ */
+ private def isRepeatableSubquery(s: SubqueryExpression): Boolean = {
+ !s.plan.isStreaming && analyze(s.plan).exists { t =>
+ s match {
+ case d: DynamicPruningSubquery =>
+ d.broadcastKeyIndices.forall(i => isStable(d.buildKeys(i),
t.unstable))
+ case _: Exists => true
+ case _: ScalarSubquery =>
+ !t.unstableMultiplicity &&
s.plan.outputSet.intersect(t.unstable).isEmpty
+ case _ => s.plan.outputSet.intersect(t.unstable).isEmpty
+ }
+ }
+ }
+
+ /** 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 {
+ // Whether a materialized leaf returns the same rows is known only once it
is materialized.
+ case m: MaterializedLeafNode if m.materializedMetadata.isEmpty =>
+ Left(s"the hinted side reads ${m.nodeName}, which is not materialized
yet")
+
+ // The source may have absorbed an operator from the hinted side, so name
that rather than the
+ // scan the user did not write.
+ case s: DataSourceV2ScanRelation if !s.isOutputRepeatable =>
+ Left(s"the hinted side reads ${s.name} with an aggregate, limit, offset,
top-N or sample " +
+ "pushed into the source, which may return different rows when
evaluated again")
+
+ case l: LeafNode =>
+ if (l.isOutputRepeatable) {
+ Right(Taint(AttributeSet.empty))
+ } else {
+ Left(s"the hinted side reads ${l.nodeName}, which may return different
rows when " +
Review Comment:
**Finding 7.** `InlineCTE` keeps a CTE when it is `MATERIALIZED` or not
deterministic. The kept CTE reaches this branch as a `CTERelationRef` leaf,
whose `isOutputRepeatable` is the default `false`. So this query gets no filter:
```sql
WITH d AS MATERIALIZED (SELECT k FROM dim WHERE attr = 'x')
SELECT /*+ RUNTIME_FILTER(d) */ f.v FROM fact f JOIN d ON f.k = d.k
```
It logs "the hinted side reads CTERelationRef, which may return different
rows when evaluated again". Without `MATERIALIZED`, the same query gets its
filter. Rejecting a non-deterministic CTE is right, but a `MATERIALIZED` one is
as repeatable as its definition. The analysis could follow the reference to the
definition in the enclosing `WithCTE`, though I haven't tried it. Otherwise
`CTERelationRef` could get its own reason, e.g. "the hinted side reads a CTE
that is not inlined", so the user can act on it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
}
+ currentApplicationSide
+ } else if (applicationHinted) {
+ // Unlike the heuristics below, a hinted side adds its own filter
even on a key that
Review Comment:
**Finding 8.** This branch skips `hasBloomFilter`, so a hinted join builds
one filter per key pair. That includes two pairs with the same expression on
the application side:
```sql
SELECT /*+ RUNTIME_FILTER(d) */ f.v FROM fact f JOIN dim2 d ON f.k = d.k AND
f.k = d.k2
```
This stacks two `might_contain(..., xxhash64(k))` on `f.k`. Each one takes a
slot of `runtimeFilter.number.threshold` and a probe per row. Skipping a key
that got its filter earlier in this loop keeps what this comment is for, a
filter on a key that already has one from another join:
```scala
hasBloomFilter(currentApplicationSide, applicationSideKey) &&
!hasBloomFilter(applicationSide, applicationSideKey)
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -302,6 +350,11 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
* - The filter creation side has a selective predicate, or its exact
materialized row count
* is smaller than the application side's distinct join-key count
* - The max filterApplicationSide scan size is greater than a configurable
threshold
+ *
+ * All three predict whether a filter pays off, so a [[RuntimeFilterHint]]
on the creation side
+ * (`hinted`) waives them: the creation side is used as it is, whatever its
shape, as long as it
+ * is a repeatable source, since the filter subquery re-executes it. Returns
the reason when no
+ * creation side is extracted; only a hinted one is reported.
Review Comment:
**Finding 9.** This paragraph describes a `hinted` parameter and a returned
reason. The method has neither. Hinted sides never reach it: they go through
`tryInject` and `runtimeFilterHintRejection`.
```suggestion
*
* A [[RuntimeFilterHint]] bypasses this method, which only estimates
whether a filter pays
* off: `tryInject` builds the hinted filter directly once
`runtimeFilterHintRejection` accepts
* the creation side.
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
Review Comment:
**Finding 10.** These two flags read inverted at the call sites. For the
left application side, the call passes `injectLeftHinted, injectRightHinted`.
So `creationHinted` is true exactly when the application side is the hinted
source and must not be filtered. And `applicationHinted` is true when the
creation side is hinted, which is the case the list above describes ("unless
the creation side is hinted"). A name such as `isHintedDirection` /
`isHintedSource` would make a swapped argument visible.
Separately, the `false` that `var appliedHint` starts with (line 506) is
overwritten at line 582 before anything reads it, so the variable can be
declared there.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
Review Comment:
**Finding 11.** `canPruneLeft` and `canPruneRight` (`joins.scala:467-475`)
don't list `Cross`, and `CROSS JOIN ... WHERE f.k = d.k` stays a `Cross` join
with an equi condition. So `SELECT /*+ RUNTIME_FILTER(d) */ f.v FROM fact f
CROSS JOIN dim2 d WHERE f.k = d.k` gets no filter and the warning "the left
side of a cross join cannot be pruned". The same query as an `INNER JOIN` gets
one. Matching `_: InnerLike` in both helpers would fix the hint and the
heuristics alike.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,56 +484,191 @@ 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: Join if isRuntimeFilterHintDecided(join.hint) => join
+ // Only a hinted join has anything to do with Bloom filters disabled.
+ case join: Join if !bloomFilterEnabled &&
!hasRuntimeFilterHint(join.hint) => join
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
- 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)) {
- val oldLeft = newLeft
- val oldRight = newRight
- // Check if:
- // 1. The current join type supports prune the left side with
runtime filter
- // 2. The current join is a shuffle join or a broadcast join that
- // has a shuffle below it
- // 3. There is no bloom filter on the left key yet
- val hasShuffle = isProbablyShuffleJoin(left, right, hint)
- if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
+ // 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)) {
+ reportHintNotApplied(join,
+ "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, or `currentApplicationSide` when no filter was
added, recording the
+ // reason when this direction is the hinted one. 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, and
the key is not fixed
+ // by a literal, unless the creation side is hinted
+ def tryInject(
+ applicationSide: LogicalPlan,
+ currentApplicationSide: LogicalPlan,
+ applicationSideKey: Expression,
+ creationSide: LogicalPlan,
+ creationSideKey: Expression,
+ canPrune: Boolean,
+ applicationHinted: Boolean,
+ creationHinted: Boolean,
+ sideName: String): LogicalPlan = {
+ if (!canPrune) {
+ if (applicationHinted) {
+ // `ExistenceJoin.sql` throws, as the type never appears in SQL.
+ val joinDescription = joinType match {
+ case _: ExistenceJoin => "an existence"
+ case _ => s"a ${joinType.sql.toLowerCase(Locale.ROOT)}"
}
+ hintBlocked(s"the $sideName side of $joinDescription join cannot
be pruned")
}
+ currentApplicationSide
+ } else if (applicationHinted) {
+ // Unlike the heuristics below, a hinted side adds its own filter
even on a key that
+ // already has one.
+ runtimeFilterHintRejection(applicationSideKey, creationSide,
creationSideKey) match {
+ case Some(reason) =>
+ hintBlocked(reason)
+ currentApplicationSide
+ case None =>
+ injectFilter(applicationSideKey, currentApplicationSide,
FilterCreationSide(
+ creationSideKey, creationSide, useMaterializedThreshold =
false, hinted = true))
+ }
+ } else if (creationHinted || !(hasShuffle ||
probablyHasShuffle(applicationSide)) ||
+ hasBloomFilter(currentApplicationSide, applicationSideKey)) {
+ currentApplicationSide
+ } else {
+ extractBeneficialFilterCreatePlan(
+ applicationSide, creationSide, applicationSideKey,
creationSideKey)
+ // Checked last, as computing the constraints may be costly.
+ .filterNot(_ => isFixedByLiteral(applicationSide,
applicationSideKey))
+ .map(injectFilter(applicationSideKey, currentApplicationSide, _))
+ .getOrElse(currentApplicationSide)
+ }
+ }
+ // A DPP filter prunes by whole partitions rather than by rows, so it
is preferred: a key
+ // it prunes gets no Bloom filter, while the other keys still may. The
hint is honored by
+ // a DPP filter that [[PartitionPruning]] marked as honoring it. A
hinted side that cannot
+ // honor the hint on a key gets DPP from its own estimates, as without
the hint, so such a
+ // DPP filter does not count as the hint applied.
+ val prunedKeys = leftKeys.lazyZip(rightKeys).map { (l, r) =>
Review Comment:
**Finding 12.** `runtimeFilterHintRejection` runs the full
`RuntimeFilterSourceAnalysis.analyze` walk of the hinted side, subquery plans
included, for every key pair:
- once in `PartitionPruning.honorsHint`;
- again here, in `prunedKeys` and in `tryInject`.
Only the final `isStable(key, ...)` depends on the key. Computing the side's
`Either[String, Taint]` once per join, then checking each key against it, would
do the same work in one pass.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,29 +432,49 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, 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)
+ /**
+ * The DPP filters on `applicationKey` built from `creationKey` that prune
`plan`. This rule runs
+ * after the DPP filters are pushed down and cleaned up, so only a surviving
one is found,
+ * wherever the pushdown left it; it is the one [[PartitionPruning]]
inserted for the same join
+ * key pair, whose pruning key it rewrote on the way down as the lineage
tracking does.
+ */
+ private def dynamicPruningFilters(
+ plan: LogicalPlan,
+ applicationKey: Expression,
+ creationKey: Expression): Seq[DynamicPruningSubquery] = {
+ if (!plan.containsPattern(DYNAMIC_PRUNING_SUBQUERY)) return Nil
+ findExpressionAndTrackLineageDown(applicationKey, plan).toSeq.flatMap {
case (trackedKey, _) =>
+ plan.flatMap {
+ case Filter(condition, _) => condition.collect {
+ case d: DynamicPruningSubquery if
d.pruningKey.semanticEquals(trackedKey) &&
+ d.broadcastKeyIndices.exists(i =>
d.buildKeys(i).semanticEquals(creationKey)) => d
+ }
+ case _ => Nil
+ }
+ }
+ }
+
+ /**
+ * Whether `plan` constrains `key` to a literal, e.g. a partition key fixed
by a static partition
+ * filter. Every row then has the same key, so a filter on it prunes next to
nothing.
+ */
+ private def isFixedByLiteral(plan: LogicalPlan, key: Expression): Boolean = {
Review Comment:
**Finding 3.** `isFixedByLiteral` came from the partition-key question in
the [decision
challenge](https://github.com/apache/spark/pull/58635#pullrequestreview-5611231620)
and
[r4346072936](https://github.com/apache/spark/pull/58635#discussion_r4346072936).
It reads all of the application side's constraints, though. So it also skips a
data column fixed by a literal, whether the literal is written or inferred
through the join. The base builds a Bloom filter there. The SPARK-58272 test
shows it: its `c2 = 8` had to become `c2 IN (8, 9)`.
Every application row has the same key, so the filter keeps all of them or
none:
- When the creation side has the literal, the filter keeps every row, and
skipping it saves its cost.
- When the creation side doesn't have it, the filter drops every row, and
the skip loses that.
I measured the second case with shuffle joins and
`applicationSideScanSizeThreshold = 0`:
- `fact(k = id % 10, v)`, 200k rows.
- `dim(k = id % 10 + 100, attr)`, 1k rows, with `attr = 'x'` when `id % 100
= 0`.
| fact rows shuffled | base, AQE off / on | head, AQE off / on |
|---|---|---|
| `fact f JOIN dim d ON f.k = d.k WHERE f.k = 5 AND d.attr = 'x'` | 0 / 0 |
20,000 / 0 |
| `dim d LEFT JOIN (SELECT * FROM fact WHERE k = 5) f ON d.k = f.k WHERE
d.attr = 'x'` | 0 / 0 | 20,000 / 20,000 |
The base has a Bloom filter in all four runs, and the head in none. Both
return the same rows (0 and 10).
- In the inner join, `d.k = 5` is inferred for `dim`, so `dim` is empty and
AQE drops the fact scan.
- In the left outer join nothing is inferred for `dim`, which keeps its 10
rows, so AQE can't help.
The filter's cost is bounded, since its creation side has to pass the size
threshold. What it saves can be the whole fact side's shuffle. So I'd drop the
skip, which gives a partition key fixed by a literal a Bloom filter too.
Without the skip, only `do not add bloom filter on a key fixed by a literal`
fails. If the partition-key plans should stay as they were, limit the skip to
the keys whose DPP filter `CleanupDynamicPruningFilters` drops for a literal,
e.g. by having that rule record them. Either way, the description and this doc
comment shouldn't say the filter prunes next to nothing.
--
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]
