dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4437009548
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
Review Comment:
**1. [High] A key whose dropped DPP filter has a non-deterministic build
plan now gets a bloom filter**
`isKept` is false for such a filter, so the rule now builds a bloom filter
by evaluating that build side a second time. `CleanupDynamicPruningFilters`
rewrites a non-deterministic DPP filter to `true` to avoid exactly this, and
the comment at `SparkOptimizer.scala` L142-L144 calls it correctness behavior.
`extract` does not stop it, since its only determinism check is on the
materialized path (L195, while L135 and L237 have none).
Every `dim.k` exists in `fact` here, so `lost` must be 0:
```sql
SELECT count(*) total, count_if(f.k IS NULL) lost
FROM (SELECT k FROM dim WHERE cast(reflect('java.lang.Math', 'random') AS
DOUBLE) < 0.5) d
LEFT JOIN fact f ON d.k = f.k
```
With `fact` partitioned by `k` and `autoBroadcastJoinThreshold=-1`, a
released Spark 4.3.0 gives `[109,0]` with its default configs, where the DPP
filter keeps the bloom filter off this key. With DPP off, which adds the same
bloom filter, it gives `[94,47]`, `[103,46]` and `[111,63]`. A `dim` row in the
join's sample whose key is not in the subquery's sample loses its `fact` row.
The fact scan must be over the application-side threshold, and at the
default broadcast threshold there must be a shuffle under `f`, such as an
aggregate. `rand()` mostly hides this, because its fixed seed repeats the
sample when both scans read the same splits, so `reflect`, `java_method` and
`try_reflect` are the clear triggers.
Counting `!dpp.deterministic` as kept here would keep master's outcome for
these keys. As it stands, this contradicts "The results stay the same" in the
description.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
Review Comment:
**5. [Low] `MergeSubplans` can change a reused broadcast after this check**
This checks where the bloom filter lands, but not its subquery.
`MergeSubplans`, in the next batch, can merge the new subquery with a scalar
subquery inside a reused broadcast. It rewrites the broadcast in the main plan
(`MergeSubplans.scala` L231-L248) and does not visit the DPP filter's
`buildQuery`, so `candidatePlan.sameResult(sparkPlan)` becomes false and the
DPP filter on `p.f5` is planned as `true`:
```sql
SELECT /*+ broadcast(d) */ p.a5 FROM bf5part p
JOIN (SELECT * FROM bf2 WHERE b2 > (SELECT percentile_approx(b3, 0.5) FROM
bf3)) d
ON p.f5 = d.c2
UNION ALL
SELECT q.a5 FROM bf5part q JOIN (SELECT * FROM bf3 WHERE a3 = 62) x ON q.f5
= x.c3
```
On master `q.f5` has no bloom filter, so `p.f5` keeps its DPP filter. Here
`p.f5` was already counted as kept, so it ends with no runtime filter at all.
The results are the same.
The conditions are narrow: an object-hash aggregate such as
`percentile_approx` or `mode`, over the same relation as the creation side,
with no filter or the same filter, and a side `x` that cannot be broadcast,
since otherwise `q.f5` counts as kept. `InjectRuntimeFilterSuite` excludes
`MergeSubplans` (`InjectRuntimeFilterSuite.scala` L215), so its tests cannot
see this. This item comes from reading the code.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// 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. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
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) && !(hasDroppedDpp &&
goesIntoReused(l, left))) {
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .foreach(creationSide => newLeft = injectFilter(l, newLeft,
creationSide))
Review Comment:
**3. [Medium] A failable key on a partition column is evaluated before the
data filters**
A bloom filter on a partition key becomes a partition filter
(`FileSourceStrategy.scala` L212-L213) and is evaluated on the driver for each
partition (`DataSourceScanExec.scala` L688), before any data filter.
`isSimpleExpression` accepts a cast, so a key that can fail now runs on
partitions whose rows never reach the join.
```sql
-- fact(v, x) PARTITIONED BY (p STRING); every row of partition p = 'abc'
has x <= 0
SELECT f.v FROM fact f JOIN dim d ON cast(f.p AS INT) = d.k WHERE d.a = 1
AND f.x > 0
```
In a released Spark 4.3.0 with `autoBroadcastJoinThreshold=-1` and ANSI on
(the default), this succeeds with the default configs, because the DPP filter
is `true` and the cast only sees rows that pass `f.x > 0`. With DPP off, which
adds the same bloom filter, it fails with `CAST_INVALID_INPUT` for `'abc'`.
Hive takes the same steps.
The cast does not have to be in the query: under ANSI, `f.p = d.k` with an
integer `d.k` gets a cast key too. The creation side must be non-empty, since a
null bloom filter does not evaluate the key, and a partition with surviving
rows fails on master as well. This is not in the Known limits.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
Review Comment:
**6. [Low] `reused` can see less of the plan than physical planning does**
The rule sees only the plan it runs on, but an uncorrelated IN subquery runs
the whole optimizer on its own first and joins the main plan later through
`RewritePredicateSubquery`. In this query, at `autoBroadcastJoinThreshold=500`,
`p.f5` gets the bloom filter in the subquery's run, where no broadcast is
visible, and physical planning then finds the hinted broadcast of `x` and keeps
the DPP filter:
```sql
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)
```
That leaves a kept DPP filter and a bloom filter on one key. The results are
the same, the cost is one redundant subquery, and the shape needs a hint or a
join type that rules out the broadcast, so this is minor. But "It holds no
fewer, unless the statistics change" here and "Every broadcast that a DPP
filter could reuse is in this list" in the description do not hold, and the
temporary check's "none" covers only the suites it ran over.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11093,6 +11093,42 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
after {
catalog.clearTables()
+ catalog.clearFunctions()
+ }
+
+ test("SPARK-60702: a storage-partitioned join gets the bloom filter that its
dropped dpp " +
+ "filter used to keep out") {
+ // The scan offers the source column of `bucket(4, id)` for runtime
filtering, so
+ // `PartitionPruning` adds a DPP filter on `f.id`. Nothing can broadcast
its build plan, so
+ // physical planning plans it as `true`. The key gets a bloom filter
instead, and the join
+ // still shuffles neither side.
+ catalog.createFunction(Identifier.of(Array.empty,
UnboundBucketFunction.name()),
+ UnboundBucketFunction)
+ createTable("fact", Array(Column.create("id", LongType),
Column.create("v", IntegerType)),
+ Array(bucket(4, "id")))
+ sql("INSERT INTO testcat.ns.fact VALUES " +
+ (0 until 16).map(i => s"($i, ${i * 10})").mkString(", "))
+ createTable("dim", Array(Column.create("id", LongType),
Column.create("flag", IntegerType)),
+ Array(bucket(4, "id")))
+ sql("INSERT INTO testcat.ns.dim VALUES (1, 1), (2, 0), (3, 1), (5, 1), (6,
0), (9, 1)")
+ Seq(true, false).foreach { adaptive =>
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString) {
+ val df = sql("SELECT f.id, f.v FROM testcat.ns.fact f JOIN
testcat.ns.dim d " +
+ "ON f.id = d.id WHERE d.flag = 1")
+ checkAnswer(df, Seq(Row(1, 10), Row(3, 30), Row(5, 50), Row(9, 90)))
+ val plan = df.queryExecution.executedPlan
+ val runtimeFilters = collectScans(plan)
+ .filter(_.table.name().endsWith("fact"))
+ .flatMap(_.runtimeFilters)
+
assert(runtimeFilters.exists(_.exists(_.isInstanceOf[BloomFilterMightContain])),
plan)
Review Comment:
**8. [Low] On a non-identity transform, the new bloom filter does not prune
splits**
This shows that the bloom filter is among the runtime filters, not that it
prunes anything. A bucket number cannot answer a lookup on the hash of `id`, so
this Catalyst fixture receives the bloom filter in `filter()`, plans its
partitions again and keeps every split. A `SupportsRuntimeV2Filtering` scan
does not see it at all, because `V2ExpressionBuilder` has no translation for
it. I did not read Iceberg's scan, the description's example.
So for a storage-partitioned join on a non-identity transform, the new
filter is row-level only. There is no shuffle to save, although
`isProbablyShuffleJoin` returns true from hints and sizes alone, and the cost
is a subquery job plus a lookup per row. Rejected rows do skip the sort and the
join, so this is not necessarily a net loss, and it depends on selectivity.
Could the Known limits say this? The description's example can read as if the
scan gets pruned.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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)) {
Review Comment:
**2. [Medium] A bloom filter on a Hive partition key can fail a query that
succeeds on master**
A bloom filter on a Hive partition key lands in `partitionPruningPred`. Its
subquery sends `HiveTableScanExec.prunedPartitions` to `prunePartitions`
(`HiveTableScanExec.scala` L175), which casts the raw value of every partition
column with no handling of `__HIVE_DEFAULT_PARTITION__`. On master, the dropped
DPP filter is planned as `true`, so there is no subquery and this path returns
the partitions without that cast.
```sql
CREATE TABLE fact(a INT) PARTITIONED BY (p1 INT, p2 INT) STORED AS TEXTFILE;
-- with a partition (p1 = 1, p2 = NULL)
SELECT f.a FROM fact f JOIN dim d ON f.p1 = d.k WHERE d.x = 1;
```
In a released Spark 4.3.0 with `autoBroadcastJoinThreshold=-1`, this returns
its rows with the default configs. With DPP off, which adds the same bloom
filter, it fails with `CAST_INVALID_INPUT` for the value
`__HIVE_DEFAULT_PARTITION__`.
It takes a non-string partition column with a NULL partition that the query
does not select, and table statistics over the application-side threshold. With
a single partition column, the inferred `isnotnull` removes the NULL partition
first. The missing handling in `prunePartitions` is in existing code, and this
PR adds a new way to reach it. The diff has no Hive test.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1067,262 @@ 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") {
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
62",
Review Comment:
**10. [Medium] The new V1 queries return no rows, so a non-null bloom filter
never meets a partition value**
`bf2.a2 = 62` matches no row, and neither do the `a3 = 62`, `d.a5 = 62`,
`b.b5 = 1` and `q.f5 = 51` predicates in the other new queries. So
`checkAnswer` in `checkWithAndWithoutFeatureEnabled` compares an empty result
with an empty result.
For a bloom filter on a V1 partition column (`bf5part.f5`,
`bf5filtered.f5`), the creation side is always empty, so the bloom filter is
null (`BloomFilterAggregate.scala` L195-L197) and `might_contain` returns null
without hashing the key. No committed test evaluates a non-null bloom filter
against partition values, which is the main V1 path of this PR. Only the
description's scratch run covers it, and the V2 suites that do check rows do
not go through it. If that evaluation over-pruned, this suite would still pass.
`bf2.a2 = 36` gives one row here (`c2 = 40` matches `f5 = 40`). The plan
shape should stay the same since CBO is off, but I did not run it.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterPushDownSuite.scala:
##########
@@ -352,10 +352,8 @@ class StorageFilterPushDownSuite
test("a scan with a column named like the checked column gets the filter,
and runtime " +
s"filtering still resolves that column (AQE $aqe)") {
// Runtime filtering resolves the declared filter attributes by name,
which the checked column
- // the scan appends would make ambiguous. Dynamic partition pruning on
`k` would take the
- // bloom's place.
- withBloomJoin(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ // the scan appends would make ambiguous.
+ withBloomJoin(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
Review Comment:
**11. [Low] `runtimeFilterCalls > 0` no longer shows that the bloom filter
reached runtime filtering**
This is about L362-L363, which are outside the diff.
With DPP on, that assertion no longer shows that "the bloom must reach
runtime filtering". The DPP filter is planned as
`DynamicPruningExpression(true)` here. For this iterative `v2` source with
`identity(k)`, that `true` alone reaches `filter()` in the second pass, because
its empty references pass `f.references.subsetOf(filterAttrs)`
(`PushDownUtils.scala` L220 and L226).
`DataSourceV2EnhancedRuntimePartitionFilterSuite.scala` L124-L131 pins that
behavior.
So if only this V2 path lost the bloom filter, the assertion would stay
green, while with DPP off it would fail. The bloom filter does reach `filter()`
today, and only the proof is gone. Counting only predicates with non-empty
`references()` in the fixture would restore it. Copying the
`scan.runtimeFilters` assertion from the next test would show only that it
reached `runtimeFilters`.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1067,262 @@ 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") {
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
62",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
62").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+ }
+
+ 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") {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone.
+ val query = "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"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ val query = "select * from bf5filtered q join bigapp b on q.f5 = b.k " +
+ "where q.a5 > 50 and b.b5 = 1"
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilter(query)
+ val bloomFilterPlans = sql(query).queryExecution.optimizedPlan.collect
{
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ assert(!bloomFilterPlans.head.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, bloomFilterPlans.head)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+ }
+
+ 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) {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ } 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.
+ 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") {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the broadcast join builds the side
the dpp filter " +
+ "prunes") {
+ val query = "select /*+ broadcast(t) */ * from " +
+ "(select f5, sum(a5) from bf5part group by f5) t join bf2 on t.f5 =
bf2.c2 " +
+ "where bf2.a2 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter can project its
values from " +
+ "a broadcast") {
+ def query(broadcast: String): String = "select * from bf5part join " +
+ s"(select /*+ broadcast($broadcast) */ bf3.f3 from bf2 join bf3 on
bf2.c2 = bf3.c3 " +
+ "where bf3.a3 = 62) t on bf5part.f5 = t.f3"
+ val projection =
SQLConf.DYNAMIC_PARTITION_PRUNING_BROADCAST_PROJECTION_ENABLED.key -> "true"
+ withDppTestConf() {
Review Comment:
**17. [Nit] The projection-off control relies on a default**
This control for "With the projection off" in the description relies on the
default of `dynamicPartitionPruning.broadcastProjection.enabled`, while the
blocks below pin `true`. Pinning `false` here would keep its meaning if that
default changes.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
+ // rule. Without AQE, the right side of a null-aware anti join also counts.
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 {
+ case join @ ExtractEquiJoinKeys(_, _, _, _, _, left, right, _) =>
+ getBroadcastBuildSide(join, hintOnly = true, conf)
Review Comment:
**23. [Nit] `getBroadcastHashJoinBuildSide` could replace this chain**
This touches [the earlier
thread](https://github.com/apache/spark/pull/59916#discussion_r4416576984), so
here is only the new part. `getBroadcastHashJoinBuildSide` is what the arm at
L399 already calls through `canPlanAsBroadcastHashJoin`. Using it here adds no
condition to this rule and changes one thing, the `SHUFFLE_HASH` veto.
That veto is exact: it fires only when `JoinSelection` cannot reach a
size-based broadcast (`SparkStrategies.scala` L342-L346), so the list still
holds every plan that physical planning broadcasts, and
`JoinSelectionHelperSuite.scala` L189 pins it. A `SHUFFLE_HASH`-hinted join
with a shuffle below the pruned side would then get the bloom filter instead of
no runtime filter. `MERGE` and the key types stay as the limit says, and your
objection still holds for `MERGE`, since `createSortMergeJoin()` can return
`None`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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.
Physical planning drops them,
Review Comment:
**20. [Low] The reason in this comment holds only in part**
A creation plan is only Project and Filter over a leaf, so the subquery has
no broadcast, and an `onlyInBroadcast` DPP filter in it is always planned as
`true`. What it adds is planning work. The effect that matters is at L135:
`isSimpleExpression` also sees the build plan's pattern bits, so a UDF, or a
`LIKE` that `LikeSimplification` cannot rewrite, in the build plan stops
extraction.
That also shows a gap, since this strips only the key's own filters. With
`q` partitioned by `(f5, g5)`:
```sql
SELECT * FROM q JOIN b ON q.f5 = b.k AND q.g5 = b.c WHERE q.a5 > 50 AND b.s
LIKE '%x%y%'
```
both DPP filters are dropped, but each key stops at the other key's filter,
so neither gets a bloom filter, as on master. The test at
`InjectRuntimeFilterSuite.scala` L1231 has a simple build plan and does not
cover this effect. Could the comment give this reason? Stripping on
`d.onlyInBroadcast || keyDpps.contains(d)` would cover the case, at the price
of changing plans for keys without a DPP filter.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
Review Comment:
**21. [Low] The exchange-reuse-off branch could go away**
With exchange reuse off, this branch handles filters that can never run.
`PartitionPruning.insertPredicate` passes `reuseEnabled || hasBenefit`
(`PartitionPruning.scala` L118) and still sets `onlyInBroadcast` from the
config (`PartitionPruning.scala` L126), and both planners plan such a filter as
`true`.
Changing that check to `reuseEnabled ||
(!conf.dynamicPartitionPruningReuseBroadcastOnly && hasBenefit)` would stop
inserting them. A dropped DPP filter would then need reuse on, so the comment
at L471, this `if` and L476-L478 would go.
The cost is the expected value at `InjectRuntimeFilterSuite.scala` L1119,
from `Seq(Literal.TrueLiteral)` to empty, and the matching wording in the
description. The reuse-off test in `DynamicPartitionPruningSuite.scala` L1125
only looks at whether a subquery or a broadcast exists. This is optional and
could be a separate change, since it touches a file outside this PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1067,262 @@ 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") {
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
62",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
62").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
Review Comment:
**27. [Nit] The same assertion pair is copied six times**
This pair, a bloom filter added and the DPP filter planned as `true`,
appears at L1093, L1118, L1272, L1287, L1298 and L1314. A small helper next to
`dynamicPruningFilters` would shorten it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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
Review Comment:
**25. [Nit] Steps 1 and 2 of this comment restate the code**
Steps 1 and 2 restate L384-L385 and L480. Step 3 is worth keeping, because
it is the only place that explains why `goesIntoReused` has two arms, and L502
refers to it. A short comment on `reused` and on `isKept` could replace the
first two steps.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
Review Comment:
**9. [Low] With CBO on, a key can get a subquery that re-reads a large
dimension**
`onlyInBroadcast` is also what `PartitionPruning` sets when
`pruningHasBenefit` is false, the case where re-running the filtering side
costs more than the pruning gains. Counting such a filter as dropped sends the
key to `extractBeneficialFilterCreatePlan`, whose size checks look at the
filtered estimate and never at the bytes the creation side scans.
For example, with CBO on, column statistics and
`autoBroadcastJoinThreshold=-1`, take a 12 GB `fact` partitioned by `p` and a
200 GB `dim` not clustered on `id`. `WHERE dim.id = 7` is estimated at about
one row, so the bloom filter passes both size checks, and its subquery reads
`dim` again to prune at most 12 GB. On master the key is skipped. Without CBO
this cannot happen for a single table, since a `Filter` keeps its child's size.
The bloom filter also shrinks the shuffle and the join, so I cannot say from
the code that this is a net loss. Comparing the creation side's scan bytes with
the application side's, as `PartitionPruning.calculatePlanOverhead` does, would
keep that trade for these keys.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// 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. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
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) && !(hasDroppedDpp &&
goesIntoReused(l, left))) {
+ 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. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
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) &&
+ !(hasDroppedDpp && goesIntoReused(r, right))) {
+ extractBeneficialFilterCreatePlan(right, creationPlan(left), r,
l)
+ .foreach(creationSide => newRight = injectFilter(r, newRight,
creationSide))
Review Comment:
**4. [Medium] A new bloom filter can remove exchange reuse of the fact scan**
Two scans that canonicalize the same on master, because a `true` DPP filter
is removed in canonicalization (`DataSourceScanExec.scala` L1018), no longer
match when each carries a `might_contain` with a different subquery. Exchange
reuse is lost for everything under the join.
```sql
WITH t AS (SELECT k, sum(v) s FROM fact GROUP BY k)
SELECT t.k, d.x FROM t JOIN (SELECT k, x FROM dim WHERE x % 7 = 1) d ON t.k
= d.k
UNION ALL
SELECT t.k, e.x FROM t JOIN (SELECT k, x FROM dim WHERE x % 5 = 2) e ON t.k
= e.k
```
In a released Spark 4.3.0 with `autoBroadcastJoinThreshold=-1`, AQE on or
off, this plan has one `Reused` node with the default configs and none with DPP
off, which adds the same bloom filters. The result rows are the same.
Whether it costs more depends on selectivity. Bloom filters that prune
little mean up to twice the scan and shuffle plus two subquery jobs, and ones
that prune a lot can be cheaper. Could this go in the Known limits? The
`SPARK-32509` test (`DynamicPartitionPruningSuite.scala` L1572) pins this
reuse, but its tables stay under the application-side threshold, so it cannot
see the change.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1067,262 @@ 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") {
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
62",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
62").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+ }
+
+ 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") {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone.
Review Comment:
**7. [Low] A bloom filter on a key pinned by an equality prunes nothing**
Constraint inference has already put `t.c5 = 51` under the aggregate, so
every `t` row has `c5 = 51`, and this bloom filter holds only the hash of 51. A
bloom filter has no false negatives, so it removes rows only when `q` is empty,
and the kept DPP filter on `t.f5` already covers that from the same broadcast.
So this test pins a bloom filter that prunes nothing, at the cost of a
second read of `q`, a lookup per remaining row of `t`, and one of the 10
`runtimeFilter.number.threshold` slots, which adds to Known limit 5.
`CleanupDynamicPruningFilters` drops the DPP filter on such a key because it
"can't prune anything". Could a key with a dropped DPP filter skip the bloom
filter when the application side pins it, for example through
`side.constraints`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
+ // rule. Without AQE, the right side of a null-aware anti join also counts.
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 {
+ case join @ ExtractEquiJoinKeys(_, _, _, _, _, left, right, _) =>
+ getBroadcastBuildSide(join, hintOnly = true, conf)
+ .orElse(getBroadcastBuildSide(join, hintOnly = false, conf))
+ .map {
+ case BuildLeft => withoutDppFilters(left)
Review Comment:
**13. [Low] No test pins the DPP stripping on the `BuildLeft` arm**
Changing this to `case BuildLeft => left` changes no assertion in the diff.
The only test where a side with a DPP filter on top is reused is
`InjectRuntimeFilterSuite.scala` L1190, with `q` on the right, which goes
through L394. A mirror of that query with `q` on the left would pin it:
```sql
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
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// 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. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
val hasShuffle = isProbablyShuffleJoin(left, right, hint)
Review Comment:
**18. [Low] The fix reaches fewer queries than the description suggests**
This is about the description, not the heuristics, which this PR does not
change. These keys still end with no runtime filter, as on master, and the
Known limits do not say so:
- **Default thresholds.** `autoBroadcastJoinThreshold` and
`creationSideThreshold` are both 10MB. For `fact JOIN dim`, a `dim` of at most
10MB is broadcast and the DPP filter is kept, and a larger one is skipped at
L70-L73 by the same size. So the shuffle join and storage-partitioned join
examples in the description need a lower broadcast threshold, a creation side
that is a smaller part of the join side, a join type or hint that rules out the
broadcast, or a materialized cache. Every new test moves the threshold.
- **Join type.** `isProbablyShuffleJoin` ignores it. `select * from bf5part
right join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 = 62` at threshold 5000
plans a sort merge join with a `true` DPP filter and no bloom filter, and the
`left join` and `left semi join` mirrors do the same. The test at
`InjectRuntimeFilterSuite.scala` L1284 passes because of its aggregate, while
its name and "a right outer join whose right side is small enough to broadcast
gets the bloom filter" in the description credit the join type.
- **Statistics.** A partitioned catalog table that was never analyzed
reports `defaultSizeInBytes`, which L283 counts as 0, so it fails the
application-side check.
Could the Known limits list these, and the description's bullet mention the
shuffle below the join?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
Review Comment:
**12. [Low] No test pins the second half of the union check**
Removing `|| r.exists(_.fastEquals(side))` changes no assertion in the diff.
It needs a side with a union inside a reused plan, and the only test that
reaches this arm (`InjectRuntimeFilterSuite.scala` L1170) has `d` inside the
side. Making `q` a union in the query at `InjectRuntimeFilterSuite.scala` L1148
would pin it:
```sql
SELECT /*+ broadcast(r) */ * FROM bf5part p JOIN (SELECT q.k rf, q.v ra FROM
(SELECT f5 k, a5 v FROM bf5filtered UNION ALL SELECT f5, a5 FROM
bf5filtered) q
JOIN bf2 ON q.k = bf2.c2 WHERE bf2.a2 = 62) r ON p.f5 = r.rf
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
+ // rule. Without AQE, the right side of a null-aware anti join also counts.
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 {
+ case join @ ExtractEquiJoinKeys(_, _, _, _, _, left, right, _) =>
+ getBroadcastBuildSide(join, hintOnly = true, conf)
+ .orElse(getBroadcastBuildSide(join, hintOnly = false, conf))
+ .map {
+ case BuildLeft => withoutDppFilters(left)
+ case BuildRight => withoutDppFilters(right)
Review Comment:
**19. [Low] Known limit 1 is narrower than what the code counts**
Known limit 1 and the comment at L384-L385 describe a plan that physical
planning does not broadcast. In these two shapes it does broadcast, and the
filter is still `true`:
- **A broadcast side that keeps its own DPP filter.** `withoutDppFilters`
assumes that cleanup may drop the filter. Remove `and q.f5 = 51` from the query
at `InjectRuntimeFilterSuite.scala` L1190 and the filter on `q.f5` survives
cleanup. `isKept` is true for `t.f5`, but the scan of `q` has a DPP filter that
the build plan lacks, so `sameResult` is false in both planners.
- **AQE with the same plan on other keys.** `reused` compares plans, while
`PlanAdaptiveDynamicPruningFilters.scala` L59 compares the exchange, including
its hash mode. If another join broadcasts the same plan on a different key, the
filter is `true`, and AQE is the default. The null-aware anti join arm at
L398-L399 makes this distinction for the same reason, and this arm does not.
The outcome is master's in both, so this is harmless, but these keys get no
runtime filter and the limit reads narrower than it is.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ 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, right, l, 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.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// 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. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
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) && !(hasDroppedDpp &&
goesIntoReused(l, left))) {
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
Review Comment:
**14. [Low] No test pins `creationPlan` on the left application side**
With plain `right` here, every assertion in the diff still passes. The test
at `InjectRuntimeFilterSuite.scala` L1231 has `q` on the left, which goes
through L531. The query at `InjectRuntimeFilterSuite.scala` L1190 does strip a
DPP filter here, but nothing looks at its bloom filter's subquery. A mirror of
`InjectRuntimeFilterSuite.scala` L1238, with `bigapp b JOIN bf5filtered q`,
would pin it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +302,39 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
checkWithAndWithoutFeatureEnabled(query, shouldReplace = false)
}
+ // Runs `f` with the thresholds of the DPP tests and `confs`, with AQE on
and off unless `confs`
+ // sets it.
+ def withDppTestConf(confs: (String, String)*)(f: => Unit): Unit = {
+ val adaptiveModes = confs.collectFirst {
+ case (SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, adaptive) => Seq(adaptive)
+ }.getOrElse(Seq("true", "false"))
+ adaptiveModes.foreach { adaptive =>
+ withSQLConf(Seq(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000",
+ 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
Review Comment:
**16. [Low] `dynamicPruningFilters` reports a class, which does not show
reuse**
The class shows that physical planning found a broadcast to reuse, but not
that the exchange is reused. `SubqueryBroadcastExec` is also what the
null-aware anti join test expects at L1276, for a filter whose comment says it
builds its own broadcast. So in the tests that say the filter reuses a
broadcast, a regression where it builds its own would pass with AQE off.
`DynamicPartitionPruningSuite.scala` L194-L252 checks the real reuse through
`ReusedExchangeExec`, but it is on an abstract suite, so using it takes moving
it to a shared trait. The rule predicts kept filters, not reuse, so this does
not hide a wrong decision by the rule. It only leaves reuse on these query
shapes unpinned.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1067,262 @@ 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") {
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
62",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
62").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ }
+ }
+
+ 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") {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone.
+ val query = "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"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ val query = "select * from bf5filtered q join bigapp b on q.f5 = b.k " +
+ "where q.a5 > 50 and b.b5 = 1"
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilter(query)
+ val bloomFilterPlans = sql(query).queryExecution.optimizedPlan.collect
{
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
Review Comment:
**15. [Low] This test does not check that `q` is the creation side**
`b.b5 = 1` is selective as well, so `b` could also be the creation side, and
only the size check set at L1241 decides. If the choice of side changed so that
`q` took the bloom filter, these assertions would all still pass without
reaching `creationPlan`. Asserting that the leaf of `bloomFilterPlans.head`
reads `bf5filtered` would close this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ 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. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 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.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
Review Comment:
**22. [Low] A helper that returns all lineages could replace the union arm**
This arm exists because `findExpressionAndTrackLineageDown` keeps only the
first union branch that has a lineage (`predicates.scala` L137-L139). A sibling
helper that returns all lineages, with the existing method as its `headOption`,
would make this one check: any lineage leaf inside a plan in `reused`. Every
branch is already walked before `headOption`, so this adds no traversal, and
existing callers do not change.
It would also narrow Known limit 2. With the query at
`InjectRuntimeFilterSuite.scala` L1170 keyed on `q.f5`, the bloom filter
reaches only `p` and `q`, but this arm skips it because `d` is in the side. The
one thing to handle is a branch with no lineage, since pushdown still sends it
a copy of the filter: keep the current check for that branch, or accept that
corner.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -11093,6 +11093,42 @@ class KeyGroupedPartitioningCatalystRuntimeFilterSuite
after {
catalog.clearTables()
+ catalog.clearFunctions()
+ }
+
+ test("SPARK-60702: a storage-partitioned join gets the bloom filter that its
dropped dpp " +
+ "filter used to keep out") {
+ // The scan offers the source column of `bucket(4, id)` for runtime
filtering, so
+ // `PartitionPruning` adds a DPP filter on `f.id`. Nothing can broadcast
its build plan, so
+ // physical planning plans it as `true`. The key gets a bloom filter
instead, and the join
+ // still shuffles neither side.
+ catalog.createFunction(Identifier.of(Array.empty,
UnboundBucketFunction.name()),
+ UnboundBucketFunction)
+ createTable("fact", Array(Column.create("id", LongType),
Column.create("v", IntegerType)),
+ Array(bucket(4, "id")))
+ sql("INSERT INTO testcat.ns.fact VALUES " +
+ (0 until 16).map(i => s"($i, ${i * 10})").mkString(", "))
+ createTable("dim", Array(Column.create("id", LongType),
Column.create("flag", IntegerType)),
+ Array(bucket(4, "id")))
+ sql("INSERT INTO testcat.ns.dim VALUES (1, 1), (2, 0), (3, 1), (5, 1), (6,
0), (9, 1)")
+ Seq(true, false).foreach { adaptive =>
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ENABLED.key -> "true",
Review Comment:
**26. [Nit] `V2_BUCKETING_ENABLED` restates the default**
This is already the default (`SQLConf.scala` L2661), and sibling tests in
this trait omit it, so the line can go.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
+ // rule. Without AQE, the right side of a null-aware anti join also counts.
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 {
+ case join @ ExtractEquiJoinKeys(_, _, _, _, _, left, right, _) =>
+ getBroadcastBuildSide(join, hintOnly = true, conf)
+ .orElse(getBroadcastBuildSide(join, hintOnly = false, conf))
+ .map {
+ case BuildLeft => withoutDppFilters(left)
+ case BuildRight => withoutDppFilters(right)
+ }
+ // Without AQE, physical planning keeps a DPP filter whose build plan this
join broadcasts,
+ // though the filter builds its own broadcast. AQE plans such a filter as
`true`.
+ case join @ ExtractSingleColumnNullAwareAntiJoin(_, _)
+ if !conf.adaptiveExecutionEnabled && canPlanAsBroadcastHashJoin(join,
conf) =>
+ Seq(withoutDppFilters(join.right))
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // The plans whose broadcast a DPP filter may reuse: its build plan, and for
an `onlyInBroadcast`
+ // filter, the source plan of its broadcast value projection.
+ private def reusablePlans(dpp: DynamicPruningSubquery): Seq[LogicalPlan] =
+ dpp.buildQuery +: dpp.usableBroadcastValueProjection.filter(_ =>
dpp.onlyInBroadcast)
+ .map(_.sourcePlan).toSeq
+
+ // The DPP filters on the key, which sit right on top of the join sides, as
this rule is called
+ // just after DPP.
@tailrec
- private def hasDynamicPruningSubquery(
+ private def dynamicPruningSubqueries(
Review Comment:
**24. [Nit] `dynamicPruningSubqueries` no longer needs the two-plan
recursion**
With the `||` short-circuit gone, the accumulator and the two-plan recursion
no longer buy anything. The two arms never interleave (the left chain, then the
right chain), so a per-side walker gives the same list:
```scala
private def dynamicPruningSubqueries(
side: LogicalPlan,
key: Expression): Seq[DynamicPruningSubquery] = side match {
case Filter(dpp: DynamicPruningSubquery, child) =>
val rest = dynamicPruningSubqueries(child, key)
if (dpp.pruningKey.fastEquals(key)) dpp +: rest else rest
case _ => Nil
}
```
The caller at L494 would use `dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)`, and `@tailrec` and its import would go.
--
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]
