dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4448448541
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || (isAttributeAtLeaves(key, side) &&
+ creationPlan(other).deterministic && !goesIntoReused(key,
side))
Review Comment:
**1. [Medium] `deterministic` treats a non-inlined, non-deterministic CTE as
deterministic**
This check, which answers [round-1 item
1](https://github.com/apache/spark/pull/59916#discussion_r4437009548), passes a
creation side that reads a CTE through a `CTERelationRef`. The ref is a leaf
whose only expressions are its output attributes (`cteOperators.scala`
L216-L224), so `QueryPlan.deterministic` (`QueryPlan.scala` L150-L151) is true
even when the CTE is not. `InlineCTE` keeps such a CTE when it has more than
one reference or is `MATERIALIZED` (`InlineCTE.scala` L90 and L96-L97), and
`ReplaceCTERefWithRepartition` runs only after this rule
(`SparkOptimizer.scala` L121).
With exchange reuse off, nothing shares the CTE's repartition between the
join and the bloom filter's subquery (`ReuseExchangeAndSubquery.scala` L45, and
the AQE stage cache at `AdaptiveSparkPlanExec.scala` L899 and L923), so the
subquery draws a second sample. With `spark.sql.exchange.reuse=false`, `fact`
partitioned by `k` and over the application-side threshold, and every `dim.k`
in `fact`, `lost` must be 0:
```sql
WITH d AS (
SELECT k, x FROM dim WHERE cast(reflect('java.lang.Math', 'random') AS
DOUBLE) < 0.5)
SELECT count(*) total, count_if(t.k IS NULL) lost
FROM d LEFT JOIN (SELECT k, sum(v) s FROM fact GROUP BY k) t ON d.k = t.k
WHERE d.x = 1
UNION ALL SELECT count(*), 0 FROM d
```
With reuse off, `PartitionPruning` still adds the `onlyInBroadcast` DPP
filter on `t.k` when it has a benefit, `reused` is empty, and `t.k` gets a
bloom filter built from `Filter(x = 1, CTERelationRef)`. Rows of `t` whose key
is in the join's sample but not in the subquery's sample are pruned, so `lost >
0`. On master the DPP filter keeps the bloom filter off and `lost` is 0. With
exchange reuse on, the default, the CTE shuffle is shared and the result is
right. I traced this from the code and did not run it.
Keys without a DPP filter have the same blind spot on master (the
SPARK-60808 class). Treating a `CTERelationRef` as non-deterministic when its
`CTERelationDef` is not, for example by collecting the ids of non-deterministic
definitions from the `WithCTE` at the top of the plan, would close it for these
keys.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
Review Comment:
**2. [Medium] A union branch where the lineage stops passes the attribute
check**
`findExpressionAndTrackAllLineagesDown` returns `Nil` for a union branch
where it cannot follow the key, since at a `Union` it matches only an output
attribute (`predicates.scala` L143-L150). `forall` is then true for that
branch, but pushdown still copies the bloom filter into it with the aliases
expanded. So the check that answers [round-1 item
3](https://github.com/apache/spark/pull/59916#discussion_r4437009558) can still
let a failable key become a V1 partition filter.
With ANSI on (the default), the suite's shuffle-join thresholds, and `sp(v
INT) PARTITIONED BY (p STRING)` with a partition `p = 'abc'` whose rows all
have `v <= 0`:
```sql
SELECT * FROM (
SELECT f5 AS k, a5 AS v FROM bf5part
UNION ALL
SELECT coalesce(cast(x AS INT), 0) AS k, sum(v) AS v
FROM (SELECT p AS x, v FROM sp WHERE v > 0 UNION ALL SELECT p, v FROM sp
WHERE v > 0)
GROUP BY x) u
JOIN bf2 ON u.k = bf2.c2 WHERE bf2.a2 = 36
```
- `PartitionPruning` takes the first lineage, `bf5part.f5`
(`PartitionPruning.scala` L61), and adds an `onlyInBroadcast` DPP filter on
`u.k`, which this rule counts as dropped.
- In the second branch, the aggregate's alias turns the key into
`coalesce(cast(x AS INT), 0)`, the inner `Union` returns `Nil`, and L487 checks
only `(f5, bf5part)`.
- `PushPredicateThroughNonJoin` pushes the bloom filter below the aggregate
with the alias expanded, and then into both inner branches. `Cast` is not
`throwable`, so the throwable-ordering checks on master do not stop it.
- `FileSourceStrategy` makes it a partition filter
(`FileSourceStrategy.scala` L212-L213), and `dynamicallySelectedPartitions`
evaluates it on every partition (`DataSourceScanExec.scala` L423-L432), so
`cast('abc' AS INT)` fails with `CAST_INVALID_INPUT`.
On master the DPP filter keeps the bloom filter off this key, its pushed
copy is planned as `true`, and the query succeeds. `coalesce` keeps an inferred
`isnotnull` from evaluating the cast first, as the comment of the new
not-an-attribute test says. A `GROUP BY` on the expression, or a `Project`
directly over the inner union, is safe, because `PullOutGroupingExpressions`
and `PushProjectionThroughUnion` turn it into aliases in each branch. I traced
this from the code and did not run it.
The Known limit about a branch without a lineage gives a literal column as
its example, which is harmless, and does not say that the query can fail.
Treating a branch without a lineage as a failure here, when the key has
references, would close this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
Review Comment:
**4. [Low] The attribute check also turns away keys that cannot fail**
This check stands in for "evaluating the key as a partition filter cannot
throw", but it also rejects keys that cannot throw, and those keys go back to
having no runtime filter, as on master. The most common case is a widening cast:
```sql
-- fact(v) PARTITIONED BY (p INT), dim(k BIGINT, x)
SELECT * FROM fact f JOIN dim d ON f.p = d.k WHERE d.x = 1
```
Type coercion makes the key `cast(f.p AS BIGINT)` in both ANSI and non-ANSI
mode (`TypeCoercionHelper.scala` L737-L744). `PartitionPruning` needs only its
references to be partition columns (`PartitionPruning.scala` L68), so the DPP
filter is added and planned as `true` in a shuffle join, and L487 rejects the
key. Round 1 gave it the bloom filter. `FLOAT` and `DOUBLE` keys are rejected
the same way, because `NormalizeFloatingNumbers` wraps both join keys in
`KnownFloatingPointNormalized(NormalizeNaNAndZero(_))` before
`PartitionPruning` runs (`Optimizer.scala` L303). The application side opposite
the DPP filter, whose leaf need not be partitioned at all, is rejected too.
Accepting an attribute under an integral or decimal widening cast and under
the normalization wrapper, or requiring an attribute only at V1 and Hive
partition-column leaves, would keep these keys. `DATE` to `TIMESTAMP` can
overflow, so it would have to stay out. At least the Known limits could mention
this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
Review Comment:
**3. [Low] A failable creation-side expression runs over partitions that the
main plan prunes**
`creationPlan` keeps the DPP filters on the other keys, and this check looks
only at the application side. The bloom filter's subquery is planned on its own
(`PlanSubqueries` calls `QueryExecution.prepareExecutedPlan`, and AQE uses
`InsertAdaptiveSparkPlan.compileSubquery`), so an `onlyInBroadcast` DPP filter
in it finds no broadcast and becomes `true` (`PlanDynamicPruningFilters.scala`
L123-L124), while the main plan keeps the same filter. The creation side then
reads partitions that the join never reads, and a failable key or filter there
runs on rows that master never evaluates.
With AQE off, `spark.sql.autoBroadcastJoinThreshold=-1`, ANSI on, `A`
partitioned by `a` and `C` partitioned by `q`:
```sql
SELECT f.v FROM A f JOIN C c ON f.a = CAST(c.s AS INT) AND f.b = c.q
WHERE f.x = 1 AND c.flag = 1
UNION ALL
SELECT /*+ BROADCAST(g) */ g.v FROM X JOIN (SELECT * FROM A WHERE x = 1) g
ON X.k1 = g.a AND X.k2 = g.b
```
- `PartitionPruning` adds `dpp(f.a)` with build plan `C` and `dpp(c.q)` with
build plan `A` to the same join (`PartitionPruning.scala` L350-L359).
- `dpp(c.q)` reuses the broadcast of `g` in the other branch, so it is kept,
and `dpp(f.a)` is dropped. `g` is another instance of `A`, so `goesIntoReused`
is false.
- The bloom filter on `f.a` reads `Filter(dpp(c.q), Filter(flag = 1, C))`,
where that DPP filter is `true`.
If a row `('abc', flag = 1)` sits in a `q` partition that the main plan
prunes, this PR fails with `CAST_INVALID_INPUT` and master succeeds. A failable
creation-side filter such as `CAST(c.t AS INT) = 1` fails the same way. An
implicit `INT = STRING` key does not reach this, because ANSI casts both sides
to `BIGINT` and L487 then rejects the application key. With AQE on, `g` must
join on `(a, b)` in that order, because AQE compares the broadcast mode. I
traced this from the code and did not run it.
The shape is narrow, and keys without a DPP filter already have it on
master. Stripping the other keys' DPP filters, as discussed in [round-1 item
20](https://github.com/apache/spark/pull/59916#discussion_r4437009601), would
not help, since the subquery would still read the same partitions. Skipping the
bloom filter for these keys when the creation side still has a DPP filter that
`isKept` counts as kept would keep master's outcome.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || (isAttributeAtLeaves(key, side) &&
+ creationPlan(other).deterministic && !goesIntoReused(key,
side))
Review Comment:
**5. [Low] The determinism check looks at the whole join side, not at the
bloom filter's subquery**
`creationPlan(other).deterministic` covers every column and node of the join
side, so it also blocks bloom filters whose subquery is deterministic:
```sql
SELECT * FROM bf5part p
JOIN (SELECT c2, input_file_name() src FROM bf2 WHERE a2 = 36) d ON p.f5 =
d.c2
```
Under the suite's shuffle-join thresholds, the DPP filter on `p.f5` is
planned as `true` and this check is false, so `p.f5` gets no runtime filter.
But `ColumnPruning` at L83 removes `src`, and the subquery
`Aggregate(bloom(xxhash64(c2)), Project(c2, Filter(a2 = 36, bf2)))` is
deterministic. `rand()`, `uuid()` and `monotonically_increasing_id()` behave
the same, and so does a non-deterministic filter that stays above an inner
join, where `extract` starts again below the join.
The hazard of SPARK-60808 is where every bloom filter's subquery is built.
Checking `aggregate.deterministic` next to the `PYTHON_UDF` check at L86-L88
would decide both shapes exactly and replace this check. Applied only to keys
with a dropped DPP filter, it keeps master's plans. Applied to every key, it
would also fix SPARK-60808, and the expectation at
`InjectRuntimeFilterSuite.scala` L1361 would flip. Neither covers the
`CTERelationRef` case in item 1.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1075,361 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter has a
non-deterministic build " +
+ "plan") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assert(getNumBloomFilters(sql(query).queryExecution.optimizedPlan) ==
0)
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ // Without DPP, the key has no DPP filter, and the bloom filter
comes as on master.
+ assert(getNumBloomFilters(sql(query).queryExecution.optimizedPlan)
== 1)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the key of the dropped dpp
filter is not an " +
Review Comment:
**8. [Low] No test pins the all-lineages walk of the attribute check**
Replacing the walk at `InjectRuntimeFilter.scala` L487 with
`findExpressionAndTrackLineageDown` (first lineage only) changes no assertion.
Both queries here have one lineage, and the union test at L1173 has attribute
keys in both branches. With that change, a bloom filter on a union whose first
branch is an attribute and whose second is an expression would get through. A
union variant here would pin it:
```sql
SELECT * FROM (SELECT f5 k, a5 v FROM bf5part
UNION ALL SELECT coalesce(f5 + 1, 0) k, a5 v FROM bf5filtered) u
JOIN bf2 ON u.k = bf2.c2 WHERE bf2.a2 = 36
```
This PR adds no bloom filter there, and the first-lineage variant does. The
query succeeds either way, so only the plan assertion catches it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
Review Comment:
**10. [Low] No test needs the conjunct collection in `reused`**
Reverting L467-L471 to round 1's `plan.collect { case Filter(dpp:
DynamicPruningSubquery, _) => dpp }` should change no assertion. In the
IN-subquery test (`InjectRuntimeFilterSuite.scala` L1385), the run over the
whole plan gets a fresh standalone DPP filter on `p.f5` from
`PartitionPruning`, whose build plan already matches `x`, and `CombineFilters`
later drops the duplicate, since `DynamicPruning.scala` L112-L117 zeroes the
`exprId`. The collection cannot change `isKept`, only `goesIntoReused`, and
only when the fresh copy's build plan differs from the merged one, for example
when the filtering side has a DPP filter of its own.
That test also has no positive control. Without `broadcast(x)`, the DPP
filter on `p.f5` is really dropped, and I would expect the run over the whole
plan to give `p.f5` its bloom filter.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
Review Comment:
**6. [Low] In an IN subquery, the key's original DPP filter stays in the
creation side**
In an uncorrelated IN subquery, the subquery's own run adds the DPP filter,
skips the key here (`inSubquery`), and `CombineFilters` then merges the DPP
filter into the scan's filter as a conjunct (`Optimizer.scala` L2409-L2415). In
the run over the whole plan, `PartitionPruning` adds a fresh copy on top,
because its skip matches only a standalone `Filter(DynamicPruningSubquery, _)`
(`PartitionPruning.scala` L318-L319). `creationPlan` strips only that copy, so
the original stays in the bloom filter's subquery.
Without a UDF or a `LIKE`, that only costs planning the DPP filter's build
plan inside the subquery, where it becomes `true`. With one,
`isSimpleExpression` sees the subquery's pattern bits through
`SubqueryExpression.treePatternBits` and rejects the creation side at L133:
```sql
SELECT * FROM bf3 WHERE bf3.a3 IN (
SELECT b.b5 FROM bf5filtered q JOIN bigapp b ON q.f5 = b.k
WHERE q.a5 > 50 AND CAST(b.k AS STRING) LIKE '%1%0%')
```
With the thresholds of the creation-side test (application side 20000) and
`autoBroadcastJoinThreshold=-1`, this key gets no runtime filter, while the
same join at the top level gets the bloom filter. The outcome is never worse
than master. I traced this from the code and did not run it. Stripping DPP
conjuncts on the key's pruning key as well, or finding `keyDpps` among
conjuncts, would cover it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1075,361 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
Review Comment:
**9. [Low] Five negative tests have no control showing that the bloom filter
would otherwise be added**
This assertion, and the ones at L1187, L1230, L1336 and L1379, check only
that no bloom filter appears. If a size drifted across the 3000 and 2000
thresholds, or creation-side extraction changed, they would stay green while
`isKept`, the union lineage, the determinism check and the attribute check went
unexercised. L1124, L1164, L1293 and L1359 already have a control with
`DYNAMIC_PARTITION_PRUNING_ENABLED` off, and the same control here would show
that the guard is what keeps the bloom filter out.
`withDppTestConf` also leaves `reuseBroadcastOnly` at its default, and every
one of these tests needs it to be true, for the same reason as the projection
default in [round-1 item
17](https://github.com/apache/spark/pull/59916#discussion_r4437009614).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,21 +377,51 @@ 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)
- case _ => false
- }
+ // The plans that a broadcast hash join of `plan` could broadcast: the side
that
+ // `getBroadcastHashJoinBuildSide` picks. For an equi-join, that is the side
with a broadcast
+ // hint, or else a side small enough to broadcast that the join type can
build from, unless a
+ // `SHUFFLE_HASH` hint asks for a shuffle hash join. For a null-aware anti
join, it is the right
+ // side under the null-aware anti join thresholds. It ignores the other join
hints, so the list
+ // can hold more plans than physical planning broadcasts. It holds no fewer,
unless the
+ // statistics change after this rule. It takes each side without the DPP
filters on top of it,
+ // since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
Review Comment:
**14. [Low] The optimizer now logs a hash join warning for joins on keys
without binary equality**
`getBroadcastHashJoinBuildSide` evaluates `hashJoinSupported` eagerly, as
the first operand of `noShufflePlannedBefore` (`joins.scala` L435-L436), and
`hashJoinSupported` logs a warning for keys that are not binary-stable
(`joins.scala` L402-L415). Whenever `reused` is computed, which happens for any
main plan with an `onlyInBroadcast` DPP filter, `broadcastablePlans` visits
every join. So each join on such keys, for example a `UTF8_LCASE` key in a
comma join or in a semi join from an IN subquery, logs "Hash based joins are
not supported ..." once more, on top of the warning from `JoinSelection`. Round
1's `getBroadcastBuildSide` did not log. This comes from my own suggestion in
[round-1 item
23](https://github.com/apache/spark/pull/59916#discussion_r4437009644).
Swapping the two operands of `noShufflePlannedBefore` in the shared helper
would run `hashJoinSupported` only under a `SHUFFLE_HASH` hint. No plan changes.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
Review Comment:
**15. [Nit] One lineage walk could serve both checks**
`goesIntoReused` and `isAttributeAtLeaves` are each used once, in the same
expression at L511-L512 and with the same `(key, side)`, so a key with a
dropped DPP filter walks its lineage twice per side, and L309 walks it a third
time. One walk does the same job:
```scala
!hasDroppedDpp || (key.isInstanceOf[Attribute] &&
creationPlan(other).deterministic &&
findExpressionAndTrackAllLineagesDown(key, side).forall { case (leafKey,
leaf) =>
leafKey.isInstanceOf[Attribute] &&
!reused.exists(_.exists(_.fastEquals(leaf)))
})
```
Every conjunct is pure, and an empty lineage gives the same result. The
saving is small, so this is mostly about keeping the pushdown rule in one
place, which is also where the fix for item 2 would go. Computing the lineages
once in `canTakeBloomFilter` and passing them to both helpers would keep the
names that the overview uses.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,21 +377,51 @@ 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)
- case _ => false
- }
+ // The plans that a broadcast hash join of `plan` could broadcast: the side
that
+ // `getBroadcastHashJoinBuildSide` picks. For an equi-join, that is the side
with a broadcast
+ // hint, or else a side small enough to broadcast that the join type can
build from, unless a
+ // `SHUFFLE_HASH` hint asks for a shuffle hash join. For a null-aware anti
join, it is the right
+ // side under the null-aware anti join thresholds. It ignores the other join
hints, so the list
+ // can hold more plans than physical planning broadcasts. It holds no fewer,
unless the
+ // statistics change after this rule. It takes each side without the DPP
filters on top of it,
+ // since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // The plans whose broadcast a DPP filter may reuse: its build plan, and for
an `onlyInBroadcast`
+ // filter, the source plan of its broadcast value projection.
+ private def reusablePlans(dpp: DynamicPruningSubquery): Seq[LogicalPlan] =
+ dpp.buildQuery +: dpp.usableBroadcastValueProjection.filter(_ =>
dpp.onlyInBroadcast)
Review Comment:
**7. [Nit] A projected DPP filter counts as kept even where physical
planning cannot use it**
For an `onlyInBroadcast` DPP filter with a broadcast value projection,
`isKept` is true when any broadcastable side has the same result as the
projection's source plan. Without AQE, the planner also requires a broadcast
hash join that is not null-aware and has the same broadcast mode
(`PlanDynamicPruningFilters.scala` L70-L73), while `reused` includes the right
side of a null-aware anti join and ignores the keys. A planned projection can
also fail open at run time past `broadcastProjection.maxRows`, `maxBytes` or
`maxSourceBytes`, where `InSubqueryExec.eval` returns true. In both cases the
key gets neither pruning nor a bloom filter, as on master. The feature is
internal and off by default, so a line in the Known limits would be enough.
Today they mention only the AQE mode check and SPARK-60703.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +303,46 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
checkWithAndWithoutFeatureEnabled(query, shouldReplace = false)
}
+ // Runs `f` with the thresholds of the DPP tests and `confs`, with AQE on
and off unless `confs`
+ // sets it.
+ def withDppTestConf(confs: (String, String)*)(f: => Unit): Unit = {
+ val adaptiveModes = confs.collectFirst {
+ case (SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, adaptive) => Seq(adaptive)
+ }.getOrElse(Seq("true", "false"))
+ adaptiveModes.foreach { adaptive =>
+ withSQLConf(Seq(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000",
Review Comment:
**13. [Low] The temporary check had nothing to find in most of the suites
the description lists**
The description says that the check found nothing over the DPP suites (V1,
V2 and V2 filter, AQE on and off) and the TPC-DS and TPC-H plan stability
suites. Like this helper, a suite has to lower the application-side threshold,
whose default is 10GB (`SQLConf.scala` L989-L996), before any bloom filter can
appear. The DPP suites neither lower it nor inject size statistics, and their
tables have at most about a thousand rows, so `satisfyByteSizeRequirement`
rejects every bloom filter there on master, on this PR and in the control. The
same holds for the TPC-H suite and the TPC-DS suites without statistics. So
only the sf100 suites with statistics and the suites that lower the threshold
can show the case, which matches 54 of the 62 control hits being in sf100.
Could the description say so, or add a run of the DPP suites with a lowered
threshold?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
Review Comment:
**11. [Low] This comment does not hold for EXISTS, and scalar subqueries
lose round 1's decision**
An uncorrelated EXISTS becomes `IsNotNull(ScalarSubquery(...))` through
`RewriteNonCorrelatedExists` in the Finish Analysis batch
(`finishAnalysis.scala` L69-L79), before `OptimizeSubqueries` runs
(`Optimizer.scala` L222-L223), so it is never joined with the rest of the
query. A correlated subquery never reaches `inSubquery`, since `apply` returns
it unchanged (L548). So the run over the whole plan decides only uncorrelated
IN subqueries, and it can do so because `PartitionPruning` adds a fresh
standalone DPP filter there, while the subquery's own copy is a conjunct of the
scan filter by then. Neither the comment nor the description says that, and the
Known limits name only scalar subqueries.
For a scalar subquery, including a former EXISTS, the subquery's own run
already sees exactly the broadcasts its DPP filters can reuse, because
`PlanSubqueries` and `compileSubquery` plan it on its own and the DPP planners
look only in that plan. Round 1 decided those keys correctly, and now they keep
master's outcome. With the broadcast threshold at 500, as in the IN-subquery
test, `p.f5` here gets no runtime filter: its DPP filter is `true`, and it gets
no bloom filter.
```sql
SELECT * FROM bf3 WHERE EXISTS (
SELECT 1 FROM bf5part p JOIN bf2 d ON p.f5 = d.c2 WHERE d.a2 = 36)
```
Wording that matches the code could be: "So does every DPP filter in an
uncorrelated subquery. An IN subquery is joined with the rest of the query
later, where `PartitionPruning` puts a new DPP filter on the key and this rule
decides it against every broadcast. A scalar subquery, which includes an
uncorrelated EXISTS, is planned on its own and keeps master's behavior."
Limiting `inSubquery` to IN subqueries would keep round 1's result for the
others, but `Subquery` does not record the kind.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,21 +377,51 @@ 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)
- case _ => false
- }
+ // The plans that a broadcast hash join of `plan` could broadcast: the side
that
Review Comment:
**12. [Nit] The header of `broadcastablePlans` restates the callee and has
drifted**
Since this method now calls `getBroadcastHashJoinBuildSide` as is, the
header restates its rules, and three details differ from the code:
- The hint path is also limited by the join type (`joins.scala` L310-L315),
so a broadcast hint on the left side of a left outer join is ignored. The
comment ties "that the join type can build from" to the size path only.
- The `SHUFFLE_HASH` veto applies only to keys that a hash join supports
(`joins.scala` L435-L436).
- Under AQE, L392 returns `Nil` for a null-aware anti join, although
`JoinSelection` still broadcasts its right side. So "For a null-aware anti
join, it is the right side" and "It holds no fewer" hold only as "no fewer
plans whose broadcast a DPP filter can reuse".
Pointing at the callee's doc (`joins.scala` L418-L427) and stating only what
this method adds, the AQE case and the stripped DPP filters, would keep it from
drifting again.
--
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]
