dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4462890686
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,21 +378,59 @@ 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.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
Review Comment:
**2. [Medium] `Cast.canUpCast` accepts `DECIMAL(19,0)` to `BIGINT`, which
can overflow**
`isSafeKey`, which answers [round-2 item
4](https://github.com/apache/spark/pull/59916#discussion_r4448448582), takes
`Cast.canUpCast` to mean that the cast cannot fail. For a decimal to an
integral type, `canUpCast` asks `DecimalType.isTighterThan` (`UpCastRule.scala`
L37), which accepts a precision below that of `DecimalType.forType`
(`DecimalType.scala` L87-L92). That is right for `TINYINT`, `SMALLINT` and
`INT`. But `LongDecimal` is `DecimalType(20, 0)` (`DecimalType.scala` L128)
while `Long.MaxValue` has 19 digits, so `DECIMAL(19,0)` to `BIGINT` passes, and
the ANSI cast throws `CAST_OVERFLOW` above `Long.MaxValue` (`Cast.scala`
L1178-L1180, `Decimal.scala` L335-L341). The TODO at `Cast.scala` L804-L806
already notes the hole. Among the numeric pairs that `canUpCast` accepts, this
is the only one that can throw.
With ANSI on and the suite's DPP test thresholds, take `fd(v INT, d
DECIMAL(19,0))` partitioned by `d`, with a partition `d = 36` whose rows have
`v > 0` and a partition `d = 9999999999999999999` whose rows all have `v <= 0`:
```sql
SELECT * FROM fd JOIN bf2 ON CAST(fd.d AS BIGINT) = bf2.c2 WHERE fd.v > 0
AND bf2.a2 = 36
```
- `PartitionPruning` adds an `onlyInBroadcast` DPP filter on `CAST(fd.d AS
BIGINT)`, since it needs only the references to be partition columns
(`PartitionPruning.scala` L68). The shuffle join drops it.
- The leaf key `cast(d AS BIGINT)` passes this check. The bloom filter
becomes a dynamic partition filter that evaluates the key on every partition
(`DataSourceScanExec.scala` L423-L437), so the query fails.
- On master the DPP filter keeps the bloom filter off. The inferred null
check is only `isnotnull(d)`, since `Cast` is null-intolerant (`Cast.scala`
L709, `QueryPlanConstraints.scala` L198-L203). For an aliased key,
SimplifyCasts gives the same (`expressions.scala` L1335-L1336). So the cast
runs only on rows with `v > 0`, and the query succeeds. Round 2 rejected this
key.
Master already fails the same query when the DPP filter is kept, the
SPARK-60809 class, but here the PR adds the failure for a key that this check
says cannot fail. I traced this from the code and did not run it.
Excluding `(DecimalType, IntegralType)` here, as `Cast.nullable` does for
`try_cast` (`Cast.scala` L807), and adding this query as a negative test would
close it. Type coercion never produces this cast, so only an explicit `CAST`
would lose its bloom filter. The off-by-one in `isTighterThan` predates this PR.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
+ canPushDown(key, side) && creation.deterministic &&
+ !creation.containsPattern(CTE) &&
!creation.exists(_.expressions.exists(_.exists {
Review Comment:
**1. [Medium] The kept-DPP check does not see DPP filters that reach the
creation scan later**
This check, which answers [round-2 item
3](https://github.com/apache/spark/pull/59916#discussion_r4448448570), looks
only at the DPP filters inside `creationPlan(other)` while the rule runs. The
bloom filter's subquery is fixed at injection (L84, L90), so two kinds of DPP
filters still prune the creation scan in the main plan but not in that subquery:
- A kept DPP filter that `PartitionPruning` put on an ancestor join.
`PartitionPruning` runs bottom-up (`PartitionPruning.scala` L316) and puts it
on top of this join, as a child of the upper join. This rule also runs
bottom-up (L498) and decides this join first. Only the later batch "Pushdown
Filters from PartitionPruning" (`SparkOptimizer.scala` L81-L82) moves the
filter into the creation scan, and that batch does not enter subquery plans.
- A DPP filter with a benefit inside the creation side but above a join.
L528 lets it through because it also runs as its own subquery in the bloom
filter's subquery, but `extract` restarts below the join (L151-L153), so
`creationSide.plan` does not contain it.
With ANSI on, `spark.sql.autoBroadcastJoinThreshold=-1`, the
application-side threshold at 1, and AQE on or off, take `ra(v, x, a)`
partitioned by `a`, `rc(s STRING, flag, q)` partitioned by `q` with rows `('1',
1, 10)`, `('2', 1, 20)` and `('abc', 1, 99)`, and `rd(k, dflag)` with rows
`(10, 1)` and `(20, 1)`:
```sql
SELECT /*+ BROADCAST(d) */ f.v FROM ra f
JOIN rc c ON f.a = CAST(c.s AS INT)
JOIN rd d ON c.q = d.k
WHERE f.x = 1 AND c.flag = 1 AND d.dflag = 1
```
- `PartitionPruning` puts `dpp(f.a)` (build plan: the `rc` side) on the
lower join, and `dpp(c.q)` (build plan: the `rd` side) on top of the lower
join. `reused` holds the hinted `rd` side, so `dpp(c.q)` is kept and `dpp(f.a)`
is dropped.
- At the lower join, `creation` is `Filter(flag = 1, rc)` with no DPP
filter, so `f.a` gets a bloom filter that hashes `CAST(s AS INT)` over all of
`rc`.
- In the main plan, `dpp(c.q)` moves into the `rc` scan and reuses the
broadcast of `rd` (`PlanDynamicPruningFilters.scala` L63-L70), so `q = 99` is
pruned. The bloom filter's subquery is planned on its own, reads `q = 99`, and
fails with `CAST_INVALID_INPUT`.
- On master, `dpp(f.a)` keeps the bloom filter off this key, and the query
returns rows.
The second kind fails the same way with `reuseBroadcastOnly=false`, when
`ra` is joined on `f.a = s.t` to `(SELECT /*+ BROADCAST(y) */ c.t FROM rc c
JOIN ry y ON c.u = y.u JOIN rd d ON c.q = d.k WHERE c.flag = 1 AND CAST(c.s AS
INT) > 0 AND d.dflag = 1) s`, and `dpp(c.q)` has a benefit while `dpp(f.a)`
does not. The new tests cover a DPP filter on the same join and one directly
over the scan (`InjectRuntimeFilterSuite.scala` L1480 and L1505). I traced this
from the code and did not run it.
One way to close both: find, in the whole plan, every DPP filter whose
pruning key reaches a leaf of `creationSide.plan` through
`findExpressionAndTrackAllLineagesDown`, and add the bloom filter only when
each of them is inside `creationSide.plan` and is not `onlyInBroadcast`. The
same test would also make item 8 exact.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +303,59 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
checkWithAndWithoutFeatureEnabled(query, shouldReplace = false)
}
+ // Runs `f` with the thresholds of the DPP tests, `reuseBroadcastOnly` on,
and `confs`, with AQE
+ // on and off unless `confs` sets it.
+ def withDppTestConf(confs: (String, String)*)(f: => Unit): Unit = {
+ val adaptiveModes = confs.collectFirst {
+ case (SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, adaptive) => Seq(adaptive)
+ }.getOrElse(Seq("true", "false"))
+ adaptiveModes.foreach { adaptive =>
+ withSQLConf(Seq(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive) ++ confs: _*) {
+ withClue(s"adaptive=$adaptive: ")(f)
+ }
+ }
+ }
+
+ // Runs `query` and returns the DPP filters that physical planning left on
the file source scans:
+ // the class of the subquery that each filter reads, or the filter itself if
it reads none.
+ def dynamicPruningFilters(query: String): Seq[Any] = {
+ // Keep the scans in the final plan even if they return no rows.
+ withSQLConf(
+ SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key ->
AQEPropagateEmptyRelation.ruleName) {
+ val df = sql(query)
+ df.collect()
+ collect(df.queryExecution.executedPlan) {
+ case scan: FileSourceScanExec => scan.partitionFilters.collect {
+ case DynamicPruningExpression(in: InSubqueryExec) => in.plan.getClass
+ case DynamicPruningExpression(filter) => filter
+ }
+ }.flatten
+ }
+ }
+
+ // Asserts that `query` gets a bloom filter, and that physical planning
plans its DPP filter as
+ // `true`.
+ def assertRewroteWithBloomFilterForDroppedDpp(query: String): Unit = {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+
+ // The keys that the bloom filters of `query` check.
+ def bloomFilterKeys(query: String): Seq[Expression] =
+ sql(query).queryExecution.optimizedPlan.flatMap(_.expressions)
+ .flatMap(_.collect { case BloomFilterMightContain(_, XxHash64(Seq(key),
_)) => key })
+
+ // Asserts that `query` gets a bloom filter without DPP, so that a test that
sees none with DPP
+ // shows that the rule kept it out.
+ def assertBloomFilterWithoutDpp(query: String): Unit =
Review Comment:
**6. [Nit] `assertBloomFilterWithoutDpp` passes on a bloom filter on any
key**
The control passes as soon as any key has a bloom filter. In the two tests
with two candidate keys, the other key keeps it green even if the key under
test stops being eligible.
- In the test at L1149, DPP off gives bloom filters on both `p.f5` and
`t.f5`. If `bf2` shrank below the 2000-byte broadcast threshold, the first join
would become a broadcast join without a shuffle below it, and `p.f5` would lose
eligibility. The control (through `t.f5`), the negative check and the reuse
check at L1162 would all still pass, so the cross-branch path would go
unasserted.
- In the test at L1480, `f.b` alone satisfies the control.
Asserting the count there, or the key as L1525 does, would make these
controls specific.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
+ canPushDown(key, side) && creation.deterministic &&
+ !creation.containsPattern(CTE) &&
!creation.exists(_.expressions.exists(_.exists {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast &&
isKept(dpp)
Review Comment:
**5. [Low] No test pins the `isKept` half of the new creation check**
Replacing `dpp.onlyInBroadcast && isKept(dpp)` with `dpp.onlyInBroadcast`
passes every test, although it would also turn away a creation side whose
`onlyInBroadcast` DPP filter is dropped. Of the tests at
`InjectRuntimeFilterSuite.scala` L1505 and L1480, the first sets
`reuseBroadcastOnly` to false, so its DPP filter is not `onlyInBroadcast`, and
the second has a kept one, which both versions reject. Running the first query
with the default `reuseBroadcastOnly` would pin it. Both DPP filters are then
dropped: a3b03fe gives bloom filters on `f.a` and `f.b`, and the narrower check
moves the first one to `c.t`. The count stays two, so the test would have to
assert the keys.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
Review Comment:
**8. [Nit] The kept-DPP check looks at the whole creation join side**
Like the determinism and CTE checks, the new check scans all of
`creationPlan(other)`. The bloom filter's subquery reads only what
`extractSelectiveFilterOverScan` returns: a Project/Filter chain over one leaf,
restarted below each join (L151-L153). So a kept DPP filter on another table of
that side turns away a bloom filter that could never see it.
For example, in `f1 JOIN (SELECT d.k, f2.v FROM dim d JOIN f2 ON d.j = f2.p
WHERE d.x = 1) s ON f1.p = s.k`, the inner join broadcasts `dim` and keeps the
DPP filter on `f2.p`. The bloom filter on `f1.p` would read only `Filter(x = 1,
dim)`, but the check finds the kept filter in `s` and rejects it. That only
loses the optimization, as on master, but the Known limits name only the
determinism and CTE checks.
Checking just `creationSide.plan` at the call sites would not fix this. A
kept `onlyInBroadcast` DPP filter above a join inside the creation side is not
in `creationSide.plan` either, so that would reopen round-2 item 3. The
lineage-based test from item 1 would decide both cases.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,478 @@ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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 is not an " +
Review Comment:
**13. [Nit] This test name no longer matches the rule**
Since `isSafeKey`, a key that is not an attribute can get the bloom filter,
as the up-cast test at L1427 shows, so "is not an attribute" contradicts the
next test. The description already says "that can fail", and "... if the key of
the dropped dpp filter can fail" would match.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala:
##########
@@ -118,36 +118,42 @@ trait PredicateHelper extends AliasHelper with Logging {
*/
def findExpressionAndTrackLineageDown(
exp: Expression,
- plan: LogicalPlan): Option[(Expression, LogicalPlan)] = {
- if (exp.references.isEmpty) return None
+ plan: LogicalPlan): Option[(Expression, LogicalPlan)] =
+ findExpressionAndTrackAllLineagesDown(exp, plan).headOption
+
+ /**
+ * Like [[findExpressionAndTrackLineageDown]], but returns every lineage,
not only the first one.
+ * It follows each branch of a union, and each child that has the references
of exp. With
+ * `strict`, it returns no lineage at all if it cannot follow exp down one
of those.
+ */
+ def findExpressionAndTrackAllLineagesDown(
+ exp: Expression,
+ plan: LogicalPlan,
+ strict: Boolean = false): Seq[(Expression, LogicalPlan)] = {
+ if (exp.references.isEmpty) return Nil
+
+ // The lineages of each expression in the plan it is paired with.
+ def lineages(pairs: Seq[(Expression, LogicalPlan)]): Seq[(Expression,
LogicalPlan)] = {
+ val found = pairs.map { case (e, p) =>
findExpressionAndTrackAllLineagesDown(e, p, strict) }
+ if (strict && found.exists(_.isEmpty)) Nil else found.flatten
Review Comment:
**10. [Nit] The strict walk also gives up on a literal branch**
A union branch whose key has no references returns `Nil` at L133, and
`strict` turns that into a failure here. So `SELECT f5 AS k FROM bf5part UNION
ALL SELECT 0 AS k FROM bf1`, joined to `bf2` on `u.k = bf2.c2`, gets no bloom
filter for its dropped DPP filter, where 2420b70 gave it one. Pushdown would
put `might_contain(..., xxhash64(0))` into that branch, which cannot fail. The
Known limit about a branch the rule cannot follow covers this in general terms.
If it is worth handling, only a `Literal` branch should pass, since a
reference-free key such as `raise_error(...)` is not foldable (`misc.scala`
L85) and can fail.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 {
Review Comment:
**15. [Nit] Two ways to collect the DPP filters of a plan**
`reused` collects DPP filters as conjuncts of `Filter` conditions here, and
the new creation check (L527-L530) walks every expression of every node for the
same filters. For the plans this rule sees, they find the same filters, since
`PartitionPruning` creates only `Filter(dpp, _)` (`PartitionPruning.scala`
L120-L127) and later rules only merge it as a conjunct. One helper, for example
`dppFilters(plan)`, could serve both, as
`dppFilters(plan).flatMap(reusablePlans)` here and
`!dppFilters(creation).exists(d => d.onlyInBroadcast && isKept(d))` there, so
that a later change to one cannot miss the other. A
`containsPattern(DYNAMIC_PRUNING_SUBQUERY)` guard would also skip the walk for
a creation side without DPP filters.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
+ canPushDown(key, side) && creation.deterministic &&
Review Comment:
**7. [Low] A lazily checkpointed `LogicalRDD` also hides non-determinism**
The comment at L518-L520 and the description name a CTE reference as what
hides the determinism of a creation side. A `LogicalRDD` from
`Dataset.checkpoint(eager = false)` does too: it is a leaf, so `deterministic`
is true, and it carries the statistics of its source plan (`ExistingRDD.scala`
L150-L151), so it can pass the creation-side threshold. Nothing persists it, so
the bloom filter's subquery job computes the RDD first, the checkpoint at the
end of that job computes it again, and the join reads the checkpoint. Keys from
the second run that the first run did not produce are then filtered out.
With `cast(reflect('java.lang.Math', 'random') AS DOUBLE) < 0.5` above such
a checkpoint, and a fact table partitioned by the key in a shuffle join, rows
would be lost where master, whose DPP filter keeps the bloom filter off,
returns them. `PartitionPruning` guards the same hazard for its
materialized-input path with `LogicalRDD.isCheckpointedInput`
(`PartitionPruning.scala` L274-L276), but `LogicalRDD` lives in sql/core. The
shape is contrived, so a line in the Known limits and a wording change here may
be enough. I traced this from the code and did not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
+ canPushDown(key, side) && creation.deterministic &&
+ !creation.containsPattern(CTE) &&
!creation.exists(_.expressions.exists(_.exists {
Review Comment:
**9. [Nit] The CTE check also turns away deterministic `MATERIALIZED` CTEs**
After `InlineCTE`, the CTEs left are non-deterministic ones with several
references and every `AS MATERIALIZED` one (`InlineCTE.scala` L88-L98). So
`!creation.containsPattern(CTE)` also turns away a deterministic `MATERIALIZED`
CTE. For example, with the tables of the new CTE test:
```sql
WITH d AS MATERIALIZED (SELECT k, x FROM cdim)
SELECT count(*) FROM d JOIN (SELECT k, sum(v) s FROM cfact GROUP BY k) t
ON d.k = t.k WHERE d.x = 1
```
`t.k` now gets no runtime filter, where 2420b70 gave it a correct bloom
filter.
The blanket check is still needed for a non-deterministic CTE, even with
exchange reuse on. A `COALESCE` hint makes the definition a `Repartition`
without a shuffle (`ResolveHints.scala` L196-L197), which
`ReplaceCTERefWithRepartition` keeps as is
(`ReplaceCTERefWithRepartition.scala` L44-L48), so there is no exchange to
share.
A precise version would fold the definitions of each `WithCTE`, in order,
into the set of ids whose plan is deterministic and reads only CTEs already in
the set, and reject only a reference outside that set. At least the Known
limits could say that a deterministic CTE also keeps the bloom filter out.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. So that side
+ // must be deterministic (SPARK-60808), and must not read a CTE,
whose reference hides
+ // the determinism of its plan. Nor may it keep an
`onlyInBroadcast` DPP filter that
+ // physical planning keeps in the join. In the bloom filter's
subquery that filter finds
+ // no broadcast, so the creation side would read partitions that
the join never reads.
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || {
+ val creation = creationPlan(other)
+ canPushDown(key, side) && creation.deterministic &&
+ !creation.containsPattern(CTE) &&
!creation.exists(_.expressions.exists(_.exists {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast &&
isKept(dpp)
+ case _ => false
+ }))
+ }
// 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)
- }
+ !hasBloomFilter(newLeft, l) && canTakeBloomFilter(l, left,
right)) {
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .foreach(creationSide => newLeft = injectFilter(l, newLeft,
creationSide))
}
// Did we actually inject on the left? If not, try on the right
// Check if:
// 1. The current join type supports prune the right 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 right key yet
+ // 4. The key has no dropped DPP filter, or step 3 above allows
the bloom filter
if (newLeft.fastEquals(oldLeft) && canPruneRight(joinType) &&
- (hasShuffle || probablyHasShuffle(right)) &&
!hasBloomFilter(newRight, r)) {
- extractBeneficialFilterCreatePlan(right, left, r, l).foreach {
creationSide =>
- newRight = injectFilter(r, newRight, creationSide)
- }
+ (hasShuffle || probablyHasShuffle(right)) &&
!hasBloomFilter(newRight, r) &&
+ canTakeBloomFilter(r, right, left)) {
+ extractBeneficialFilterCreatePlan(right, creationPlan(left), r,
l)
Review Comment:
**3. [Low] Another bloom filter can narrow the creation scan only in the
main plan**
The bloom filter's subquery is built from the creation side as it was before
this join's injections, and nothing checks that the creation key cannot fail. A
bloom filter on a sibling key of the same join goes into `newLeft`, but this
line builds the creation side from the original `left`. A bloom filter that an
upper join adds later reaches the creation scan only in the main plan
(`SparkOptimizer.scala` L81-L82). A fix for item 1 that follows DPP filters
would not see either.
With ANSI on, broadcast threshold -1 and the application-side threshold at
1, take `sf(s STRING, x INT, y INT)` with rows `('10', 1, 1)`, `('20', 1, 2)`
and `('abc', 1, 999)`, and `sc(y INT, flag INT, q INT)` partitioned by `q` with
rows `(1, 1, 10)` and `(2, 1, 20)`:
```sql
SELECT * FROM sf f JOIN sc c ON f.y = c.y AND CAST(f.s AS INT) = c.q
WHERE f.x = 1 AND c.flag = 1
```
- `f.y` has no DPP filter and gets a bloom filter from `sc`, as on master.
- `c.q` has a dropped DPP filter. Its left side fails `isSafeKey`, so the
rule tries the right side here. It builds a bloom filter that hashes `CAST(s AS
INT)` over the original `sf` side, all three rows, so it fails with
`CAST_INVALID_INPUT`.
- On master `c.q` is skipped, and the first bloom filter removes the `'abc'`
row before the join evaluates the cast.
Master's success rests on that bloom filter, so it is not guaranteed: a
false positive, or runtime bloom filters turned off, makes master fail too.
That is why I marked this Low. Still, the PR turns it into a deterministic
failure. The same holds when an upper join adds a bloom filter on the creation
side, for example item 1's query with `rc` not partitioned. Requiring the
creation key to be safe along its lineage, or evaluating it null-on-error in
the bloom filter aggregate, would close this and the creation-key part of item
1. 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,478 @@ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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 is not an " +
+ "attribute") {
+ // 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.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36",
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36",
+ "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"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ 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 " +
Review Comment:
**4. [Low] No test pins what `isSafeKey` rejects**
This test is the only one whose outcome depends on the `Cast` arm of
`isSafeKey`, and it is positive. Replacing `Cast.canUpCast(...)` with `true`
changes no assertion, although it would accept keys such as `CAST(bigint_col AS
INT)` that can overflow. Neither does removing the
`KnownFloatingPointNormalized(NormalizeNaNAndZero(_))` arm at
`InjectRuntimeFilter.scala` L416, since every key in the suite is `INT`,
`BIGINT` or `STRING`. A negative test with a narrowing cast such as `cast(f5 as
tinyint)`, or with item 2's query, and a positive test with a `DOUBLE` key such
as `bf5part.f5 = cast(c2 as double)`, each with the DPP-off control, would pin
both arms.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
Review Comment:
**12. [Nit] The overview names `canTakeBloomFilter` as the creation-side
check**
Step 3 reads "only if pushdown can take it down safely (`canPushDown`) and
its creation side is safe (`canTakeBloomFilter`)". But `canTakeBloomFilter`
(L523-L531) is the whole gate: it calls `canPushDown` itself, and the call
sites at L540 and L553 use only it. Something like "only if
`canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its creation side is safe" would keep a later change from
adding a pushdown condition outside the gate.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +448,111 @@ 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 pushdown can take
+ // it down safely (`canPushDown`) and its creation side is safe
(`canTakeBloomFilter`).
+ // 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 already a DPP filter on the key that physical
planning may keep
Review Comment:
**11. [Nit] This comment states the opposite of the condition**
`2. There is already a DPP filter on the key that physical planning may
keep` describes the case that the condition below excludes
(`!keyDpps.exists(isKept)`). The other items here and at L532-L537 say what
must hold, such as "There is no bloom filter on the left key yet". The wording
comes from master, but since this line changed anyway, "There is no DPP filter
on the key that physical planning may keep" would match the code.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,478 @@ 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() {
+ assertBloomFilterWithoutDpp(query)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ 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 is not an " +
+ "attribute") {
+ // 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.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36",
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36",
+ "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"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ 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: do not add bloom filter whose key can fail in a union
branch the lineage " +
+ "does not reach") {
+ // The lineage of `u.k` stops at the inner union of the second branch, but
pushdown moves the
+ // bloom filter into both inner branches, where it would cast `p` on every
partition. Each
+ // row of `p = 'abc'` has `v <= 0`, so the query itself never casts
`'abc'`.
+ withTable("sp") {
+ sql("create table sp (v int, p string) using parquet partitioned by (p)")
+ sql("insert into sp select cast(id as int) + 1, '36' from range(100)")
+ sql("insert into sp select -cast(id as int), 'abc' from range(100)")
+ val query = "select * from (select f5 as k, a5 as v from bf5part union
all " +
+ "select coalesce(cast(x as int), 0) as k, sum(v) as v from (select p
as x, v from sp " +
+ "where v > 0 union all select p, v from sp where v > 0) group by x) u
" +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side reads a cte")
{
+ // The reference to `d` hides that its plan is not deterministic. With
exchange reuse off, the
+ // bloom filter's subquery would draw a second sample of `d`.
+ withTable("cdim", "cfact") {
+ spark.range(200).selectExpr("cast(id as int) k", "cast(id % 2 as int) x")
+ .write.saveAsTable("cdim")
+ spark.range(20000).selectExpr("cast(id % 200 as int) k", "id v")
+ .write.partitionBy("k").saveAsTable("cfact")
+ sql("analyze table cdim compute statistics")
+ sql("analyze table cfact compute statistics")
+ val query = "with d as (select k, x from cdim " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) " +
+ "select count(*), count_if(t.k is null) from d left join (select k,
sum(v) s from cfact " +
+ "group by k) t on d.k = t.k where d.x = 1 union all select count(*), 0
from d"
+ withDppTestConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1") {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side keeps a dpp
filter that " +
+ "physical planning keeps") {
+ // The DPP filter on `c.q` reuses the broadcast of `g`, so the join reads
only the partitions
+ // of `rc` it keeps. In the bloom filter's subquery on `f.a` it finds no
broadcast, so that
+ // subquery would read `q = 99` too, and cast `'abc'`.
+ withTable("ra", "rc", "rx") {
+ sql("create table ra (v int, x int, b int, a int) using parquet
partitioned by (a)")
+ sql("insert into ra select cast(id as int), 1, 10, 1 from range(2000)")
+ sql("insert into ra select cast(id as int), 1, 20, 2 from range(2000)")
+ sql("create table rc (s string, flag int, q int) using parquet
partitioned by (q)")
+ sql("insert into rc values ('1', 1, 10), ('2', 1, 20), ('abc', 1, 99)")
+ sql("create table rx (k1 int, k2 int) using parquet")
+ sql("insert into rx values (1, 10), (2, 20)")
+ val query = "select f.v from ra f join rc c on f.a = cast(c.s as int)
and f.b = c.q " +
+ "where f.x = 1 and c.flag = 1 union all select /*+ broadcast(g) */ g.v
from rx " +
+ "join (select * from ra where x = 1) g on rx.k1 = g.a and rx.k2 = g.b"
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
Review Comment:
**14. [Nit] The description does not say that this test runs only with AQE
off**
The description says the new tests run "with AQE on and off unless noted".
This one pins AQE off, and its bullet does not say so, while the other AQE-off
tests are marked. A reader would take the new creation check as tested under
AQE too. Adding "with AQE off" to that bullet would fix it.
--
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]
