cloud-fan commented on code in PR #58635:
URL: https://github.com/apache/spark/pull/58635#discussion_r4250017527
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -285,6 +285,10 @@ case class DataSourceV2ScanRelation(
s"RelationV2$outputString $nameWithTimeTravelSpec"
}
+ // The scan is planned once, and each of its executions reads what that plan
describes, as the
+ // two sides of a self-join do.
+ override def isOutputRepeatable: Boolean = true
Review Comment:
**Non-blocking (P2):** This returns `true` for every scan relation, but
`V2ScanRelationPushDown` removes a fully pushed `LIMIT`, `OFFSET`, TopN or
`TABLESAMPLE` from the plan before this rule runs (JDBC pushes `LIMIT` by
default and reports it as fully pushed for a single-partition read). So `SELECT
/*+ RUNTIME_FILTER(d) */ * FROM fact f JOIN (SELECT k FROM jdbc.db.dim LIMIT
100) d ON f.k = d.k` reaches `RuntimeFilterSourceAnalysis` as a bare trusted
leaf, while the same `LIMIT` over parquet is rejected as not repeatable. The
Bloom filter subquery (or a hinted standalone DPP subquery) and the join each
send their own unordered `LIMIT 100` query, and the database may return
different rows to each, so matching fact rows can be pruned. Scans that
absorbed only filters and columns are not affected.
**Recommended change:** When V2ScanRelationPushDown removes a fully pushed
LIMIT, OFFSET, or TopN, record on the DataSourceV2ScanRelation it builds that
the scan is not repeatable. Conservatively do the same for a pushed sample or
aggregate, and make isOutputRepeatable return that recorded value. Add DSv2
join tests for the hinted Bloom filter and DPP paths.
**Why this works:** A defaulted constructor field on
DataSourceV2ScanRelation carries the absorbed-operator fact from its only
producer to the leaf contract. Data fields survive copies and rebuilds, so the
value reaches RuntimeFilterSourceAnalysis unchanged, and the analysis's
existing leaf case then rejects the scan with the not-repeatable reason.
**Scope:** Carry the absorbed-operator fact from DSv2 pushdown into the scan
relation's repeatability claim.
**Compatibility:** Pushdown results, scan planning, and EXPLAIN output for
DSv2 scans are unchanged.
**Risks:** Adding a constructor field to a widely pattern-matched case class
requires updating positional patterns in rules and tests. Treating every pushed
aggregate as non-repeatable may reject some hints that are safe.
**Constraints:** Do not change what is pushed down or the resulting scan;
change only the repeatability claim. Keep catalyst independent of sql/core scan
implementation classes.
**Success:** A hinted side that reads a DSv2 scan with a fully pushed
unordered LIMIT, OFFSET, or TopN is reported as not applied with the
not-repeatable reason, on both the Bloom filter and DPP paths. A DSv2 scan that
absorbed only filters and columns remains an accepted hinted source.
**Verification:**
- **Regression:** With a V2 JDBC catalog that fully absorbs a LIMIT, a
hinted join on that subquery builds no runtime filter and logs the
not-repeatable warning.
- **Compatibility:** A hinted join on a V2 table scan with only filter
pushdown still gets its Bloom filter, or hinted DPP when the other side is
partitioned, and returns the unhinted rows.
- **Compile:** Every construction and positional pattern of
DataSourceV2ScanRelation remains well-typed with the new defaulted field.
##########
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:
**Nit (P3):** `HINT_DECIDED` is a `TreeNodeTag`, which does not survive a
rebuild of this join. Between the subquery pass and the main-plan pass,
`PartitionPruning` can insert a DPP filter above a hoisted IN subquery (for
example `bf1.c1 IN (SELECT /*+ RUNTIME_FILTER(bf3) */ p.f5 FROM bf5part p JOIN
bf3 ON p.c5 = bf3.c3)` with a selective filter on `bf1`), and
`PushPredicateThroughJoin` in the DPP pushdown batch builds a new
`Join(newLeft, newRight, joinType, newJoinCond, hint)` without the old join's
tags. This rule then revisits the hinted join, finds the Bloom filter it
injected earlier, and warns that the hint was not applied ('a runtime filter on
the join key already exists') although it was. Recording the decision in the
hint data, which every rebuild passes through, would make it durable; it should
stay out of `toString` and EXPLAIN.
**Verification:**
- **Regression:** An IN subquery containing a hinted join, whose semi-join
side receives a DPP filter pushed through that join, produces exactly one Bloom
filter and no not-applied warning.
- **Compatibility:** Existing hint composition and subquery warning-count
tests keep their current results.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -306,10 +316,24 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
* meet the following requirements:
* (1) it can not be a stream
* (2) it needs to contain a selective predicate or a cheaply-recomputable
materialized input
+ *
+ * (2) is evidence that pruning pays off, which a [[RuntimeFilterHint]] on
the filtering side
+ * (`hinted`) supplies directly. A hinted side only has to be a repeatable
source of the
+ * filtering key, see `JoinSelectionHelper.isRepeatableRuntimeFilterSource`,
since DPP
+ * re-evaluates it. It must not contain a Python UDF either: the filtering
side is planned as a
+ * standalone subquery without the Python UDF extraction the main plan gets,
so a Python UDF
+ * could not be evaluated there.
*/
- private def hasPartitionPruningFilter(plan: LogicalPlan): Boolean = {
- !plan.isStreaming &&
- (hasSelectivePredicate(plan) ||
isCheaplyRecomputableMaterializedPlan(plan))
+ private def hasPartitionPruningFilter(
+ plan: LogicalPlan,
+ hinted: Boolean,
+ filteringKey: Expression): Boolean = {
+ if (hinted) {
+ isRepeatableRuntimeFilterSource(plan, filteringKey) &&
!plan.containsPattern(PYTHON_UDF)
Review Comment:
**Non-blocking (P2):** This guard, and the same
`containsPattern(PYTHON_UDF)` check in `InjectRuntimeFilter.injectBloomFilter`,
stand in for 'nothing in the copied side needs the main plan's UDF extraction',
but only `PythonUDF`, `PythonUDAF` and `PythonAggregate` carry `PYTHON_UDF`. A
Python UDTF appears as `Generate(PythonUDTF, ...)`, an `UnevaluableGenerator`
without that pattern, and external UDFs carry `EXTERNAL_UDF`.
`RuntimeFilterSourceAnalysis` accepts a deterministic UDTF generator,
`ExtractPythonUDTFs` and `PlanExternalUDFs` rewrite only the main plan, and the
injected subquery is planned by `QueryExecution.prepareExecutedPlan` without
the optimizer. So `SELECT /*+ RUNTIME_FILTER(u) */ * FROM fact JOIN my_udtf(1)
u ON fact.k = u.k` fails with an internal error when the subquery runs, instead
of being reported as not applied. One shared check in the eligibility predicate
covering `PYTHON_UDF`, `Generate(PythonUDTF)` and `EXTERNAL_UDF`, used by both
paths, would close it.
**Verification:**
- **Regression:** A hinted join with a deterministic Python UDTF side
succeeds, builds no runtime filter, and logs the not-applied warning, for both
a shuffle join and a join on a partitioned table.
- **Behavior:** A hinted side containing a Python UDF is still rejected with
its existing reason.
##########
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)) {
Review Comment:
**Non-blocking (P2):** A seed alone does not make the sample repeatable
here: `SampleExec` seeds each partition with `resolvedSeed + partitionIndex`,
so two evaluations draw the same rows only when they scan the leaf with the
same partitioning, and the filter subquery and the join choose that separately.
For `SELECT /*+ RUNTIME_FILTER(d) */ * FROM fact JOIN (SELECT * FROM
bucketed_dim TABLESAMPLE (50 PERCENT) REPEATABLE (42)) d ON fact.k = d.k` with
`d` broadcast, `DisableUnnecessaryBucketedScan` disables bucketing for the main
plan's scan (nothing above the broadcast build side needs its partitioning) but
keeps it under the Bloom filter subquery's `AllTuples` aggregate, because
`SampleExec` is not an allowed node there. The two samples differ, and fact
rows whose keys appear only in the join's sample are pruned. Physical
partitioning is decided after this rule runs, so rejecting every `Sample` in a
hinted side, and flipping the tests that accept seeded samples, looks like the
saf
e fix.
**Verification:**
- **Regression:** A hinted join over a bucketed table sampled with
REPEATABLE, broadcast as the build side, builds no Bloom filter and returns the
unhinted rows.
- **Behavior:** Seeded-sample sources that the suites previously accepted,
including in a DPP pruning-key subquery, are now reported as not applied.
##########
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:
**Non-blocking (P2):** In this suite's fixtures `bf1.c1` takes only the
values {8, 23, 33, 58, 66, 74, 86, 88}; no `bf2.c2` value (nor the per-`b2` max
or sum built from it) is in that set, and no row has `a2 = 62`. So the
`checkAnswer` inside `assertRewroteWithBloomFilter` compares empty (or, for
outer joins, null-extended) results with and without the filter, here and in
most new hinted tests. They pin plan shape, but would still pass if an accepted
hinted filter dropped every matching row, which is the failure the
repeatability analysis exists to prevent. The hinted tests need fixtures whose
keys overlap, compared against the unhinted result; the existing `bf*` tables
can stay as they are for the unhinted tests.
**Verification:**
- **Behavior:** Each accepting hinted test's query returns at least one
joined row with the filter, and the same rows without it.
- **Regression:** Temporarily forcing a hinted Bloom filter to be built from
a non-repeatable source makes at least one of these tests fail.
##########
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:
**Nit (P3):** Catalyst can establish uniqueness beyond grouping keys:
`LogicalPlan.distinctKeys` tracks it, and this file already uses it at lines
232 and 236. Recognizing only an `Aggregate` below `Project`/`Filter` is this
helper's own conservative choice; saying so would keep a later change from
assuming a broader check is impossible.
**Verification:**
- **Inspection:** The comment's description matches the helper's Project,
Filter, and Aggregate traversal and makes no claim about Catalyst-wide
uniqueness tracking.
##########
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:
**Nit (P3):** After this rename,
`sql/core/src/test/resources/sql-tests/inputs/hints.sql:7` still says join
strategy hints 'are resolved by ResolveHints.ResolveJoinStrategyHints'; it is
the only remaining reference. Updating it to `ResolveHints.ResolveJoinHints`
(which now also resolves `RUNTIME_FILTER`) does not change the golden output.
**Verification:**
- **Inspection:** A repository search finds no remaining
ResolveJoinStrategyHints reference, and the hint golden output is unchanged.
##########
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:
**Nit (P3):** This presents correctness as the only thing that can stop the
hint, but a hinted join is also reported as not applied when
`spark.sql.optimizer.runtimeFilter.number.threshold` is reached, when
`spark.sql.optimizer.runtime.bloomFilter.enabled` is false and DPP does not
apply, when a join key is not a simple expression, and when the hinted side
contains a Python UDF. A user reading this cannot tell why a correct filter was
not built. Two more gaps in this section: 'the same rows and join keys' below
reads as full-row equality, although unstable values carried only in non-key
columns are accepted (e.g. `first(c2)` next to a `GROUP BY b2` key); and
nothing says that a hint that cannot be applied does not make Spark build a
filter in the opposite direction, which the tests pin. The
`InjectRuntimeFilter` class doc also says the hint waives the checks that
estimate payoff, while the simple-key check, which the code describes as a cost
rule, still applies.
**Verification:**
- **Inspection:** Each documented limit and rule matches a check in the
hinted Bloom filter or DPP path, and no retained check is left undocumented.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -644,3 +683,214 @@ private[sql] object NullAwareAntiJoinPlanning extends
JoinSelectionHelper {
}
}
}
+
+/**
+ * Decides whether a plan can serve as the source of a runtime filter on a
join key. A runtime
+ * filter evaluates its source separately from the join, so the key values the
source produces
+ * must be the same in both evaluations, or the filter could prune rows the
join itself matches.
+ * `deterministic` is not enough for that: Spark flags order-dependent
computations such as
+ * first, last, row_number or an unordered LIMIT as deterministic.
+ *
+ * The plan is walked bottom-up, tracking the output attributes whose values
are unstable: they
+ * come from a non-deterministic expression, an order-dependent aggregate or
window function, or
+ * an expression over such an attribute. The plan is rejected outright when
its row set is
+ * unstable: a filter, join condition or grouping consumes an unstable
attribute, an inner
+ * generate uses an unstable generator, a sample is unseeded or over anything
but a scan, or a
+ * limit is over anything but a total order; and, with its own reason, when an
operator's effect
+ * on the rows is not analyzed. The source qualifies when the key references
no unstable
+ * attribute. Values that are unstable but only carried to the output (a
`first(name)` next to a
+ * `GROUP BY id`, a row number next to the key) do not disqualify it.
+ */
+private[optimizer] object RuntimeFilterSourceAnalysis extends AliasHelper {
+
+ /**
+ * @param unstable output attributes whose values depend on evaluation order
or on chance.
+ * @param totallyOrdered whether the rows are in a total order on stable
keys, so that a limit
+ * over them keeps the same rows every time.
+ */
+ private case class Taint(unstable: AttributeSet, totallyOrdered: Boolean =
false)
+
+ private val NotRepeatable =
+ "the hinted side may produce different rows or join keys when evaluated
again"
+
+ /** Why `plan` is not a repeatable source of `key`, or None when it is. */
+ def rejection(plan: LogicalPlan, key: Expression): Option[String] = {
+ if (plan.isStreaming) {
+ Some("the hinted side is a stream")
+ } else if (!key.deterministic) {
+ Some(NotRepeatable)
+ } else {
+ analyze(plan) match {
+ case Left(reason) => Some(reason)
+ case Right(t) if key.references.intersect(t.unstable).nonEmpty =>
Some(NotRepeatable)
+ case _ => None
+ }
+ }
+ }
+
+ /**
+ * Whether `e` yields the same value on every evaluation. A subquery counts
as deterministic
+ * when its plan is, which is the very check this analysis replaces, so its
plan is analyzed
+ * too.
+ */
+ private def isStable(e: Expression, unstable: AttributeSet): Boolean = {
+ e.deterministic && e.references.intersect(unstable).isEmpty && !e.exists {
+ case s: SubqueryExpression => s.plan.isStreaming ||
+ analyze(s.plan).forall(t =>
s.plan.outputSet.intersect(t.unstable).nonEmpty)
+ case _ => false
+ }
+ }
+
+ /** Returns the taint of `plan`'s output, or the reason its row set is not
repeatable. */
+ private def analyze(plan: LogicalPlan): Either[String, Taint] = plan match {
+ case _: LeafNode => Right(Taint(AttributeSet.empty))
+
+ case p: Project => analyze(p.child).map { t =>
+ Taint(
+ AttributeSet(p.projectList.filterNot(isStable(_,
t.unstable)).map(_.toAttribute)),
+ t.totallyOrdered)
+ }
+
+ case f: Filter => analyze(f.child).flatMap { t =>
+ if (isStable(f.condition, t.unstable)) Right(t) else Left(NotRepeatable)
+ }
+
+ case j: Join => analyze(j.left).flatMap { l =>
+ analyze(j.right).flatMap { r =>
+ val unstable = l.unstable ++ r.unstable
+ if (j.condition.forall(isStable(_, unstable))) {
+ Right(Taint(unstable))
+ } else {
+ Left(NotRepeatable)
+ }
+ }
+ }
+
+ case a: Aggregate => analyze(a.child).map { t =>
+ // Grouping on an unstable value changes which rows form a group, so
every aggregate result
+ // then depends on it; a grouping expression's own value is as stable as
its input.
+ val stableGroups = a.groupingExpressions.forall(isStable(_, t.unstable))
+ val unstable = a.aggregateExpressions.filter { e =>
+ !isStable(e, t.unstable) ||
+ (e.exists(_.isInstanceOf[AggregateExpression]) &&
+ (!stableGroups || !isOrderIrrelevantAggregate(e)))
+ }
+ Taint(AttributeSet(unstable.map(_.toAttribute)))
+ }
+
+ case w: Window => analyze(w.child).map { t =>
+ val stablePartitions = w.partitionSpec.forall(isStable(_, t.unstable))
+ val unstable = w.windowExpressions.filter { e =>
+ !stablePartitions || !isStable(e, t.unstable) ||
!isOrderIrrelevantWindow(e)
+ }
+ Taint(t.unstable ++ AttributeSet(unstable.map(_.toAttribute)))
+ }
+
+ case u: Union =>
+ val taints = u.children.map(analyze)
+ taints.collectFirst { case Left(reason) => Left(reason) }.getOrElse {
+ val unstable = u.output.zipWithIndex.collect {
+ case (attr, i) if u.children.zip(taints).exists {
+ case (child, Right(taint)) =>
taint.unstable.contains(child.output(i))
+ case _ => false
+ } => attr
+ }
+ Right(Taint(AttributeSet(unstable)))
+ }
+
+ // An inner generate drops the rows for which the generator yields
nothing, so an unstable
+ // generator changes the row set; an outer generate keeps them.
+ case g: Generate => analyze(g.child).flatMap { t =>
+ if (isStable(g.generator, t.unstable)) {
+ Right(t)
+ } else if (g.outer) {
Review Comment:
The outer-generate case is fixed, thanks. The risk this thread raised still
applies at Aggregate, though: when a grouping expression is unstable, the case
returns `Taint(...)` with `unstableMultiplicity = false`, so a `count(*)` over
its stable grouping columns is accepted as a stable key. For example, with
`rnd3` a nondeterministic UDF, `SELECT a, count(*) AS c FROM (SELECT a FROM
(SELECT a, rnd3() AS r FROM dim) GROUP BY a, r) GROUP BY a`. Setting
`unstableMultiplicity = !stableGroups` there, and the matching change in the
Distinct case, would close it.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3995080338","thread_id":"inline:3995080338","verdict_sha256":"41f2e83c8f4fba6960398fb3dff5399ed0cc1480c5477d4d73b8586c2382b847"}
-->
##########
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:
Thanks for trimming the section. The remaining sentence, 'The hinted side
must produce the same rows and join keys whenever it is evaluated', still reads
as full-row equality, while the analysis accepts unstable values carried in
non-key columns (for example `first(name)` next to a `GROUP BY id` key). Saying
that the set of rows and their join-key values must be the same on every
evaluation, while other columns may differ, would match the implementation.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3995080346","thread_id":"inline:3995080346","verdict_sha256":"41f2e83c8f4fba6960398fb3dff5399ed0cc1480c5477d4d73b8586c2382b847"}
-->
--
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]
