dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4484533031
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,101 @@ 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 replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
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.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)()
+ })
+ }
+
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isNumericUpCast(c: Cast): Boolean =
+ 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])
+
+ // 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 Spark
can tell that scanning
+ // it again gives the same rows. It must not read a file metadata column
either, such as
+ // `_metadata.file_block_start`, which depends on how the scan splits its
files. The subquery can
+ // split them differently. 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.exists { node =>
+ node.references.exists {
+ case FileSourceMetadataAttribute(_) => true
+ case _ => false
+ } || (node match {
+ case Project(projectList, _) => !projectList.forall(e =>
cannotFail(trimAliases(e)))
+ case Filter(condition, _) =>
!splitConjunctivePredicates(condition).forall(cannotFail)
+ case _: CTERelationRef => true
+ case leaf: MaterializedLeafNode => !leaf.isOutputRepeatable
+ case _: LeafNode => false
Review Comment:
**1. [Low] A `foreachBatch` DataFrame hides non-determinism like a lazy
checkpoint**
The Known limits name a lazily checkpointed `LogicalRDD`, after [round-3
item 7](https://github.com/apache/spark/pull/59916#discussion_r4462890737).
That case needs `Dataset.checkpoint(eager = false)`, which round 3 called
contrived. But `foreachBatch` hands the user the same kind of leaf on every
micro-batch, with no checkpoint at all. `ForeachBatchSink.addBatch` passes
`LogicalRDD.fromDataset(data.queryExecution.toRdd, ...)`
(`ForeachBatchSink.scala` L35-L36). Nothing persists it, since the note at
`ForeachBatchSink.scala` L46-L48 leaves that to the user. It has
`fromCheckpoint = false` and the statistics of the micro-batch plan
(`ExistingRDD.scala` L204). For a file source, those are the sizes of the
batch's files, so a small batch passes the creation-side threshold.
Take this join inside the function, with
`spark.sql.autoBroadcastJoinThreshold=-1` and `fact` partitioned by `p` and
over the application-side threshold:
```scala
fact.join(df.where("flag = 1"), fact("p") === df("k"))
```
It gets an `onlyInBroadcast` DPP filter on `p` (`PartitionPruning.scala`
L126), which the shuffle join drops. This line then accepts the leaf, and the
bloom filter's subquery computes the last stage of the micro-batch a second
time. If the stream evaluates `reflect('java.lang.Math', 'random')`, or a UDF
marked `asNondeterministic()`, which the `LogicalRDD` hides, the two runs give
different rows. The join then loses the `fact` rows whose keys only its own run
has. On master, the DPP filter keeps the bloom filter off this key.
Could the Known limit name any `LogicalRDD` that is not a materialized
checkpoint, with the `foreachBatch` DataFrame as the common case? Also, "The
results stay the same" in the user-facing section does not hold under this
limit or the one on files that cannot be read, so it could point to the Known
limits. 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,101 @@ 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 replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
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.transformDown {
+ case c: Cast if isNumericUpCast(c) && cannotFail(c.child) =>
Review Comment:
**2. [Low] `cannotFail` re-runs the nested guards after a failed one**
When `cannotFail(c.child)` is false here, `transformDown` keeps the cast and
then descends into the same child (`TreeNode.scala` L586-L587), where each
nested up-cast runs its guard again. The kept cast already decides the answer,
since `canEvaluateUnconditionally` has no arm for it (`ExprUtils.scala` L276).
So with d nested numeric up-casts over something that can fail, the number of
guard calls doubles with each level, where the single `transformUp` of a90e1fb
was linear.
For example, under the suite's `withDppTestConf()`:
```scala
val e = (1 to 28).foldLeft("b2 + 1") { (acc, k) =>
s"if(d2 > $k, $acc, cast(b2 as decimal(${10 + k}, 0)))"
}
sql(s"select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36 and $e > 0")
```
Type coercion widens each `if` by one decimal digit, and
`PushFoldableIntoBranches` keeps the casts, since neither branch is foldable.
The DPP filter on `f5` is dropped, so L451 calls `cannotFail` on that conjunct,
which makes about 2^29 calls before it returns `false`. Real queries rarely
nest that deep, but generated SQL can.
This came from my round-5 suggestion. Keeping the same-type stand-ins, but
rewriting bottom-up with a guard that does not recurse, gives the same answer
for every input without the doubling. Each cast is still checked against its
own child type, since the rewritten child keeps it:
```scala
case _ => ExprUtils.canEvaluateUnconditionally(e.transformUp {
case c: Cast if isNumericUpCast(c) &&
ExprUtils.canEvaluateUnconditionally(c.child) =>
AttributeReference("upCast", c.dataType, c.nullable)()
case n @ KnownFloatingPointNormalized(NormalizeNaNAndZero(child))
if ExprUtils.canEvaluateUnconditionally(child) =>
AttributeReference("normalized", n.dataType, n.nullable)()
})
```
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,101 @@ 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 replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
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.transformDown {
+ case c: Cast if isNumericUpCast(c) && cannotFail(c.child) =>
+ AttributeReference("upCast", c.dataType, c.nullable)()
+ case n @ KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) if
cannotFail(child) =>
Review Comment:
**3. [Low] No test pins the child check of the normalization arm**
The stand-in hides the wrapper's child from `canEvaluateUnconditionally`, so
each guard is the only check of that child. The cast guard at L419 is pinned:
type coercion turns `cast(bf5part.f5 as tinyint) = bf2.c2`
(`InjectRuntimeFilterSuite.scala` L1400) into `cast(cast(f5 as tinyint) as int)
= c2`, and without that guard the key gets a bloom filter, which
`InjectRuntimeFilterSuite.scala` L1407 rejects. Dropping `if cannotFail(child)`
on this line passes every test, though. The only floating point key with a
dropped DPP filter, at `InjectRuntimeFilterSuite.scala` L1444, normalizes
`cast(f5 as double)` and `k`, which cannot fail.
Without this guard, a `FLOAT` or `DOUBLE` key over a partition column
expression that can fail would get a bloom filter that runs as a partition
filter. That is the SPARK-60809 failure, which the DPP filter keeps out on
master. Adding this query to the list at `InjectRuntimeFilterSuite.scala`
L1393-L1404 would pin it:
```scala
"select * from bf5part join (select cast(c2 as double) k, a2 from bf2) d " +
"on cast(bf5part.f5 as tinyint) = d.k where d.a2 = 36" -> 1,
```
I traced this from the code and did not run it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1096,677 @@ 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) {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ 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, also over an up-cast.
+ 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,
+ "select * from bf5part join bf2 on cast(cast(bf5part.f5 as float) as
int) = 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)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(!bloomFilterKeyNames(query).contains("f5"))
+ 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(bloomFilterKeyNames(withoutHint).contains("f5"))
+ 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: check each cast in the creation side against its own
child") {
+ // Type coercion puts an up-cast into the `coalesce`. The first query
casts the `coalesce` down,
+ // which can fail, and the second casts it up. Neither may fail to
optimize.
+ val narrow = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where cast(coalesce(bf2.a2, 0L) as int) = 36"
+ val wide = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where coalesce(bf2.a2, 0L) > 35.5D"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(narrow)
+ assertDidNotRewriteWithBloomFilter(narrow)
+ assertRewroteWithBloomFilterForDroppedDpp(wide)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side reads a file
metadata column") {
+ // `_metadata.file_block_start` depends on how the scan splits its files,
which the bloom
+ // filter's subquery can do differently from the join.
+ val query = "select * from bf5part p join bf2 d on p.f5 = d.c2 " +
+ "where d.a2 = 36 and d._metadata.file_block_start = 0"
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter under a date to " +
+ "timestamp cast") {
+ // `Cast.canUpCast` takes a date to a timestamp, which can overflow past
the year 294247.
+ withTable("fdt", "tsd") {
+ sql("create table fdt (v int, d date) using parquet partitioned by (d)")
+ sql("insert into fdt select cast(id as int), date'2020-01-01' from
range(100)")
+ sql("create table tsd (ts timestamp, flag int) using parquet")
+ sql("insert into tsd values (timestamp'2020-01-01 00:00:00', 1), " +
+ "(timestamp'2021-01-01 00:00:00', 0)")
+ val query = "select * from fdt join tsd on fdt.d = tsd.ts where tsd.flag
= 1"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
Review Comment:
**6. [Nit] The description does not say that the date to timestamp test runs
with AQE off**
This test pins AQE off, but its part of the bullet under "New tests, with
AQE on and off unless noted" ("... a decimal to bigint cast or a date to
timestamp cast ...") does not say so. The decimal test in the same bullet runs
in both modes, and every other test that pins AQE off has "with AQE off" in its
bullet. As with [round-3 item
14](https://github.com/apache/spark/pull/59916#discussion_r4462890796), "a
decimal to bigint cast or, with AQE off, a date to timestamp cast" would match
the test.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1096,677 @@ 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) {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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
Review Comment:
**5. [Nit] The pruned-side test names the wrong reason for dropping the DPP
filter**
"has the deterministic build plan `f`, so it is dropped" reads as if
determinism decides it. `isKept` (`InjectRuntimeFilter.scala` L527-L529) does
not look at determinism. The filter is dropped because nothing broadcasts `f`,
since this test sets the broadcast threshold to -1 (L1376). Next to the
previous test, whose comment says that cleanup drops a DPP filter with a
non-deterministic build plan, the sentence also suggests that this filter
escapes cleanup. It does not: it stays above the non-deterministic filter of
`d`, so `CleanupDynamicPruningFilters` turns it into `true` after this rule
anyway. Maybe:
```scala
// Nothing broadcasts `f`, the deterministic build plan of the DPP
filter on `d.k`, so that
// filter 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).
```
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +304,66 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
checkWithAndWithoutFeatureEnabled(query, shouldReplace = false)
}
+ // Runs `f` with the thresholds of the DPP tests, `reuseBroadcastOnly` on,
and `confs`, with AQE
+ // on and off unless `confs` sets it.
+ def withDppTestConf(confs: (String, String)*)(f: => Unit): Unit = {
+ val adaptiveModes = confs.collectFirst {
+ case (SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, adaptive) => Seq(adaptive)
+ }.getOrElse(Seq("true", "false"))
+ adaptiveModes.foreach { adaptive =>
+ withSQLConf(Seq(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive) ++ confs: _*) {
+ withClue(s"adaptive=$adaptive: ")(f)
+ }
+ }
+ }
+
+ // Runs `query` and returns the DPP filters that physical planning left on
the file source scans:
+ // the class of the subquery that each filter reads, or the filter itself if
it reads none.
+ def dynamicPruningFilters(query: String): Seq[Any] = {
+ // Keep the scans in the final plan even if they return no rows.
+ withSQLConf(
+ SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key ->
AQEPropagateEmptyRelation.ruleName) {
+ val df = sql(query)
+ df.collect()
+ collect(df.queryExecution.executedPlan) {
+ case scan: FileSourceScanExec => scan.partitionFilters.collect {
+ case DynamicPruningExpression(in: InSubqueryExec) => in.plan.getClass
+ case DynamicPruningExpression(filter) => filter
+ }
+ }.flatten
+ }
+ }
+
+ // Asserts that `query` gets a bloom filter, and that physical planning
plans its DPP filter as
+ // `true`.
+ def assertRewroteWithBloomFilterForDroppedDpp(query: String): Unit = {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+
+ // The keys that the bloom filters of `query` check.
+ def bloomFilterKeys(query: String): Seq[Expression] =
+ sql(query).queryExecution.optimizedPlan.flatMap(_.expressions)
+ .flatMap(_.collect { case BloomFilterMightContain(_, XxHash64(Seq(key),
_)) => key })
+
+ // The names of the columns that the bloom filters of `query` check.
+ def bloomFilterKeyNames(query: String): Seq[String] =
+ bloomFilterKeys(query).flatMap(_.references.map(_.name))
+
+ // Asserts that `query` gets `numBloomFilters` bloom filters without DPP, so
that a test that sees
+ // fewer with DPP shows that the rule kept them out. In the tests for a key
or a creation side
+ // that is not safe, that is master's bloom filter for a key without a DPP
filter, which can fail
+ // the query or drop rows there: SPARK-58486 (a collated key), SPARK-60808
(a creation side that
Review Comment:
**7. [Nit] The `assertBloomFilterWithoutDpp` doc leaves out SPARK-60842**
The doc names SPARK-58486, SPARK-60808 and SPARK-60809 for master's bloom
filters in the controls. The control for `castInKey` at L1658 shows the
SPARK-60842 kind instead. Without DPP, master adds a bloom filter on `c.q` and
one on `f.a`, built from `rc` with the key `cast(c.s as int)`. The bloom filter
on `c.q` drops the partition `q = 99` only in the join, so the subquery of the
one on `f.a` still reads `'abc'` there and fails. The description files that
case under SPARK-60842, which appears nowhere in the suite. Adding "SPARK-60842
(a creation side that can fail on rows that only the join's side drops)" would
complete the list.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1096,677 @@ 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) {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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") {
+ assertBloomFilterWithoutDpp(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,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ 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, also over an up-cast.
+ 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,
+ "select * from bf5part join bf2 on cast(cast(bf5part.f5 as float) as
int) = 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)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
+ assert(!bloomFilterKeyNames(query).contains("f5"))
+ 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(bloomFilterKeyNames(withoutHint).contains("f5"))
+ 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: check each cast in the creation side against its own
child") {
+ // Type coercion puts an up-cast into the `coalesce`. The first query
casts the `coalesce` down,
+ // which can fail, and the second casts it up. Neither may fail to
optimize.
+ val narrow = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where cast(coalesce(bf2.a2, 0L) as int) = 36"
+ val wide = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 " +
+ "where coalesce(bf2.a2, 0L) > 35.5D"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(narrow)
+ assertDidNotRewriteWithBloomFilter(narrow)
Review Comment:
**4. [Nit] The new no-bloom-filter checks run each query three times**
`assertDidNotRewriteWithBloomFilter` (L303-L305) runs the query once with
bloom filters off, and twice in `checkAnswer` with them on (`QueryTest.scala`
L180 and `CheckAnswerHelper.scala` L94). When the rule adds no bloom filter,
both settings give the same plan, so the answer check cannot catch anything. A
mutant that adds a bloom filter fails the count check, as it would fail
`assert(bloomFilterKeys(query).isEmpty)`, which seven new checks already use.
The new and changed tests call it 16 times, or 38 times with the AQE modes,
which is about 114 runs. In 29 of those calls, `dynamicPruningFilters` runs the
query again right after (L331). This line and the `cast(cast(f5 as float) as
int)` entry at L1402-L1403 add four calls in 1263713.
`assert(bloomFilterKeys(query).isEmpty)` would do for most of them. Where a
test needs to show that the query itself runs, as the decimal test at
L1503-L1520 and the union test at L1522-L1541 do, one `sql(query).collect()`
would keep that.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,101 @@ 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 replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
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.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)()
+ })
+ }
+
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isNumericUpCast(c: Cast): Boolean =
+ 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])
+
+ // 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 Spark
can tell that scanning
+ // it again gives the same rows. It must not read a file metadata column
either, such as
+ // `_metadata.file_block_start`, which depends on how the scan splits its
files. The subquery can
+ // split them differently. 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.exists { node =>
Review Comment:
**9. [Nit] The metadata check can be the first arm of the match**
The new check wraps the match in `node => ... || (node match {...})`. A
guarded first arm keeps one flat match, as round 5 did with the CTE check:
```scala
!evaluated.exists {
case node if node.references.exists(a =>
FileSourceMetadataAttribute.isValid(a.metadata)) =>
true
case Project(projectList, _) => !projectList.forall(e =>
cannotFail(trimAliases(e)))
case Filter(condition, _) =>
!splitConjunctivePredicates(condition).forall(cannotFail)
case _: CTERelationRef => true
case leaf: MaterializedLeafNode => !leaf.isOutputRepeatable
case _: LeafNode => false
case _ => true
}
```
The guard has to stay before the `Filter` arm, where a `GetStructField` over
`_metadata` passes `cannotFail`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,101 @@ 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 replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
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.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)()
+ })
+ }
+
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isNumericUpCast(c: Cast): Boolean =
+ 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])
+
+ // 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 Spark
can tell that scanning
+ // it again gives the same rows. It must not read a file metadata column
either, such as
+ // `_metadata.file_block_start`, which depends on how the scan splits its
files. The subquery can
+ // split them differently. 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.exists { node =>
+ node.references.exists {
+ case FileSourceMetadataAttribute(_) => true
Review Comment:
**8. [Nit] The metadata check turns away every `_metadata` field**
`FileSourceMetadataAttribute` matches the `_metadata` struct, so a filter on
`_metadata.file_size`, `file_name` or `file_modification_time` also turns the
creation side away. Only `file_block_start` and `file_block_length` come from
the split (`FileFormat.scala` L372-L373), and `FilePruningRunner` skips exactly
those two for the same reason (`FileIndex.scala` L59-L61). For example, the
query of the new test with `d._metadata.file_size > 0` gets no runtime filter
on `p.f5`, though the bloom filter would be exact. The outcome is master's, and
round 5 offered both scopes, so either is fine. With this scope, the Known
limits could say that any `_metadata` read turns the bloom filter away, next to
the lines on CTEs and caches.
--
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]
