dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4470999359
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
+ case And(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Or(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Not(child) => isSafeCondition(child)
+ case IsNull(child) => isSafeValue(child)
+ case IsNotNull(child) => isSafeValue(child)
+ case c: BinaryComparison => isSafeValue(c.left) && isSafeValue(c.right)
+ case In(value, list) => isSafeValue(value) && list.forall(isSafeValue)
+ case InSet(value, _) => isSafeValue(value)
+ case dpp: DynamicPruningSubquery => isSafeKey(dpp.pruningKey)
+ case _ => isSafeValue(condition)
+ }
+
+ private def isSafeValue(e: Expression): Boolean = e.isInstanceOf[Literal] ||
isSafeKey(e)
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must be deterministic (SPARK-60808) and read no CTE, whose
reference hides the
+ // determinism of its plan. A part of the join side outside that plan can
only drop rows from the
+ // join, so the bloom filter still holds every key that the join sees.
Nothing in the plan may
+ // fail either: the filters, and the projections that the key and the
filters need.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ evaluated.deterministic && !evaluated.containsPattern(CTE) &&
!evaluated.exists {
+ case Project(projectList, _) => !projectList.forall(e =>
isSafeKey(trimAliases(e)))
+ case Filter(condition, _) => !isSafeCondition(condition)
+ case _: LeafNode => false
Review Comment:
**3. [Low] The subquery opens files of partitions that only the join skips**
`canEvaluateAgain` makes sure that no expression in the subquery can fail,
and this line treats the read of the leaf as safe. But the subquery still reads
the partitions that a kept DPP filter on another key prunes only in the join.
The reply to [round-3 item
1](https://github.com/apache/spark/pull/59916#discussion_r4462890666) says that
this "only leaves the subquery more work", which holds for expressions but not
for reading files.
Take `castInFilter` at `InjectRuntimeFilterSuite.scala` L1551 without
`cast(c.s as int) > 0`, with the same tables and confs, AQE on or off:
```sql
SELECT /*+ broadcast(d) */ f.v FROM ra f JOIN rc c ON f.a = c.t
JOIN rd d ON c.q = d.k WHERE f.x = 1 AND c.flag = 1 AND d.dflag = 1
```
- `dpp(f.a)` is dropped and `dpp(c.q)` is kept, so `f.a` gets a bloom filter
whose subquery reads all of `rc`. In `castInFilter`, only the cast moves the
bloom filter to `c.t`.
- If the file under `rc/q=99` is corrupt (with the default
`spark.sql.files.ignoreCorruptFiles=false`), or was written with another type
for `t` or `flag`, the subquery fails with `FAILED_READ_FILE`, and so does the
query.
- On master, `f.a` gets no bloom filter, `dpp(c.q)` skips `q = 99` before
any file is opened, and the query returns 4000 rows.
Round 2's union query at `InjectRuntimeFilterSuite.scala` L1563 with a safe
key (`f.a = c.t`, AQE off) fails the same way, and that is the shape that the
removed round-3 check rejected. Keys without a DPP filter read the same way on
master (SPARK-60842), so a line in the Known limits may be enough. Otherwise,
the lineage-based search for kept DPP filters from round-3 item 1 would cover
both shapes. I traced this from the code and did not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
Review Comment:
**2. [Low] A collated string key can get a bloom filter that hashes raw
bytes**
`isSafeKey` accepts any `Attribute`, including a string with a non-binary
collation. The bloom filter hashes both sides with `XxHash64` (L77 and L92),
and since SPARK-52828 it hashes such a string as raw bytes (`hash.scala` L934).
`RewriteCollationJoin` wraps equi-join keys in `CollationKey`, but only in
"Finish Analysis", so the semi join that `RewritePredicateSubquery` builds
later keeps the raw keys.
With broadcasts off, the application-side threshold at 0, and `cf` in a V2
catalog that offers `k` for runtime filtering, for example `testcat` with
`PARTITIONED BY (k)`:
```sql
-- cf(k STRING COLLATE UTF8_LCASE, v INT): ('abc', 1)
-- cd(k STRING COLLATE UTF8_LCASE, flag INT): ('ABC', 1), ('zzz', 0)
SELECT v FROM cf WHERE k IN (SELECT k FROM cd WHERE flag = 1)
```
The DPP filter on `cf.k` is dropped, and this PR adds
`might_contain(bf(xxhash64(cd.k)), xxhash64(cf.k))`. `'abc'` and `'ABC'` hash
differently, so the row is dropped and the query returns nothing. Master skips
this key because of its DPP filter and returns 1.
Master already adds the same bloom filter for such a key without a DPP
filter, since v4.0.1, so the root fix belongs in `injectBloomFilter`, and I'll
file a separate JIRA for it. Here, rejecting a key whose type is not
`UnsafeRowUtils.isBinaryStable`, as L313 does for the materialized path, would
keep this PR from widening it. I traced this from the code and did not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
+ case And(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Or(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Not(child) => isSafeCondition(child)
+ case IsNull(child) => isSafeValue(child)
+ case IsNotNull(child) => isSafeValue(child)
+ case c: BinaryComparison => isSafeValue(c.left) && isSafeValue(c.right)
+ case In(value, list) => isSafeValue(value) && list.forall(isSafeValue)
+ case InSet(value, _) => isSafeValue(value)
+ case dpp: DynamicPruningSubquery => isSafeKey(dpp.pruningKey)
Review Comment:
**4. [Low] A DPP filter that cleanup drops from the join still runs in the
bloom filter's subquery**
This arm accepts a DPP filter on another key because "its own subquery, if
it has one, runs in the join as well" (L424-L425). That does not hold when
`CleanupDynamicPruningFilters` drops the filter from the main plan only. It
does so when the filter's key also has an equality
(`CleanupDynamicPruningFilters.scala` L57-L60), or when pushdown cannot move
the filter through a Project with a non-deterministic column (`Optimizer.scala`
L2623-L2624, then `CleanupDynamicPruningFilters.scala` L86-L89). The bloom
filter's subquery is fixed at L84 and never cleaned, so it plans the filter as
its own `SubqueryExec` over the whole build plan
(`PlanDynamicPruningFilters.scala` L125-L131).
Take `reuseBroadcastOnly=false`, ANSI on, broadcasts off and the
application-side threshold at 1. `fa` is partitioned by `a`, with its `'abc'`
row only in `a = 99`, and `fc` is partitioned by `q`:
```sql
SELECT /*+ BROADCAST(d) */ f.v FROM fa f JOIN fc c ON f.a = c.t AND f.b = c.q
JOIN fd d ON f.a = d.k
WHERE f.x = 1 AND CAST(f.s AS INT) > 0 AND c.flag = 1 AND c.q = 10 AND
d.dflag = 1
```
- `dpp(c.q)` has a benefit, and its build plan is the `fa` side. Cleanup
turns it into `true` in the main plan because of `c.q = 10`, and the upper
join's DPP filter prunes `fa` to `a IN (1, 2)`.
- `dpp(f.a)` is dropped, so `f.a` gets a bloom filter, and `dpp(c.q)` passes
this arm. In the subquery, its own subquery reads all of `fa` and fails with
`CAST_INVALID_INPUT` on `a = 99`. Master returns the rows.
- Replacing `c.q = 10` with a `rand()` or `input_file_name()` column of `fc`
that the query outputs does the same. That variant is new in becfca3, since
`ColumnPruning` now removes the column before the determinism check.
When nothing fails, the subquery still runs that build plan once more.
Accepting only an `onlyInBroadcast` DPP filter here, which is always `true` in
the subquery (`PlanDynamicPruningFilters.scala` L123-L124), would close this.
The test at `InjectRuntimeFilterSuite.scala` L1593 and its bullet in the
description assume the opposite and would change with it. I traced this from
the code and did not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
+ case And(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Or(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Not(child) => isSafeCondition(child)
+ case IsNull(child) => isSafeValue(child)
+ case IsNotNull(child) => isSafeValue(child)
+ case c: BinaryComparison => isSafeValue(c.left) && isSafeValue(c.right)
+ case In(value, list) => isSafeValue(value) && list.forall(isSafeValue)
+ case InSet(value, _) => isSafeValue(value)
+ case dpp: DynamicPruningSubquery => isSafeKey(dpp.pruningKey)
+ case _ => isSafeValue(condition)
+ }
+
+ private def isSafeValue(e: Expression): Boolean = e.isInstanceOf[Literal] ||
isSafeKey(e)
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must be deterministic (SPARK-60808) and read no CTE, whose
reference hides the
+ // determinism of its plan. A part of the join side outside that plan can
only drop rows from the
+ // join, so the bloom filter still holds every key that the join sees.
Nothing in the plan may
+ // fail either: the filters, and the projections that the key and the
filters need.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ evaluated.deterministic && !evaluated.containsPattern(CTE) &&
!evaluated.exists {
Review Comment:
**1. [Low] A cache of a non-repeatable plan passes the determinism check**
A leaf has no expressions, so `evaluated.deterministic` is true for an
`InMemoryRelation`, whatever plan it caches. With the suite's DPP confs:
```scala
spark.table("bf2").where("cast(reflect('java.lang.Math', 'random') as
double) < 0.5")
.cache().createOrReplaceTempView("bf2r")
sql("select * from bf5part p join bf2r d on p.f5 = d.c2 where d.a2 = 36")
```
- The DPP filter on `p.f5` is dropped. `extract` reaches the cache through
L229 (`case None if hasHitSelectiveFilter`), and this check accepts `Filter(a2
= 36, InMemoryRelation)`, so `p.f5` gets a bloom filter. On master, the DPP
filter keeps it off.
- The bloom filter's subquery and the join scan the cache separately. If a
partition is computed twice, they see different samples, and the join loses the
rows whose keys only its own sample has. That can happen under AQE, where each
reference to a cache gets its own build job with "only a per-executor write
lock" (`InMemoryRelation.scala` L274-L277), with `MEMORY_ONLY` blocks that are
not stored or are evicted, and with a lost executor.
The reply to [round-3 item
7](https://github.com/apache/spark/pull/59916#discussion_r4462890737) says that
the rule cannot see `LogicalRDD`. A cache is different: `MaterializedLeafNode`
is in catalyst, and its doc says that this rule uses it to tell "whether its
output can be scanned again safely" (`LogicalPlan.scala` L256-L259). `case
leaf: MaterializedLeafNode => !leaf.isOutputRepeatable` before L452 would close
this. It would also turn away a deterministic cache that is not materialized
yet, which then gets master's outcome, as a deterministic `MATERIALIZED` CTE
does. At least the Known limits could name caches next to the checkpoint. I
traced this from the code and did not run it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ 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() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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 " +
Review Comment:
**5. [Low] No test pins the determinism check of `canEvaluateAgain`**
Since becfca3, `isSafeCondition` rejects this test's filter, and the one in
the test at L1360, on its own: `reflect` returns a string, so `cast(... as
double)` is not a safe key (`InjectRuntimeFilter.scala` L415). Every expression
that `isSafeKey` and `isSafeCondition` accept is deterministic, except a DPP
filter, whose `deterministic` includes its build plan. So removing
`evaluated.deterministic &&` at `InjectRuntimeFilter.scala` L449 fails no test.
The check still matters for one shape: another key's DPP filter with a
benefit and a non-deterministic build plan in the creation side. Cleanup turns
it into `true` in the main plan, while the bloom filter's subquery would run it
on a second sample and drop keys that the join sees. A query in the test at
L1593, with its tables and confs, would pin it:
```scala
val query = "select f.v from (select * from ra2 where x = 1 and rand() <
2.0) f " +
"join rc2 c on f.a = c.t and f.b = c.q where c.flag = 1"
assert(bloomFilterKeys(query).isEmpty)
```
Also, this test's name and its bullet in the description ("no bloom filter
for a key whose DPP filter has a non-deterministic build plan") no longer match
the rule. The test at L1491 has such a DPP filter, through `input_file_name()`
in `d`, and gets the bloom filter. Something like "... if the creation side
that the subquery reads is not deterministic" would match.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ 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. An uncorrelated 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
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). 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).
+ // 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)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
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 no 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)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
+ key: Expression,
+ side: LogicalPlan,
+ creation: FilterCreationSide) =
+ !hasDroppedDpp || (canPushDown(key, side) &&
canEvaluateAgain(creation))
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. The key has no dropped DPP filter, or step 3 above allows
the bloom filter
val hasShuffle = isProbablyShuffleJoin(left, right, hint)
if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
!hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
- }
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .filter(canTakeBloomFilter(l, left, _))
Review Comment:
**9. [Nit] A safe transitive creation side is not tried**
This check runs on the one creation side that the search returns, so when
that side is unsafe, the search does not fall back to the other side of an
inner join. `extract` follows the transitive key at L159 only when the first
side gives no result (L151-L153). With `withDppTestConf()`:
```sql
select * from bf5part p join (select /*+ broadcast(bf3) */ bf2.c2 from bf2
join bf3 on bf2.c2 = bf3.c3 where bf2.a2 + 1 = 37 and bf3.a3 = 36) d on
p.f5 = d.c2
```
The search returns `Filter(a2 + 1 = 37, bf2)`, which `canEvaluateAgain`
rejects for the `Add`, and it never tries `Filter(a3 = 36, bf3)`, which would
pass. So `p.f5` gets no bloom filter. That is master's outcome and within the
Known limit on unsafe creation sides, hence a Nit. Checking the creation side
inside the search, as `safeLineage` does at L194-L197, would find it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
Review Comment:
**8. [Low] `ExprUtils.canEvaluateUnconditionally` already answers this
question**
`isSafeKey`, `isSafeCondition` and `isSafeValue` decide whether an
expression can be evaluated, without failing, on rows that the original plan
would not evaluate. `ExprUtils.canEvaluateUnconditionally` (`ExprUtils.scala`
L242-L267, SPARK-58066) answers that question for `FilterExec`,
`StreamedSideJoinCondition` and `RewriteDistinctAggregates`, with its own
allowlist. The two lists already differ in both directions:
- Only this one accepts a numeric up-cast, the normalization of a floating
point key, and a DPP filter on a safe key. The tests at
`InjectRuntimeFilterSuite.scala` L1431-L1449 and L1593 need these.
- Only `ExprUtils` accepts `Coalesce`, `IsNaN`, `GetStructField`,
`GetArrayStructFields` and `GetMapValue`. So with `withDppTestConf()`, `select
* from bf5part join bf2 on bf5part.f5 = bf2.c2 where coalesce(bf2.a2, 0) = 36`
gets no bloom filter for its dropped DPP filter, although its creation filter
cannot fail.
Keeping the three local arms and delegating the rest, for example `case _ =>
ExprUtils.canEvaluateUnconditionally(condition)` in `isSafeCondition`, keeps
every test's outcome as far as I can tell, and a later fix to either list would
reach both. An up-cast nested under a comparison would then need the local arms
below the operators too. Two smaller points on the same helpers:
- The decimal-to-integral exclusion at L417 copies the rule of
`Cast.nullable` for `try_cast` (`Cast.scala` L804-L808). One named rule in
`object Cast` would keep the two in step once the TODO there is done.
- A bloom filter that this rule adds at a lower join,
`BloomFilterMightContain(_, XxHash64(Seq(safeKey), _))`, cannot fail either,
and the same expression runs in the join. With broadcasts off, `select * from
bf5part p join (select bf2.c2 from bf2 join bf3 on bf2.c2 = bf3.c3 where bf2.a2
= 36 and bf3.a3 = 36) d on p.f5 = d.c2` loses the bloom filter on `p.f5`,
because the inner join adds one on `bf2.c2` first.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ 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, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // 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
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
+ case And(left, right) => isSafeCondition(left) && isSafeCondition(right)
+ case Or(left, right) => isSafeCondition(left) && isSafeCondition(right)
Review Comment:
**6. [Low] No test pins the `Or`, `Not`, `IsNull`, `IsNotNull`, `In` and
`InSet` arms**
The only creation filter in the tests that `isSafeCondition` must reject is
`cast(c.s as int) > 0` in `castInFilter` (`InjectRuntimeFilterSuite.scala`
L1551), so only the `And` and `BinaryComparison` arms are pinned. Using `||`
here, dropping `isSafeValue(value)` from the `In` arm, or returning `true` from
the `Not`, `IsNull`, `IsNotNull`, `InSet` or default arm passes every test. Yet
with `||`, `castInFilter` with `(c.t > 5 or cast(c.s as int) > 0)` gets the
bloom filter on `f.a`, and its subquery fails with `CAST_INVALID_INPUT` on `q =
99`.
Variants of `castInFilter` in that test would pin each arm, each expecting
the key `t`:
```scala
Seq("(c.t > 5 or cast(c.s as int) > 0)", "not (cast(c.s as int) in (1, 2))",
"cast(c.s as int) is null", "cast(c.s as int) is not null",
s"cast(c.s as int) in (${(1 to 11).mkString(", ")})",
"cast(cast(c.s as int) as boolean)"
).foreach { cond =>
val q = castInFilter.replace("cast(c.s as int) > 0", cond)
assert(bloomFilterKeys(q).flatMap(_.references.map(_.name)) == Seq("t"), q)
}
```
`not (cast(c.s as int) > 0)` would not reach the `Not` arm, since
`BooleanSimplification` turns it into `<=`, and 11 literals make `OptimizeIn`
produce an `InSet`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ 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. An uncorrelated 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
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). 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).
+ // 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)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
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 no 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)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
+ key: Expression,
+ side: LogicalPlan,
+ creation: FilterCreationSide) =
+ !hasDroppedDpp || (canPushDown(key, side) &&
canEvaluateAgain(creation))
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. The key has no dropped DPP filter, or step 3 above allows
the bloom filter
Review Comment:
**10. [Nit] Item 4 is no longer a condition of this `if`**
becfca3 moved this check out of the `if` into
`.filter(canTakeBloomFilter(...))` at L573 and L586, but item 4 here and at
L582 still lists it among the conditions of the `if`, which now has three.
Moving item 4 next to the `.filter`, or dropping it, would match. Step 3 of the
overview at L499-L501 also says only "its creation side is safe", and naming
`canEvaluateAgain` there would point to the new check.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ 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. An uncorrelated 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
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). 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).
+ // 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)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
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 no 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)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
Review Comment:
**11. [Nit] One helper could replace `hasDroppedDpp`, `creationPlan` and
`canTakeBloomFilter`**
These three are used only in the two chains at L572-L574 and L585-L587, and
each chain names its side and key twice, in different orders:
`extractBeneficialFilterCreatePlan(left, creationPlan(right), l, r)` and then
`canTakeBloomFilter(l, left, _)`. One local helper would make the same calls in
the same order:
```scala
def checkedCreationSide(
side: LogicalPlan, key: Expression, other: LogicalPlan, otherKey:
Expression) = {
val creationPlan = withoutDppFilters(other, keyDpps.contains)
extractBeneficialFilterCreatePlan(side, creationPlan, key, otherKey)
.filter(c => keyDpps.isEmpty || (canPushDown(key, side) &&
canEvaluateAgain(c)))
}
```
Each branch would then be `checkedCreationSide(left, l, right,
r).foreach(...)` and its mirror. This is optional, since `canTakeBloomFilter`
is already the single gate.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ 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() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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",
Review Comment:
**7. [Nit] This test and the CTE test check the same logical plan twice**
`withDppTestConf` runs its body with AQE on and then off, unless the confs
set it (L309-L311). This test and the CTE test at L1520 assert only on
`sql(query).queryExecution.optimizedPlan`, through `getNumBloomFilters` and
`bloomFilterKeys`. The rule reads the AQE setting only for a null-aware anti
join (`InjectRuntimeFilter.scala` L390), which neither query has, so the second
run repeats the first. Setting `SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->
"false"` in both would save that run, and their bullets in the description
would then say "with AQE off", as the other AQE-off tests do.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ 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() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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)
Review Comment:
**12. [Nit] This hand-written control is now `assertBloomFilterWithoutDpp`**
Since `assertBloomFilterWithoutDpp` counts exactly, L1377-L1381 can be
`assertBloomFilterWithoutDpp(query)` and
`assert(bloomFilterKeys(query).isEmpty)`, as the CTE test does at L1523-L1524.
Also, `bloomFilterKeys(q).flatMap(_.references.map(_.name))` appears at L1498,
L1561, L1587 and L1613, and `keysOnF5` at L1420 is a variant of it, so a small
helper next to `bloomFilterKeys` would shorten them. The answer check at
L1499-L1502 should stay as it is: `assertRewroteWithBloomFilter` would also
assert that column pruning has no effect on the subquery (L251), and as far as
I can trace, a second `ColumnPruning` removes a Project between the two filters
of that query's subquery.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ 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() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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 can fail") {
+ // On a partition column, the bloom filter would become a partition
filter, which evaluates
+ // the key on every partition before the data filters (SPARK-60809). An
alias hides the key only
+ // at the join, since pushdown rewrites the bloom filter through it. And
`coalesce` cannot be
+ // null, so no inferred `isnotnull` evaluates the key on every partition
first.
+ // Through a union, every branch counts, and without DPP each gets a bloom
filter. A narrowing
+ // cast can fail too.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36" -> 1,
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36" -> 1,
+ "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" -> 2,
+ "select * from bf5part join bf2 on cast(bf5part.f5 as tinyint) = bf2.c2
" +
+ "where bf2.a2 = 36" -> 1
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assertDidNotRewriteWithBloomFilter(query)
+ val filters = dynamicPruningFilters(query)
+ assert(filters.nonEmpty && filters.forall(_ == Literal.TrueLiteral),
filters)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter in an in subquery whose dpp
filter reuses a " +
+ "broadcast outside it") {
+ // The optimizer runs on the subquery on its own, where the broadcast of
`x` is not visible.
+ // The DPP filter on `p.f5` reuses it once the subquery is joined with the
rest of the query.
+ val query = "select /*+ broadcast(x) */ * from bf3 join (select c2 from
bf2 where a2 = 62) x " +
+ "on bf3.c3 = x.c2 where bf3.a3 in (select p.a5 from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) d on p.f5 = d.c2)"
+ def keysOnF5(query: String): Seq[Expression] =
+ bloomFilterKeys(query).filter(_.references.exists(_.name == "f5"))
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(keysOnF5(query).isEmpty)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ // Without the hint, nothing broadcasts `x`, and `p.f5` gets the bloom
filter.
+ val withoutHint = query.replace("/*+ broadcast(x) */ ", "")
+ assert(keysOnF5(withoutHint).nonEmpty)
+ assert(dynamicPruningFilters(withoutHint) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter
under a numeric " +
+ "up-cast") {
+ // Type coercion casts `f5` to `bigint`, which cannot fail.
+ val query = "select * from bf5part join (select cast(c2 as bigint) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter of a
floating point " +
Review Comment:
**13. [Nit] This test copies the up-cast test above**
It differs from the test at L1431 only in `double` for `bigint`. A loop over
both types would keep the queries, confs and assertions, and both arms of
`isSafeKey` would stay pinned:
```scala
Seq("bigint", "double").foreach { t =>
val query = s"select * from bf5part join (select cast(c2 as $t) k, a2 from
bf2) d " +
"on bf5part.f5 = d.k where d.a2 = 36"
withDppTestConf() {
withClue(s"$t: ")(assertRewroteWithBloomFilterForDroppedDpp(query))
}
}
```
Feel free to keep them separate, since I asked for this test in [round
3](https://github.com/apache/spark/pull/59916#discussion_r4462890709).
--
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]
