dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4473894166
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
Review Comment:
**1. [Medium] `cannotFail` checks a cast against its already-rewritten
child**
`transformUp` rewrites the children first and then applies the rule to the
rebuilt node (`TreeNode.scala` L633-L635). So when this guard reads
`c.child.dataType`, an inner up-cast is already gone. That causes two problems
that becfca3's `isSafeKey` did not have, since it checked each cast against its
own child and never rebuilt the tree:
- **The optimizer throws.** Type coercion puts an up-cast into a `Coalesce`,
`If`, `CaseWhen`, `Greatest` or `Least` (`TypeCoercionHelper.scala` L576-L580).
Once this rule strips it, that node is rebuilt with children of different
types, and the guard of the cast above it calls its `dataType`, which runs
`ComplexTypeMergingExpression.dataTypeCheck` (`Expression.scala` L1374-L1376).
With the suite's DPP confs:
```scala
withDppTestConf() {
sql("select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
"where cast(coalesce(bf2.a2, 0L) as int) = 36").collect()
}
```
The filter becomes `cast(coalesce(cast(a2 as bigint), 0L) as int) = 36`.
The DPP filter on `f5` is dropped, as in the coalesce test at
`InjectRuntimeFilterSuite.scala` L1453, so L438 calls `cannotFail` on that
conjunct. It throws `SparkIllegalArgumentException` with
`COMPLEX_EXPRESSION_UNSUPPORTED_INPUT.MISMATCHED_TYPES`, nothing catches it,
and the query fails to optimize. Master and becfca3 run it. An implicit cast
does the same, e.g. `coalesce(bf2.a2, 0L) > 35.5D`, where
`UnwrapCastInBinaryComparison` keeps the cast to `DOUBLE`.
- **A cast that can fail passes.** In `cast(cast(p as float) as int)`, the
inner INT to FLOAT is an up-cast (`UpCastRule.scala` L38) and is stripped. The
outer cast is then checked as INT to INT (`UpCastRule.scala` L34) and stripped
too, so the key passes `canPushDown`. As a partition key with a dropped DPP
filter, it gets a bloom filter that becomes a partition filter. On a partition
with `p` in `[2147483584, 2147483647]`, the FLOAT to INT cast then throws
`CAST_OVERFLOW` under ANSI (`numerics.scala` L134-L138), even if every row of
that partition fails the data filters. `SimplifyCasts` keeps such a nested cast.
Rewriting top-down, and checking each cast against its own child before
anything is replaced, avoids both. For example, with `isNumericUpCast` as the
guard at L418-L420:
```scala
case _ => ExprUtils.canEvaluateUnconditionally(e.transformDown {
case c: Cast if isNumericUpCast(c) && cannotFail(c.child) =>
AttributeReference("upCast", c.dataType, c.nullable)()
case n @ KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) if
cannotFail(child) =>
AttributeReference("normalized", n.dataType, n.nullable)()
})
```
The stand-in keeps the cast's type, so no parent sees a child of another
type. Both queries above would make good tests. I traced this from the code and
did not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ c.child
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) => child
+ })
+ }
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if scanning
it again gives the
+ // same rows. Its filters, and the projections that the key and the filters
need, must pass
+ // `cannotFail`, which takes only expressions that are deterministic in the
subquery and do not
+ // fail there. A part of the join side outside that plan can only drop rows
from the join, so the
+ // bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ !evaluated.containsPattern(CTE) && !evaluated.exists {
+ case Project(projectList, _) => !projectList.forall(e =>
cannotFail(trimAliases(e)))
+ case Filter(condition, _) =>
!splitConjunctivePredicates(condition).forall(cannotFail)
Review Comment:
**2. [Low] A filter on `_metadata.file_block_start` passes
`canEvaluateAgain`**
When this rule runs, `_metadata` is one struct attribute, so
`c._metadata.file_block_start` is a `GetStructField` over an
`AttributeReference`, which `canEvaluateUnconditionally` accepts
(`ExprUtils.scala` L270-L272). But its value is the start of the split that a
row is read from (`FileFormat.scala` L372-L373), and the bloom filter's
subquery can split the files differently from the join:
- A bucketed scan reads each file as one split, so every row has
`file_block_start = 0` in the join. In the subquery nothing needs the
bucketing, so `DisableUnnecessaryBucketedScan` turns the bucketed scan off
(`DisableUnnecessaryBucketedScan.scala` L100-L105), and the files are split by
`maxSplitBytes`.
- A kept DPP filter of another key prunes only the join's scan, and
`maxSplitBytes` follows the bytes of the selected partitions
(`DataSourceScanExec.scala` L966-L969).
For example, with `f(v, k)` partitioned by `k`, `c(k, x)` bucketed by `k`
into 2 buckets, broadcasts off, the application-side threshold at 1 and
`spark.sql.files.maxPartitionBytes=1024`:
```sql
SELECT f.v, c.x FROM f JOIN c ON f.k = c.k WHERE
c._metadata.file_block_start = 0
```
The subquery drops the rows of every later split, so their keys are missing
from the bloom filter on `f.k`, which then prunes those partitions of `f`. The
query returns fewer rows than with the bloom filter off. On master the DPP
filter keeps the bloom filter off this key, and becfca3's `isSafeKey` turned
away `GetStructField`. Keys without a DPP filter have the same exposure on
master, so only the part for keys with a dropped DPP filter is new.
This came in with the delegation that I suggested in [round-4 item
8](https://github.com/apache/spark/pull/59916#discussion_r4470999415). Spark
marks `input_file_block_start()` as `Nondeterministic` (`inputFileBlock.scala`
L71), and this field is its deterministic-looking twin. Turning away a
reference to a file metadata column (`FileSourceMetadataAttribute`) in
`canEvaluateAgain`, or at least the `file_block_start` and `file_block_length`
fields, would close it. The filter is unusual, so I marked it Low, but the
result is wrong rows. I traced this from the code and did not run it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1093,627 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the creation side that the
subquery reads is not " +
+ "deterministic") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the key of the dropped dpp
filter can fail") {
+ // On a partition column, the bloom filter would become a partition
filter, which evaluates
+ // the key on every partition before the data filters (SPARK-60809). An
alias hides the key only
+ // at the join, since pushdown rewrites the bloom filter through it. And
`coalesce` cannot be
+ // null, so no inferred `isnotnull` evaluates the key on every partition
first.
+ // Through a union, every branch counts, and without DPP each gets a bloom
filter. A narrowing
+ // cast can fail too.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36" -> 1,
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36" -> 1,
+ "select * from (select f5 k, a5 v from bf5part union all " +
+ "select coalesce(f5 + 1, 0) k, a5 v from bf5filtered) u " +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36" -> 2,
+ "select * from bf5part join bf2 on cast(bf5part.f5 as tinyint) = bf2.c2
" +
+ "where bf2.a2 = 36" -> 1
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assertDidNotRewriteWithBloomFilter(query)
+ val filters = dynamicPruningFilters(query)
+ assert(filters.nonEmpty && filters.forall(_ == Literal.TrueLiteral),
filters)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter in an in subquery whose dpp
filter reuses a " +
+ "broadcast outside it") {
+ // The optimizer runs on the subquery on its own, where the broadcast of
`x` is not visible.
+ // The DPP filter on `p.f5` reuses it once the subquery is joined with the
rest of the query.
+ val query = "select /*+ broadcast(x) */ * from bf3 join (select c2 from
bf2 where a2 = 62) x " +
+ "on bf3.c3 = x.c2 where bf3.a3 in (select p.a5 from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) d on p.f5 = d.c2)"
+ def keysOnF5(query: String): Seq[Expression] =
+ bloomFilterKeys(query).filter(_.references.exists(_.name == "f5"))
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(keysOnF5(query).isEmpty)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ // Without the hint, nothing broadcasts `x`, and `p.f5` gets the bloom
filter.
+ val withoutHint = query.replace("/*+ broadcast(x) */ ", "")
+ assert(keysOnF5(withoutHint).nonEmpty)
+ assert(dynamicPruningFilters(withoutHint) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter
under a numeric " +
+ "up-cast") {
+ // Type coercion casts `f5` to `bigint`, which cannot fail.
+ val query = "select * from bf5part join (select cast(c2 as bigint) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter of a
floating point " +
+ "type") {
+ // `NormalizeFloatingNumbers` wraps the `double` join keys, which cannot
fail.
+ val query = "select * from bf5part join (select cast(c2 as double) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation filter has a coalesce,
which cannot fail") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where coalesce(bf2.a2, 0) = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter under a " +
+ "decimal to bigint cast") {
+ // `Cast.canUpCast` takes `DECIMAL(19,0)` to `BIGINT`, but
`9999999999999999999` overflows. The
+ // query casts it only on rows with `v > 0`, which that partition does not
have.
+ withTable("fd") {
+ sql("create table fd (v int, d decimal(19,0)) using parquet partitioned
by (d)")
+ sql("insert into fd select cast(id as int) + 1, 36 from range(100)")
+ sql("insert into fd select -cast(id as int), cast('9999999999999999999'
as decimal(19,0)) " +
+ "from range(100)")
+ val query = "select * from fd join bf2 on cast(fd.d as bigint) = bf2.c2
" +
+ "where fd.v > 0 and bf2.a2 = 36"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose key can fail in a union
branch the lineage " +
+ "does not reach") {
+ // The lineage of `u.k` stops at the inner union of the second branch, but
pushdown moves the
+ // bloom filter into both inner branches, where it would cast `p` on every
partition. Each
+ // row of `p = 'abc'` has `v <= 0`, so the query itself never casts
`'abc'`.
+ withTable("sp") {
+ sql("create table sp (v int, p string) using parquet partitioned by (p)")
+ sql("insert into sp select cast(id as int) + 1, '36' from range(100)")
+ sql("insert into sp select -cast(id as int), 'abc' from range(100)")
+ val query = "select * from (select f5 as k, a5 as v from bf5part union
all " +
+ "select coalesce(cast(x as int), 0) as k, sum(v) as v from (select p
as x, v from sp " +
+ "where v > 0 union all select p, v from sp where v > 0) group by x) u
" +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36"
+ withDppTestConf() {
+ // Without DPP, pushdown puts a bloom filter into each of the three
branches.
+ assertBloomFilterWithoutDpp(query, 3)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation side has a
non-deterministic column that " +
+ "the bloom filter does not read") {
+ // The bloom filter's subquery reads only `c2` of `d`, so `src` does not
change its keys. The
+ // DPP filter is not deterministic, so cleanup removes it.
+ val query = "select * from bf5part p join (select c2, input_file_name()
src from bf2 " +
+ "where a2 = 36) d on p.f5 = d.c2"
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assert(bloomFilterKeyNames(query) == Seq("f5"))
+ val expected = withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key ->
"false") {
+ sql(query).collect()
+ }
+ checkAnswer(sql(query), expected)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side reads a cache
that can give " +
Review Comment:
**3. [Low] The cache test does not pin the repeatability half of
`isOutputRepeatable`**
`bf2` is a parquet scan, so once its cache is built,
`isCachedPlanRepeatable` holds. This test therefore only tells a cache that is
not materialized yet from a materialized one. Changing L439 of
`InjectRuntimeFilter.scala` to `case leaf: MaterializedLeafNode =>
leaf.materializedMetadata.isEmpty` passes every test, yet that version accepts
the cache of [round-4 item
1](https://github.com/apache/spark/pull/59916#discussion_r4470999327) once
`count()` materializes it. No other test reaches that arm with a dropped DPP
filter: the SPARK-58272 tests join `bf1`, which is not partitioned.
A plan check after materializing a cache of a non-repeatable plan would pin
it, and it needs no partition to be computed twice:
```scala
spark.table("bf2").where("cast(reflect('java.lang.Math', 'random') as
double) < 2.0")
.createOrReplaceTempView("bf2r")
spark.catalog.cacheTable("bf2r")
spark.table("bf2r").count()
```
`reflect` is not deterministic, so `isCachedPlanRepeatable` stays false
(`InMemoryRelation.scala` L505), and the query of this test over `bf2r` could
assert `assertBloomFilterWithoutDpp(query)` and
`bloomFilterKeys(query).isEmpty`. The comment at L1518-L1519 and the test name
could then say what each case shows: `bf2` is turned away only while Spark
cannot tell whether a second scan gives the same rows, and it would give the
same rows even if a partition were computed twice.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
Review Comment:
**4. [Low] No test pins the accepting half of the new DPP arm**
Changing this line to `case _: DynamicPruningSubquery => false` passes every
test. Three tests have another key's DPP filter in a creation side, but none
lets this arm decide:
- The `ra2`/`rc2` query at `InjectRuntimeFilterSuite.scala` L1673 runs with
`reuseBroadcastOnly` off, so its filter on `c.q` is not `onlyInBroadcast`.
- The union query at `InjectRuntimeFilterSuite.scala` L1628 is turned away
first by `cast(c.s as int)` in the root projection.
- In the `sf`/`sc` query at `InjectRuntimeFilterSuite.scala` L1647, the
filter on `c.q` sits in the creation side for `f.y`, which has no DPP filter,
so `keyDpps.isEmpty` decides.
With the default `reuseBroadcastOnly`, the `ra2`/`rc2` query drops both of
its DPP filters. The creation side of `f.a` keeps the filter on `c.q`, which is
`true` in the subquery, so head adds bloom filters on `a` and `b`, and the
mutant moves the first one to `t`. Running that query with the default as well,
and asserting `bloomFilterKeyNames(query).toSet == Set("a", "b")` and the
answer against bloom filters off, would pin this arm. It would also execute a
bloom filter whose subquery holds a DPP filter, which no test does now.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
Review Comment:
**5. [Low] No test pins the `NumericType` conjuncts of the cast arm**
Dropping `c.child.dataType.isInstanceOf[NumericType] &&
c.dataType.isInstanceOf[NumericType] &&` passes every test: the casts in the
tests are all between numeric types or from strings to numbers, which
`canUpCast` decides by itself. Yet these two conjuncts are what keep a DATE to
TIMESTAMP key out, as [round-2 item
4](https://github.com/apache/spark/pull/59916#discussion_r4448448582) asked.
`canUpCast` takes DATE to the timestamp types (`UpCastRule.scala` L41), and
`daysToMicros` throws `ArithmeticException` past the year 294247
(`SparkDateTimeUtils.scala` L372-L378).
So without them, `fd(v int, d date)` partitioned by `d` and joined on `fd.d
= dim.ts` with a TIMESTAMP `ts` gets the key `cast(d as timestamp)`, and a
bloom filter that runs as a partition filter. A partition `d = DATE
'300000-01-01'` whose rows all fail `fd.v > 0` then fails the query. A plan
test with such a join, asserting `assertBloomFilterWithoutDpp(query)` and
`bloomFilterKeys(query).isEmpty`, needs no such date.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ c.child
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) => child
+ })
+ }
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if scanning
it again gives the
+ // same rows. Its filters, and the projections that the key and the filters
need, must pass
+ // `cannotFail`, which takes only expressions that are deterministic in the
subquery and do not
+ // fail there. A part of the join side outside that plan can only drop rows
from the join, so the
+ // bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
Review Comment:
**9. [Low] The subquery could drop the DPP filters that are always `true`
there**
L416 predicts that an `onlyInBroadcast` DPP filter in the bloom filter's
subquery finds no broadcast and becomes `true`. That holds, but physical
planning first plans the filter's build plan: `PlanDynamicPruningFilters` calls
`createSparkPlan` when exchange reuse is on (`PlanDynamicPruningFilters.scala`
L56-L63), and under AQE `InsertAdaptiveSparkPlan` compiles it as a subquery of
its own (`InsertAdaptiveSparkPlan.scala` L152). Master pays that too, whenever
a creation side keeps another key's DPP filter. Dropping such filters when
`injectBloomFilter` builds the subquery would replace the prediction at L416
and save that planning. It would have to run after `extract`, whose search
would otherwise see other plans for keys without a DPP filter than on master,
and it would have to drop the filters further down the chain and inside a
larger filter (L499-L500), not only those on top.
Also, `canEvaluateAgain` checks `ColumnPruning(Project(...))` at L434-L435,
a model of the plan that `injectBloomFilter` builds at L83-L84. I found no plan
where the two differ, but checking that `aggregate` itself would keep them from
drifting apart. [Round-2 item
5](https://github.com/apache/spark/pull/59916#discussion_r4448448590) asked for
this and was declined, partly because it would miss the CTE case, which the
check at L436 now covers.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1093,627 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
Review Comment:
**12. [Nit] Three DPP-off controls still run their queries**
This control and the ones at L1184-L1186 and L1316-L1318 call
`assertRewroteWithBloomFilter`, which runs each query three times: at L244, and
twice in `checkAnswer` at L249. That is 30 runs in all. With DPP off they only
exercise master's path, and the comment at L1131-L1132 says that they show the
bloom filter that the join would otherwise get.
`assertBloomFilterWithoutDpp(query)` shows that from one optimized plan, as the
other 15 controls do.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1093,627 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the creation side that the
subquery reads is not " +
+ "deterministic") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the key of the dropped dpp
filter can fail") {
+ // On a partition column, the bloom filter would become a partition
filter, which evaluates
+ // the key on every partition before the data filters (SPARK-60809). An
alias hides the key only
+ // at the join, since pushdown rewrites the bloom filter through it. And
`coalesce` cannot be
+ // null, so no inferred `isnotnull` evaluates the key on every partition
first.
+ // Through a union, every branch counts, and without DPP each gets a bloom
filter. A narrowing
+ // cast can fail too.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36" -> 1,
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36" -> 1,
+ "select * from (select f5 k, a5 v from bf5part union all " +
+ "select coalesce(f5 + 1, 0) k, a5 v from bf5filtered) u " +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36" -> 2,
+ "select * from bf5part join bf2 on cast(bf5part.f5 as tinyint) = bf2.c2
" +
+ "where bf2.a2 = 36" -> 1
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assertDidNotRewriteWithBloomFilter(query)
+ val filters = dynamicPruningFilters(query)
+ assert(filters.nonEmpty && filters.forall(_ == Literal.TrueLiteral),
filters)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter in an in subquery whose dpp
filter reuses a " +
+ "broadcast outside it") {
+ // The optimizer runs on the subquery on its own, where the broadcast of
`x` is not visible.
+ // The DPP filter on `p.f5` reuses it once the subquery is joined with the
rest of the query.
+ val query = "select /*+ broadcast(x) */ * from bf3 join (select c2 from
bf2 where a2 = 62) x " +
+ "on bf3.c3 = x.c2 where bf3.a3 in (select p.a5 from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) d on p.f5 = d.c2)"
+ def keysOnF5(query: String): Seq[Expression] =
+ bloomFilterKeys(query).filter(_.references.exists(_.name == "f5"))
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(keysOnF5(query).isEmpty)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ // Without the hint, nothing broadcasts `x`, and `p.f5` gets the bloom
filter.
+ val withoutHint = query.replace("/*+ broadcast(x) */ ", "")
+ assert(keysOnF5(withoutHint).nonEmpty)
+ assert(dynamicPruningFilters(withoutHint) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter
under a numeric " +
+ "up-cast") {
+ // Type coercion casts `f5` to `bigint`, which cannot fail.
+ val query = "select * from bf5part join (select cast(c2 as bigint) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter of a
floating point " +
+ "type") {
+ // `NormalizeFloatingNumbers` wraps the `double` join keys, which cannot
fail.
+ val query = "select * from bf5part join (select cast(c2 as double) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation filter has a coalesce,
which cannot fail") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where coalesce(bf2.a2, 0) = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter under a " +
+ "decimal to bigint cast") {
+ // `Cast.canUpCast` takes `DECIMAL(19,0)` to `BIGINT`, but
`9999999999999999999` overflows. The
+ // query casts it only on rows with `v > 0`, which that partition does not
have.
+ withTable("fd") {
+ sql("create table fd (v int, d decimal(19,0)) using parquet partitioned
by (d)")
+ sql("insert into fd select cast(id as int) + 1, 36 from range(100)")
+ sql("insert into fd select -cast(id as int), cast('9999999999999999999'
as decimal(19,0)) " +
+ "from range(100)")
+ val query = "select * from fd join bf2 on cast(fd.d as bigint) = bf2.c2
" +
+ "where fd.v > 0 and bf2.a2 = 36"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose key can fail in a union
branch the lineage " +
+ "does not reach") {
+ // The lineage of `u.k` stops at the inner union of the second branch, but
pushdown moves the
+ // bloom filter into both inner branches, where it would cast `p` on every
partition. Each
+ // row of `p = 'abc'` has `v <= 0`, so the query itself never casts
`'abc'`.
+ withTable("sp") {
+ sql("create table sp (v int, p string) using parquet partitioned by (p)")
+ sql("insert into sp select cast(id as int) + 1, '36' from range(100)")
+ sql("insert into sp select -cast(id as int), 'abc' from range(100)")
+ val query = "select * from (select f5 as k, a5 as v from bf5part union
all " +
+ "select coalesce(cast(x as int), 0) as k, sum(v) as v from (select p
as x, v from sp " +
+ "where v > 0 union all select p, v from sp where v > 0) group by x) u
" +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36"
+ withDppTestConf() {
+ // Without DPP, pushdown puts a bloom filter into each of the three
branches.
+ assertBloomFilterWithoutDpp(query, 3)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation side has a
non-deterministic column that " +
+ "the bloom filter does not read") {
+ // The bloom filter's subquery reads only `c2` of `d`, so `src` does not
change its keys. The
+ // DPP filter is not deterministic, so cleanup removes it.
+ val query = "select * from bf5part p join (select c2, input_file_name()
src from bf2 " +
+ "where a2 = 36) d on p.f5 = d.c2"
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
Review Comment:
**6. [Nit] Round-4 item 7's AQE setting went to another test**
[Round-4 item
7](https://github.com/apache/spark/pull/59916#discussion_r4470999407) was on
the test of a non-deterministic pruned side (becfca3 L1374) and on the CTE
test, which only check `optimizedPlan`. a90e1fb set AQE off in the CTE test
(L1578) and here, in a test that runs `collect()` and `checkAnswer`
(L1509-L1512). So the pruned-side test at L1379-L1383 still checks the same
plan twice, and this test lost its AQE-on answer check, the only one with a
non-deterministic column in a creation side. Moving
`SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false"` from here to L1379, and the
"with AQE off" in the description with it, would match the reply.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,19 +473,85 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. An uncorrelated IN subquery is joined with the
rest of the query
+ // later, where `PartitionPruning` puts a new DPP filter on the key,
and this rule decides
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if `creationSide`
+ // allows it: pushdown can take it down safely (`canPushDown`), and the
subquery can evaluate
+ // the creation side again (`canEvaluateAgain`). Pushdown moves the
bloom filter along its
+ // key, down to the leaves the key comes from, and into every branch of
a union. There it
+ // must not land in a plan in `reused`. It would change the broadcast,
and physical planning
+ // would drop the DPP filter that reuses it. A reused broadcast
elsewhere in the join side
+ // does not matter, such as a broadcast dimension joined to the same
fact table. The key must
+ // also be safe at every leaf (`cannotFail`). On a partition column,
the bloom filter becomes
+ // a partition filter, which evaluates the key on every partition
before the data filters. So
+ // a key that can fail could fail the query (SPARK-60809).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ cannotFail(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
Review Comment:
**10. [Nit] A probe that keeps a row when its key fails would cover
SPARK-60809 for every key**
`cannotFail(leafKey)` here keeps a key that can fail off a partition filter,
which evaluates it on partitions that the join never reads (SPARK-60809). If
`BloomFilterMightContain` treated an error from its value expression as "might
contain", the probe would stay exact: a row that it keeps either reaches the
join, which evaluates the same key and raises the same error, or is dropped
before. DSv2 runtime partition predicates already keep a partition that they
fail to evaluate (`PushDownUtils.scala` L495-L499). That would also fix the
same failure for keys without a DPP filter on master.
It would not make this rule much smaller, though. `canEvaluateAgain` still
needs `cannotFail` for the subquery, the `reused` check on this line still
needs the strict walk, and codegen would have to keep the value out of
subexpression elimination for the catch to see its error. Since it changes
every bloom filter, a separate JIRA issue would fit better than this PR.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1093,627 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the creation side that the
subquery reads is not " +
+ "deterministic") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the key of the dropped dpp
filter can fail") {
+ // On a partition column, the bloom filter would become a partition
filter, which evaluates
+ // the key on every partition before the data filters (SPARK-60809). An
alias hides the key only
+ // at the join, since pushdown rewrites the bloom filter through it. And
`coalesce` cannot be
+ // null, so no inferred `isnotnull` evaluates the key on every partition
first.
+ // Through a union, every branch counts, and without DPP each gets a bloom
filter. A narrowing
+ // cast can fail too.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36" -> 1,
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36" -> 1,
+ "select * from (select f5 k, a5 v from bf5part union all " +
+ "select coalesce(f5 + 1, 0) k, a5 v from bf5filtered) u " +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36" -> 2,
+ "select * from bf5part join bf2 on cast(bf5part.f5 as tinyint) = bf2.c2
" +
+ "where bf2.a2 = 36" -> 1
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assertDidNotRewriteWithBloomFilter(query)
+ val filters = dynamicPruningFilters(query)
+ assert(filters.nonEmpty && filters.forall(_ == Literal.TrueLiteral),
filters)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter in an in subquery whose dpp
filter reuses a " +
+ "broadcast outside it") {
+ // The optimizer runs on the subquery on its own, where the broadcast of
`x` is not visible.
+ // The DPP filter on `p.f5` reuses it once the subquery is joined with the
rest of the query.
+ val query = "select /*+ broadcast(x) */ * from bf3 join (select c2 from
bf2 where a2 = 62) x " +
+ "on bf3.c3 = x.c2 where bf3.a3 in (select p.a5 from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) d on p.f5 = d.c2)"
+ def keysOnF5(query: String): Seq[Expression] =
Review Comment:
**13. [Nit] `keysOnF5` is now `bloomFilterKeyNames`**
[Round-4 item
12](https://github.com/apache/spark/pull/59916#discussion_r4470999462) named
this helper too. `keysOnF5(q).isEmpty` is
`!bloomFilterKeyNames(q).contains("f5")`, so L1424 and L1428 could use that,
and L1421-L1422 could go.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ c.child
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) => child
+ })
+ }
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if scanning
it again gives the
+ // same rows. Its filters, and the projections that the key and the filters
need, must pass
+ // `cannotFail`, which takes only expressions that are deterministic in the
subquery and do not
+ // fail there. A part of the join side outside that plan can only drop rows
from the join, so the
+ // bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ !evaluated.containsPattern(CTE) && !evaluated.exists {
Review Comment:
**14. [Nit] The CTE check can be one more arm of the leaf match**
The check is needed, since `CTERelationRef` is a `LeafNode`, but `case _:
CTERelationRef => true` before L440 would do the same as one more arm of this
match, and the `CTE` import could go. The two differ in one case. A DPP
filter's pattern bits include those of its build plan
(`PlanExpression.treePatternBits` in catalyst's `expressions/subquery.scala`),
so another key's `onlyInBroadcast` DPP filter whose build plan reads a CTE that
`InlineCTE` keeps, such as a `MATERIALIZED` one, trips L436, though L416 treats
that filter as `true`. That only loses a bloom filter, like the case of
[round-2 item
6](https://github.com/apache/spark/pull/59916#discussion_r4448448593).
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1093,627 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the creation side that the
subquery reads is not " +
+ "deterministic") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the key of the dropped dpp
filter can fail") {
+ // On a partition column, the bloom filter would become a partition
filter, which evaluates
+ // the key on every partition before the data filters (SPARK-60809). An
alias hides the key only
+ // at the join, since pushdown rewrites the bloom filter through it. And
`coalesce` cannot be
+ // null, so no inferred `isnotnull` evaluates the key on every partition
first.
+ // Through a union, every branch counts, and without DPP each gets a bloom
filter. A narrowing
+ // cast can fail too.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 + 1 = bf2.c2 where bf2.a2
= 36" -> 1,
+ "select * from (select coalesce(f5 + 1, 0) as k, a5 from bf5part) p " +
+ "join bf2 on p.k = bf2.c2 where bf2.a2 = 36" -> 1,
+ "select * from (select f5 k, a5 v from bf5part union all " +
+ "select coalesce(f5 + 1, 0) k, a5 v from bf5filtered) u " +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36" -> 2,
+ "select * from bf5part join bf2 on cast(bf5part.f5 as tinyint) = bf2.c2
" +
+ "where bf2.a2 = 36" -> 1
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assertDidNotRewriteWithBloomFilter(query)
+ val filters = dynamicPruningFilters(query)
+ assert(filters.nonEmpty && filters.forall(_ == Literal.TrueLiteral),
filters)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter in an in subquery whose dpp
filter reuses a " +
+ "broadcast outside it") {
+ // The optimizer runs on the subquery on its own, where the broadcast of
`x` is not visible.
+ // The DPP filter on `p.f5` reuses it once the subquery is joined with the
rest of the query.
+ val query = "select /*+ broadcast(x) */ * from bf3 join (select c2 from
bf2 where a2 = 62) x " +
+ "on bf3.c3 = x.c2 where bf3.a3 in (select p.a5 from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) d on p.f5 = d.c2)"
+ def keysOnF5(query: String): Seq[Expression] =
+ bloomFilterKeys(query).filter(_.references.exists(_.name == "f5"))
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(keysOnF5(query).isEmpty)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ // Without the hint, nothing broadcasts `x`, and `p.f5` gets the bloom
filter.
+ val withoutHint = query.replace("/*+ broadcast(x) */ ", "")
+ assert(keysOnF5(withoutHint).nonEmpty)
+ assert(dynamicPruningFilters(withoutHint) == Seq(Literal.TrueLiteral))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter
under a numeric " +
+ "up-cast") {
+ // Type coercion casts `f5` to `bigint`, which cannot fail.
+ val query = "select * from bf5part join (select cast(c2 as bigint) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter for a key with a dropped dpp filter of a
floating point " +
+ "type") {
+ // `NormalizeFloatingNumbers` wraps the `double` join keys, which cannot
fail.
+ val query = "select * from bf5part join (select cast(c2 as double) k, a2
from bf2) d " +
+ "on bf5part.f5 = d.k where d.a2 = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation filter has a coalesce,
which cannot fail") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where coalesce(bf2.a2, 0) = 36"
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter under a " +
+ "decimal to bigint cast") {
+ // `Cast.canUpCast` takes `DECIMAL(19,0)` to `BIGINT`, but
`9999999999999999999` overflows. The
+ // query casts it only on rows with `v > 0`, which that partition does not
have.
+ withTable("fd") {
+ sql("create table fd (v int, d decimal(19,0)) using parquet partitioned
by (d)")
+ sql("insert into fd select cast(id as int) + 1, 36 from range(100)")
+ sql("insert into fd select -cast(id as int), cast('9999999999999999999'
as decimal(19,0)) " +
+ "from range(100)")
+ val query = "select * from fd join bf2 on cast(fd.d as bigint) = bf2.c2
" +
+ "where fd.v > 0 and bf2.a2 = 36"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose key can fail in a union
branch the lineage " +
+ "does not reach") {
+ // The lineage of `u.k` stops at the inner union of the second branch, but
pushdown moves the
+ // bloom filter into both inner branches, where it would cast `p` on every
partition. Each
+ // row of `p = 'abc'` has `v <= 0`, so the query itself never casts
`'abc'`.
+ withTable("sp") {
+ sql("create table sp (v int, p string) using parquet partitioned by (p)")
+ sql("insert into sp select cast(id as int) + 1, '36' from range(100)")
+ sql("insert into sp select -cast(id as int), 'abc' from range(100)")
+ val query = "select * from (select f5 as k, a5 as v from bf5part union
all " +
+ "select coalesce(cast(x as int), 0) as k, sum(v) as v from (select p
as x, v from sp " +
+ "where v > 0 union all select p, v from sp where v > 0) group by x) u
" +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36"
+ withDppTestConf() {
+ // Without DPP, pushdown puts a bloom filter into each of the three
branches.
+ assertBloomFilterWithoutDpp(query, 3)
+ assertDidNotRewriteWithBloomFilter(query)
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation side has a
non-deterministic column that " +
+ "the bloom filter does not read") {
+ // The bloom filter's subquery reads only `c2` of `d`, so `src` does not
change its keys. The
+ // DPP filter is not deterministic, so cleanup removes it.
+ val query = "select * from bf5part p join (select c2, input_file_name()
src from bf2 " +
+ "where a2 = 36) d on p.f5 = d.c2"
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assert(bloomFilterKeyNames(query) == Seq("f5"))
+ val expected = withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key ->
"false") {
+ sql(query).collect()
+ }
+ checkAnswer(sql(query), expected)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side reads a cache
that can give " +
+ "other rows when scanned again") {
+ // A cache that is not materialized can compute a partition twice, once
for the join and once
+ // for the bloom filter's subquery. A materialized cache gives the same
rows again.
+ val query = "select * from bf5part p join bf2c d on p.f5 = d.c2 where d.a2
= 36"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ val confs = Seq(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
Review Comment:
**15. [Nit] One `withDppTestConf` block is enough in the cache test**
AQE is pinned, so each `withDppTestConf(confs: _*)` block runs once, and the
assertions before `count()` only build optimized plans. One block with
`spark.table("bf2c").count()` between the two parts, without `val confs`, would
do the same. Two blocks stay right if the AQE pin goes away, so this could go
with item 3.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ c.child
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) => child
+ })
+ }
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if scanning
it again gives the
+ // same rows. Its filters, and the projections that the key and the filters
need, must pass
+ // `cannotFail`, which takes only expressions that are deterministic in the
subquery and do not
+ // fail there. A part of the join side outside that plan can only drop rows
from the join, so the
+ // bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ !evaluated.containsPattern(CTE) && !evaluated.exists {
+ case Project(projectList, _) => !projectList.forall(e =>
cannotFail(trimAliases(e)))
+ case Filter(condition, _) =>
!splitConjunctivePredicates(condition).forall(cannotFail)
+ case leaf: MaterializedLeafNode => !leaf.isOutputRepeatable
Review Comment:
**7. [Nit] The Known limits name only caches that are not materialized yet**
`isOutputRepeatable` is also false for a fully materialized cache whose plan
Spark cannot show to be repeatable. `isCachedLogicalPlanRepeatable` accepts
only a `Project`, `Filter`, `SubqueryAlias`, `Range`, `LocalRelation` or
trusted file source (`InMemoryRelation.scala` L519-L529), and
`isCachedRDDRepeatable` needs strict file reads (`InMemoryRelation.scala`
L462-L464). So a materialized cache of an aggregate, a join, a DSv2 table such
as an Iceberg one, an unconverted Hive SerDe table or a JDBC table, or of a
read with `ignoreMissingFiles` or `ignoreCorruptFiles`, gives no bloom filter
either. Uncached, the same DSv2 or JDBC relation passes `case _: LeafNode =>
false` at L440, so caching such a dimension turns the new bloom filter off.
The outcome is master's, so only the wording needs a change. The Known
limits ("So does a cache that is not materialized yet") and the test comment at
`InjectRuntimeFilterSuite.scala` L1518-L1519 could say "a cache that Spark
cannot show to give the same rows again, such as one that is not materialized
yet".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,86 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it takes out two wrappers
that cannot fail
+ // either: a numeric up-cast, and the normalization that
`NormalizeFloatingNumbers` puts around a
+ // floating point join key. `canUpCast` also takes `DECIMAL(19,0)` to
`BIGINT`, which can
+ // overflow. So a decimal to an integral type does not count. An
`onlyInBroadcast` DPP filter
+ // counts too, since it finds no broadcast in the bloom filter's subquery
and is `true` there.
+ private def cannotFail(e: Expression): Boolean = e match {
Review Comment:
**11. [Nit] `cannotFail` turns away `CollationKey`**
`RewriteCollationJoin` turns a collated inner-join key into
`CollationKey(k)` before `PartitionPruning` and this rule run. When a DSv2 scan
offers such a column for runtime filtering, the key gets a DPP filter. When
that filter is dropped, the key passes `isBinaryStable`, since it is a
`BINARY`, but `cannotFail` turns it away at L520 and as a creation key, since
`CollationKey` is not in the `canEvaluateUnconditionally` whitelist.
`CollationKey` cannot fail, and `XxHash64` over its bytes matches exactly the
strings that the collation calls equal, so the key could get the bloom filter.
The outcome is master's. One more arm next to the cast arm, along with the fix
for item 1, would do.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,19 +473,85 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. An uncorrelated IN subquery is joined with the
rest of the query
+ // later, where `PartitionPruning` puts a new DPP filter on the key,
and this rule decides
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if `creationSide`
Review Comment:
**8. [Nit] Step 3 of the overview does not name `isBinaryStable`**
Step 3 lists what `creationSide` checks as `canPushDown` and
`canEvaluateAgain`, but the gate at L552-L553 also checks
`UnsafeRowUtils.isBinaryStable(key.dataType)`, which a90e1fb added. Only the
last sentence of the comment at L543-L544 gives the reason. [Round-3 item
12](https://github.com/apache/spark/pull/59916#discussion_r4462890778) and
[round-4 item
10](https://github.com/apache/spark/pull/59916#discussion_r4470999435) brought
the overview in line with the gate. Adding "the key's type compares by its
bytes (`isBinaryStable`)" to this step would keep it so.
--
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]
