pan3793 commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r4260743904
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1270,6 +1325,600 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
}
}
+ test("RUNTIME_FILTER hint waives the creation side's selective predicate
requirement") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val query = "select * from bf1 join bf2 on bf1.c1 = bf2.c2"
+ // Without a selective predicate on either side there is no evidence a
filter pays off.
+ assertDidNotRewriteWithBloomFilter(query)
+ // The hint supplies that evidence, and picks which side to build from.
+ assertRewroteWithBloomFilter(
+ "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2")
+ assert(bloomFilterApplicationSideKeys(
+ "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2") == Set("c1"))
+ assert(bloomFilterApplicationSideKeys(
+ "select /*+ RUNTIME_FILTER(bf1) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2") == Set("c2"))
+ }
+ }
+
+ test("RUNTIME_FILTER hint waives the application side's scan size
threshold") {
+ // A threshold above the application side's scan size, so only the hint
can trigger a rewrite.
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "1GB",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ assertDidNotRewriteWithBloomFilter(
+ "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62")
+ assertRewroteWithBloomFilter(
+ "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2")
+ }
+ }
+
+ test("RUNTIME_FILTER hint waives the creation side's size threshold") {
+ // A threshold below any creation side's size, so only the hint can
trigger a rewrite. The
+ // threshold estimates whether the filter is worth building; the filter's
own size is bounded
+ // by the max number of bits regardless.
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.RUNTIME_BLOOM_FILTER_CREATION_SIDE_THRESHOLD.key -> "1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ assertDidNotRewriteWithBloomFilter(
+ "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62")
+ assertRewroteWithBloomFilter(
+ "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2")
+ }
+ }
+
+ test("RUNTIME_FILTER hint waives the shuffle requirement") {
+ // `bf2` is below the broadcast threshold, so this is a broadcast join
with no shuffle below
+ // the application side, where a filter is not expected to pay off.
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000") {
+ assertDidNotRewriteWithBloomFilter(
+ "select * from bf1 join bf2 on bf1.c1 = bf2.c2 where bf2.a2 = 62")
+ assertRewroteWithBloomFilter(
+ "select /*+ RUNTIME_FILTER(bf2) */ * from bf1 join bf2 on bf1.c1 =
bf2.c2")
+ }
+ }
+
+ test("RUNTIME_FILTER hint waives the application side's single-leaf lineage
requirement") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // The application side's key is an aggregate result, which cannot be
traced to a scan. The
+ // heuristics use that lineage to predict that the filter reaches a
scan; the filter above
+ // the aggregate is valid regardless, and prunes the aggregate's output
before the join.
+ val aggregatedApplicationSide =
+ """
+ |FROM (SELECT max(c1) AS c1 FROM bf1 GROUP BY b1) t
+ | JOIN bf2
+ | ON t.c1 = bf2.c2
+ """.stripMargin
+ assertDidNotRewriteWithBloomFilter(s"SELECT * $aggregatedApplicationSide
WHERE bf2.a2 = 62")
+ assertRewroteWithBloomFilter(
+ s"SELECT /*+ RUNTIME_FILTER(bf2) */ * $aggregatedApplicationSide")
+ }
+ }
+
+ test("RUNTIME_FILTER hint accepts a creation side the heuristics reject") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // Left to itself the rule looks for a selective predicate over a scan,
a search an
+ // `Aggregate` stops, so this is not rewritten despite the selective
predicate below it. The
+ // hint designates the creation side instead of searching for one, so
the shape stops
+ // mattering.
+ val distinctSource =
+ """
+ |FROM bf1
+ | JOIN (SELECT DISTINCT c2 FROM bf2 WHERE a2 = 62) t
+ | ON bf1.c1 = t.c2
+ """.stripMargin
+ assertDidNotRewriteWithBloomFilter(s"SELECT * $distinctSource")
+ assertRewroteWithBloomFilter(s"SELECT /*+ RUNTIME_FILTER(t) */ *
$distinctSource")
+ // Likewise for a `Limit` the search does not descend through, here a
top-n of grouping
+ // keys, whose order is total because the keys are unique.
+ assertRewroteWithBloomFilter(
+ """
+ |SELECT /*+ RUNTIME_FILTER(t) */ *
+ |FROM bf1
+ | JOIN (SELECT c2 FROM bf2 WHERE a2 = 62 GROUP BY c2 ORDER BY
c2 LIMIT 5) t
+ | ON bf1.c1 = t.c2
+ """.stripMargin)
+ }
+ }
+
+ test("RUNTIME_FILTER hint builds a filter from a computed join key") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // The filter is built from the hinted side's output, so a join key the
creation side computes
+ // rather than reads from a scan -- here an aggregate result -- works
like any other.
+ assertRewroteWithBloomFilter(
+ """
+ |SELECT /*+ RUNTIME_FILTER(t) */ *
+ |FROM bf1
+ | JOIN (SELECT max(c2) AS c2 FROM bf2 GROUP BY b2) t
+ | ON bf1.c1 = t.c2
+ """.stripMargin)
+ }
+ }
+
+ test("RUNTIME_FILTER hint takes effect through one mechanism") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
+ // `bf5part` is partitioned by `f5`, and the selective `bf2.a2 = 62`
makes DPP prune it on
+ // that key. The hint is honored by that DPP filter, so no Bloom filter
is added: not on the
+ // key DPP already prunes, matching the unhinted behavior asserted
above, and not on the
+ // other join key either, where the heuristics would add one ("add bloom
filter if dpp
+ // filter exists on a different column" above).
+ val oneKey = "select /*+ RUNTIME_FILTER(bf2) */ * from bf5part join bf2
" +
+ "on bf5part.f5 = bf2.c2 where bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(oneKey)
+ assert(hasDynamicPruning(oneKey))
+ val twoKeys = "select /*+ RUNTIME_FILTER(bf2) */ * from bf5part join bf2
" +
+ "on bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(twoKeys)
+ assert(hasDynamicPruning(twoKeys))
+ }
+ }
+
+ test("RUNTIME_FILTER hint is honored by a DPP filter only when that filter
reaches the scan") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ spark.udf.register("bf_nondeterministic_true", udf((_: Int) =>
true).asNondeterministic())
+ // The hint gets a DPP filter on `bf5part`'s partition key, but a
barrier on the pruned side
+ // keeps it from being pushed to the scan, so it is dropped later: a
non-deterministic
+ // filter, or a window whose partitioning does not cover the key. The
Bloom filter needs no
+ // pushdown, so the hint is honored with one instead, and exactly one of
the two remains.
+ Seq(
+ ("SELECT f5 FROM bf5part WHERE bf_nondeterministic_true(f5)", 1),
+ ("SELECT f5 FROM (SELECT f5, row_number() OVER (PARTITION BY c5 ORDER
BY f5) rn " +
+ "FROM bf5part) WHERE rn = 1", 1),
+ // The DPP filter is pushed into both union branches, where the limit
blocks it and the
+ // `Range` cannot use it. The Bloom filter is pushed into both
branches too.
+ ("SELECT f5 FROM (SELECT f5 FROM bf5part ORDER BY f5 LIMIT 2) " +
+ "UNION ALL SELECT CAST(id AS INT) AS f5 FROM range(3)", 2)
+ ).foreach { case (prunedSide, numBloomFilters) =>
+ val query = "SELECT /*+ RUNTIME_FILTER(bf2) */ f.f5, bf2.c2 " +
+ s"FROM ($prunedSide) f JOIN bf2 ON f.f5 = bf2.c2"
+ assertRewroteWithBloomFilter(query, numBloomFilters)
+ assert(!hasDynamicPruning(query))
+ }
+ // A DPP filter that does reach the scan honors the hint on its own,
even when the pruned
+ // side is not deterministic elsewhere (here on the other side of a join
the DPP filter is
+ // pushed past), so no Bloom filter is added on top of it.
+ val reaching = "SELECT /*+ RUNTIME_FILTER(bf2) */ f.f5, f.nd, bf2.c2
FROM " +
+ "(SELECT p.f5, b.nd FROM bf5part p JOIN " +
+ "(SELECT c3, bf_nondeterministic_true(a3) AS nd FROM bf3) b ON p.c5 =
b.c3) f " +
+ "JOIN bf2 ON f.f5 = bf2.c2"
+ assertDidNotRewriteWithBloomFilter(reaching)
+ assert(hasDynamicPruning(reaching))
+ }
+ }
+
+ test("RUNTIME_FILTER hint is ignored for a creation side that is not
repeatable") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // The filter subquery re-executes the creation side, so one that can
produce different rows
+ // or keys on re-evaluation could yield keys the join itself never sees
and prune rows that
+ // do match. Catalyst flags most of these as deterministic.
+ val reason = "the hinted side may produce different rows or join keys
when evaluated again"
+ spark.udf.register("bf_nondeterministic_array",
+ udf((i: Int) => Seq(i)).asNondeterministic())
+ def query(source: String): String =
+ s"SELECT /*+ RUNTIME_FILTER(t) */ * FROM bf1 JOIN ($source) t ON
bf1.c1 = t.c2"
+ Seq(
+ // A non-deterministic expression.
+ "SELECT c2 FROM bf2 WHERE rand() < 0.5",
+ // A limit keeps whichever rows arrive first: unordered, or ordered on
a key that is not
+ // unique, so that ties at the cutoff can be resolved differently.
+ "SELECT c2 FROM bf2 WHERE a2 = 62 LIMIT 5",
+ "SELECT c2 FROM bf2 WHERE a2 = 62 ORDER BY c2 LIMIT 5",
+ // A sample without a seed draws a different sample per evaluation.
+ "SELECT c2 FROM bf2 TABLESAMPLE (50 PERCENT)",
+ // A key computed by an order-dependent aggregate function.
+ "SELECT first(c2) AS c2 FROM bf2 GROUP BY b2",
+ "SELECT any_value(c2) AS c2 FROM bf2 GROUP BY b2",
+ // A key computed by a window function.
+ "SELECT row_number() OVER (ORDER BY c2) AS c2 FROM bf2",
+ // An inner generate drops the rows for which an unstable generator
yields nothing.
+ "SELECT c2 FROM bf2 LATERAL VIEW
explode(bf_nondeterministic_array(a2)) x AS v",
+ // A key computed by a subquery whose own plan is not repeatable.
+ "SELECT (SELECT max(c2) FROM bf2 TABLESAMPLE (50 PERCENT)) AS c2 FROM
bf2",
+ // An outer generate keeps every row but repeats it as many times as
the unstable
+ // generator yields values, so an aggregate over the rows is unstable.
+ "SELECT count(*) AS c2 FROM bf2 " +
+ "LATERAL VIEW OUTER explode(bf_nondeterministic_array(a2)) x AS v",
+ // An average of integers accumulates in floating point, which is not
associative.
+ "SELECT avg(c2) AS c2 FROM bf2 GROUP BY b2"
+ ).foreach { source =>
+ assertNoBloomFilters(query(source))
+ assertHintNotApplied(query(source), reason)
+ }
+ // A seeded sample over a scan is repeatable.
+ assertRewroteWithBloomFilter(
+ query("SELECT c2 FROM bf2 TABLESAMPLE (50 PERCENT) REPEATABLE (42)"))
+ }
+ }
+
+ test("RUNTIME_FILTER hint accepts a creation side whose unstable values do
not reach the key") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // Only the key values and the row set have to be repeatable. An
order-dependent value that
+ // is merely carried to the output, next to a stable key, does not
disqualify the side, and
+ // neither does a filter on a window aggregate whose frame is the whole
partition.
+ def query(source: String, key: String): String =
+ s"SELECT /*+ RUNTIME_FILTER(t) */ * FROM bf1 JOIN ($source) t ON
bf1.c1 = t.$key"
+ spark.udf.register("bf_nondeterministic_array",
+ udf((i: Int) => Seq(i)).asNondeterministic())
+ Seq(
+ ("SELECT b2, first(c2) AS c2 FROM bf2 GROUP BY b2", "b2"),
+ ("SELECT c2, v FROM bf2 LATERAL VIEW OUTER
explode(bf_nondeterministic_array(a2)) x AS v",
+ "c2"),
+ ("SELECT (SELECT max(c2) FROM bf2) AS c2, b2 FROM bf2", "c2"),
+ ("SELECT sum(c2) AS c2 FROM bf2 GROUP BY b2", "c2"),
+ ("SELECT c2, sum(a2) OVER (PARTITION BY b2 ORDER BY c2 " +
+ "ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS s FROM
bf2", "c2"),
+ ("SELECT c2 FROM (SELECT * FROM bf2) TABLESAMPLE (50 PERCENT)
REPEATABLE (42) x", "c2"),
+ ("SELECT c2, row_number() OVER (ORDER BY a2) AS rn FROM bf2", "c2"),
+ ("SELECT c2, sum(a2) OVER (PARTITION BY b2) AS s FROM bf2", "c2"),
+ ("SELECT c2 FROM (SELECT c2, sum(a2) OVER (PARTITION BY b2) AS s FROM
bf2) WHERE s > 0",
+ "c2")
+ ).foreach { case (source, key) =>
+ assertRewroteWithBloomFilter(query(source, key))
Review Comment:
Fixed in cbef594a838. Hinted tests join `bf4`, which shares `bf2`'s keys,
and each accepting test must return rows matching the unfiltered answer.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,55 +481,158 @@ 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 join.getTagValue(HINT_DECIDED).isDefined => join
Review Comment:
Fixed in cbef594a838. When the tag is lost, a Bloom filter already built
from the hinted side's key counts as the hint applied; tested with the
IN-subquery shape.
##########
docs/sql-ref-syntax-qry-select-hints.md:
##########
@@ -172,6 +172,40 @@ 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
+
+Runtime filter hints allow users to suggest which side of a join Spark should
build a runtime
+filter from, to prune the other side. Spark decides on its own whether a
runtime filter is worth
+building, based on statistics; the hint overrides that decision, but not the
requirements that make
Review Comment:
Fixed in cbef594a838. The docs state the key-set contract, that configured
limits may decline the hint, and that there is no opposite-direction fallback.
The `InjectRuntimeFilter` scaladoc now lists the simple-key requirement as
retained.
##########
docs/sql-ref-syntax-qry-select-hints.md:
##########
@@ -172,6 +172,55 @@ SELECT /*+ SHUFFLE_REPLICATE_NL(t1) */ * FROM t1 INNER
JOIN t2 ON t1.key = t2.ke
SELECT /*+ BROADCAST(t1), MERGE(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key =
t2.key;
```
+### Runtime Filter Hints
+
+A runtime filter prunes one side of a join using the join key values found on
the other side, so
+rows that cannot match are discarded early. Spark decides on its own whether
such a filter is worth
+building, based on estimates of how much data it would save. Runtime filter
hints let users make
+that decision instead, for the cases where the estimates are unavailable or
wrong.
+
+#### Runtime Filter Hints Types
+
+* **RUNTIME_FILTER**
+
+ Suggests that Spark build a runtime filter from the hinted relation and
use it to prune the
+ other side of the join. Use it when the hinted side is known to match only
a small fraction
+ of the other side, but Spark does not choose a runtime filter on its own,
typically because
+ table statistics are missing or misleading. The hinted side may be any
relation or subquery,
+ and is never itself pruned. The hint does not choose how the pruning is
done; Spark picks the
+ mechanism. `RUNTIME_FILTER` can be combined with a join strategy hint.
+
+The hint overrides Spark's cost estimates, but not the requirements that make
a runtime filter
+correct, so Spark is not guaranteed to follow it. A side that join semantics
forbid pruning is
+never pruned, e.g. the left side of a `LEFT OUTER` join, whose rows must all
appear in the output.
+The hinted side must produce the same rows each time it is evaluated, since
building the filter
Review Comment:
Fixed in cbef594a838.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -578,3 +623,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:
+ * - 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;
+ * - 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 causes and
+ * which makes every aggregate over the rows unstable.
+ * 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 repeatable leaf, a limit is over
anything but a total order),
+ * when a leaf cannot promise the same rows again, and, each with its own
reason, 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 extends AliasHelper {
+
+ /**
+ * @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.
+ * @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,
+ unstableMultiplicity: Boolean = false,
+ 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 {
+ 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)
+
+ /**
+ * 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.
+ */
+ 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 _ => 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 {
+ 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. The groups
themselves occur
+ // once each, whatever the input multiplicity.
+ val stableGroups = a.groupingExpressions.forall(isStable(_, t.unstable))
+ val unstable = a.aggregateExpressions.filter { e =>
+ !isStable(e, t.unstable) ||
+ (e.exists(_.isInstanceOf[AggregateExpression]) &&
+ (!stableGroups || t.unstableMultiplicity || !isOrderIrrelevant(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 || t.unstableMultiplicity || !isStable(e,
t.unstable) ||
+ !isOrderIrrelevantWindow(e)
+ }
+ t.copy(unstable = t.unstable ++
AttributeSet(unstable.map(_.toAttribute)),
+ totallyOrdered = false)
+ }
+
+ 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.copy(totallyOrdered = false))
+ } else if (g.outer) {
+ Right(Taint(t.unstable ++ AttributeSet(g.generatorOutput),
unstableMultiplicity = true))
+ } 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), t.unstableMultiplicity)
+ }
+
+ case s: Sort => analyze(s.child).map { t =>
+ val stableOrder = s.order.forall(o => isStable(o.child, t.unstable))
+ t.copy(totallyOrdered = s.global && stableOrder &&
!t.unstableMultiplicity &&
+ 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 and partitioning even then: only a seeded sample over a
repeatable leaf, through
+ // projections and filters that keep both, is repeatable. A predicate with
a subquery is not
+ // such a filter, as its plan may differ between the two evaluations.
+ case s: Sample =>
+ @tailrec
+ def isOrderPreservingScan(p: LogicalPlan): Boolean = p match {
+ case Project(projectList, child) if
projectList.forall(_.deterministic) =>
+ isOrderPreservingScan(child)
+ case Filter(condition, child)
+ if condition.deterministic &&
!SubqueryExpression.hasSubquery(condition) =>
+ isOrderPreservingScan(child)
+ case l: LeafNode => l.isOutputRepeatable
+ case _ => false
+ }
+ if (s.seed.isDefined && isOrderPreservingScan(s.child)) {
+ analyze(s.child)
+ } else {
+ Left(NotRepeatable)
+ }
+
+ // Each distinct row occurs once, whatever the input multiplicity.
+ case d: Distinct => analyze(d.child).map(t => Taint(t.unstable))
+
+ // The rows and their values are unchanged; a shuffle loses the order.
+ case _: SubqueryAlias | _: Repartition | _: RepartitionByExpression | _:
RebalancePartitions =>
+ analyze(plan.children.head).map(t => t.copy(totallyOrdered = false))
+
+ // The filter would evaluate the observation a second time under the same
name.
+ case _: CollectMetrics =>
+ Left("the hinted side contains CollectMetrics, whose observation the
runtime filter " +
+ "would evaluate twice")
+
+ // 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 expression's value is independent of the order its input rows
arrive in. Only
+ * aggregate functions whose result is exact whatever the accumulation order
qualify: floating
+ * point sums are not associative, so a sum over a floating point input, an
average with a
+ * floating point accumulator (every non-decimal, non-interval input) and
the central moments
+ * are order-dependent. A Bloom filter aggregate, which
[[InjectRuntimeFilter]] injects for an
+ * inner join, merges commutatively.
+ */
+ private def isOrderIrrelevant(e: Expression): Boolean = e match {
+ case _: AttributeReference => true
+ case ae: AggregateExpression =>
isOrderIrrelevantFunction(ae.aggregateFunction)
+ case _: UserDefinedExpression => false
+ case _ => e.children.forall(isOrderIrrelevant)
+ }
+
+ private def isOrderIrrelevantFunction(f: AggregateFunction): Boolean = f
match {
+ case _: Min | _: Max | _: Count | _: BitAggregate | _:
BloomFilterAggregate => true
+ case s: Sum => !isFloatingPoint(s.child.dataType)
+ case a: Average => !isFloatingPoint(a.sumDataType)
+ case _ => false
+ }
+
+ private def isFloatingPoint(dataType: DataType): Boolean = dataType match {
+ case FloatType | DoubleType => true
+ case _ => false
+ }
+
+ /**
+ * A window function's value depends on the row order within its frame,
unless the frame is
+ * the whole partition and the function is order-irrelevant.
+ */
+ private def isOrderIrrelevantWindow(e: NamedExpression): Boolean = e match {
+ case Alias(WindowExpression(_: AggregateExpression, spec), _) =>
+ val wholePartition = spec.frameSpecification match {
+ case SpecifiedWindowFrame(_, UnboundedPreceding, UnboundedFollowing)
=> true
+ case UnspecifiedFrame => spec.orderSpec.isEmpty
+ case _ => false
+ }
+ wholePartition && isOrderIrrelevant(e)
+ case _ => false
+ }
+
+ /**
+ * Whether `sortKeys` cover a key of `plan` that is proven unique, so that
sorting on them is a
+ * total order. The only uniqueness Catalyst can establish is an aggregate's
grouping keys.
Review Comment:
Fixed in cbef594a838. The helper is replaced by `LogicalPlan.distinctKeys`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala:
##########
@@ -77,22 +78,27 @@ object ResolveHints {
* is not aliased differently), subquery, or common table expression that
match the specified
* name.
*
+ * [[RuntimeFilterHint]] takes the same per-relation form, e.g.
"RUNTIME_FILTER(a)", and is
+ * resolved here too, onto the same [[HintInfo]] as a join strategy hint on
that relation, so
+ * the two can accompany each other.
+ *
* The hint resolution works by recursively traversing down the query plan
to find a relation or
* subquery that matches one of the specified relation aliases. The
traversal does not go past
* beyond any view reference, with clause or subquery alias.
*
* This rule must happen before common table expressions.
*/
- object ResolveJoinStrategyHints extends Rule[LogicalPlan] {
+ object ResolveJoinHints extends Rule[LogicalPlan] {
Review Comment:
Fixed in cbef594a838.
--
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]
