peter-toth commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4472713998
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter has a
non-deterministic build " +
+ "plan") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
Review Comment:
Done. Both run with AQE off now, and their bullets say so.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,98 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `key` cannot fail: an attribute, a numeric up-cast of
such a key, or the
+ // normalization that `NormalizeFloatingNumbers` puts around a floating
point join key.
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isSafeKey(key: Expression): Boolean = key match {
+ case _: Attribute => true
+ case c: Cast if c.child.dataType.isInstanceOf[NumericType] &&
+ c.dataType.isInstanceOf[NumericType] &&
Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType]) =>
+ isSafeKey(c.child)
+ case KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) =>
isSafeKey(child)
+ case _ => false
+ }
+
+ // Whether evaluating `condition` cannot fail: comparisons of safe keys and
literals, null
+ // checks, and their combinations. A DPP filter on a safe key counts too,
since its own subquery,
+ // if it has one, runs in the join as well.
+ private def isSafeCondition(condition: Expression): Boolean = condition
match {
Review Comment:
Done in
[a90e1fb](https://github.com/apache/spark/pull/59916/commits/a90e1fb8b02274ebca49b0364b216577b1d4fd75):
`cannotFail` asks `ExprUtils.canEvaluateUnconditionally`, after it takes out
the numeric up-casts and the float normalization, which cannot fail either. The
only other case is an `onlyInBroadcast` DPP filter. Your `coalesce` query now
gets the bloom filter, and it is a new test. I kept the decimal rule local,
since moving it into `object Cast` would touch `try_cast`. The bloom filter of
a lower join still keeps the bloom filter out. That only loses it, and the
Known limits say so.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ object InjectRuntimeFilter extends Rule[LogicalPlan]
with PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. An uncorrelated IN subquery is joined with the
rest of the query
+ // later, where `PartitionPruning` puts a new DPP filter on the key,
and this rule decides
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). On a partition column, the bloom filter becomes
a partition filter,
+ // which evaluates the key on every partition before the data filters.
So a key that can
+ // fail could fail the query (SPARK-60809).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is no DPP filter on the key that physical planning may
keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
+ key: Expression,
+ side: LogicalPlan,
+ creation: FilterCreationSide) =
+ !hasDroppedDpp || (canPushDown(key, side) &&
canEvaluateAgain(creation))
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. The key has no dropped DPP filter, or step 3 above allows
the bloom filter
val hasShuffle = isProbablyShuffleJoin(left, right, hint)
if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
!hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
- }
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .filter(canTakeBloomFilter(l, left, _))
Review Comment:
Agreed that it is master's outcome. I'd leave it, and the Known limits now
say that the search does not try the other side.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ object InjectRuntimeFilter extends Rule[LogicalPlan]
with PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. An uncorrelated IN subquery is joined with the
rest of the query
+ // later, where `PartitionPruning` puts a new DPP filter on the key,
and this rule decides
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). On a partition column, the bloom filter becomes
a partition filter,
+ // which evaluates the key on every partition before the data filters.
So a key that can
+ // fail could fail the query (SPARK-60809).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is no DPP filter on the key that physical planning may
keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
+ key: Expression,
+ side: LogicalPlan,
+ creation: FilterCreationSide) =
+ !hasDroppedDpp || (canPushDown(key, side) &&
canEvaluateAgain(creation))
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. The key has no dropped DPP filter, or step 3 above allows
the bloom filter
Review Comment:
Fixed. The item is gone from both lists, and the overview names
`canEvaluateAgain`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +485,106 @@ object InjectRuntimeFilter extends Rule[LogicalPlan]
with PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
So does every DPP
+ // filter in a subquery. An uncorrelated IN subquery is joined with the
rest of the query
+ // later, where `PartitionPruning` puts a new DPP filter on the key,
and this rule decides
+ // that one against all the broadcasts. A scalar subquery, which
includes an uncorrelated
+ // EXISTS, is planned on its own, so a key there with a DPP filter gets
no bloom filter.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if
+ // `canTakeBloomFilter` allows it: pushdown can take it down safely
(`canPushDown`), and its
+ // creation side is safe. Pushdown moves the bloom filter along its
key, down to the leaves
+ // the key comes from, and into every branch of a union. There it must
not land in a plan in
+ // `reused`. It would change the broadcast, and physical planning would
drop the DPP filter
+ // that reuses it. A reused broadcast elsewhere in the join side does
not matter, such as a
+ // broadcast dimension joined to the same fact table. The key must also
be safe at every
+ // leaf (`isSafeKey`). On a partition column, the bloom filter becomes
a partition filter,
+ // which evaluates the key on every partition before the data filters.
So a key that can
+ // fail could fail the query (SPARK-60809).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ // Pushdown rewrites the key of the bloom filter through the aliases on
its way down. Where the
+ // key cannot be followed, pushdown can still move a copy of the bloom
filter, so the check
+ // fails there.
+ def canPushDown(key: Expression, side: LogicalPlan): Boolean = {
+ val lineages = findExpressionAndTrackAllLineagesDown(key, side, strict =
true)
+ lineages.nonEmpty && lineages.forall { case (leafKey, leaf) =>
+ isSafeKey(leafKey) && !reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, l) ++
dynamicPruningSubqueries(right, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is no DPP filter on the key that physical planning may
keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ // The bloom filter evaluates its creation side again, apart from
the join. Its subquery
+ // can also read more of that side than the join does, since
filters on other keys, DPP
+ // filters or bloom filters, can narrow only the join's side.
+ def canTakeBloomFilter(
Review Comment:
Done. One local `creationSide(side, key, other, otherKey)` replaces the
three, close to your sketch. It also checks the key type from item 2.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1088,566 @@ class InjectRuntimeFilterSuite extends
SharedSparkSession
test("Runtime bloom filter join: do not add bloom filter if dpp filter
exists " +
"on the same column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertDidNotRewriteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.f5 = bf2.c2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where
bf2.a2 = 62"
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
}
}
test("Runtime bloom filter join: add bloom filter if dpp filter exists on " +
"a different column") {
-
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
- SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000") {
- assertRewroteWithBloomFilter("select * from bf5part join bf2 on " +
- "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62")
+ withDppTestConf(SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key
-> "false") {
+ val query = "select * from bf5part join bf2 on " +
+ "bf5part.c5 = bf2.c2 and bf5part.f5 = bf2.f2 where bf2.a2 = 62"
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(classOf[SubqueryExec]))
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the dpp filter needs a broadcast that
the shuffle " +
+ "join lacks") {
+ // `bf2.a2 = 36` keeps a row of `bf2` that joins, so the bloom filter is
not empty when it meets
+ // the partition values of `bf5part`.
+ Seq(
+ "select * from bf5part join bf2 on bf5part.f5 = bf2.c2 where bf2.a2 =
36",
+ "select * from bf2 join bf5part on bf2.c2 = bf5part.f5 where bf2.a2 =
36").foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ assert(sql(query).collect().nonEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter reuses the
broadcast of " +
+ "the join") {
+ // The aggregate shuffles below the broadcast join, so the join may get a
bloom filter. The
+ // join broadcasts `bf2` by a hint, with the partitioned table on either
side, or by its size.
+ // Without DPP, or without exchange reuse to give the DPP filter the
broadcast, the join gets
+ // the bloom filter.
+ Seq(
+ "select /*+ broadcast(bf2) */ * from (select f5, sum(a5) from bf5part
group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "2000",
+ "select /*+ broadcast(bf2) */ * from bf2 join (select f5, sum(a5) from
bf5part " +
+ "group by f5) t on bf2.c2 = t.f5 where bf2.a2 = 62" -> "2000",
+ "select * from (select f5, sum(a5) from bf5part group by f5) t " +
+ "join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62" -> "5000"
+ ).foreach { case (query, threshold) =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> threshold) {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ withSQLConf(SQLConf.EXCHANGE_REUSE_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if another join broadcasts the
build plan of " +
+ "the dpp filter") {
+ // The first join is a shuffle join, and its DPP filter reuses the
broadcast of the second.
+ val query = "select p.f5, p.a5, d.* from bf5part p join bf2 d on p.f5 =
d.c2 " +
+ "where d.a2 = 62 union all select /*+ broadcast(d) */ t.f5, t.s, d.*
from " +
+ "(select f5, sum(a5) s from bf5part group by f5) t join bf2 d on t.f5 =
d.c2 " +
+ "where d.a2 = 62"
+ withDppTestConf() {
+ // Without DPP, both `p.f5` and `t.f5` get a bloom filter.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+ // One of the DPP filters can read the broadcast through a reused
subquery.
+ val reuses = Set[Any](classOf[SubqueryBroadcastExec],
classOf[ReusedSubqueryExec])
+ val filters = dynamicPruningFilters(query)
+ assert(filters.size == 2 && filters.forall(reuses.contains), filters)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter") {
+ // The DPP filter on `q.f5` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. That bloom filter would go into `r`, the broadcast that the
DPP filter on `p.f5`
+ // reuses. `q` is on either side of its join.
+ Seq(
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf5filtered q join bf2 on q.f5 = bf2.c2 where bf2.a2 = 62) r on p.f5
= r.rf",
+ "select /*+ broadcast(r) */ * from bf5part p join (select q.f5 rf, q.a5
ra from " +
+ "bf2 join bf5filtered q on bf2.c2 = q.f5 where bf2.a2 = 62) r on p.f5
= r.rf"
+ ).foreach { query =>
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter in " +
+ "another union branch") {
+ // The DPP filter on `u.k` cannot reuse a broadcast, so it does not keep
the bloom filter off
+ // its key. Pushdown would move that bloom filter into both branches, and
so into `d`, the
+ // broadcast that the DPP filter on `q.f5` reuses. Over the cache of `d`,
cleanup drops the
+ // copy of the DPP filter on `u.k`, but not the bloom filter.
+ val query = "select * from (select p.f5 k, p.a5 v from bf5part p union all
select " +
+ "/*+ broadcast(d) */ d.c2 k, q.a5 v from bf5part q join bf2c d on q.f5 =
d.c2 " +
+ "where d.a2 = 62) u join bf3 on u.k = bf3.c3 where bf3.a3 = 62"
+ withTempView("bf2c") {
+ withCache("bf2c") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Without DPP, pushdown puts a bloom filter into each branch.
+ assertBloomFilterWithoutDpp(query, 2)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter off a key whose dpp filter reuses a
broadcast side " +
+ "that has a dpp filter of its own") {
+ // The DPP filter on `q.f5` sits on top of `q` when the rule runs, and
cleanup drops it later
+ // because of `q.f5 = 51`. Then the DPP filter on `t.f5` reuses the
broadcast of `q`, so `t.f5`
+ // gets no bloom filter. `t.c5` gets one, since the DPP filter on its key
is gone. `q` is on
+ // either side of the join.
+ Seq(
+ "select /*+ broadcast(q) */ * from (select f5, c5, sum(a5) s from
bf5part " +
+ "where b5 > 0 group by f5, c5) t join bf5filtered q on t.f5 = q.c5 and
t.c5 = q.f5 " +
+ "where q.a5 > 50 and q.f5 = 51",
+ "select /*+ broadcast(q) */ * from bf5filtered q join (select f5, c5,
sum(a5) s from " +
+ "bf5part where b5 > 0 group by f5, c5) t on q.c5 = t.f5 and q.f5 =
t.c5 " +
+ "where q.a5 > 50 and q.f5 = 51"
+ ).foreach { query =>
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter that takes a broadcast from a dpp
filter that is " +
+ "not only in broadcast") {
+ // With `reuseBroadcastOnly` off, the DPP filter on `p.f5` has a benefit,
so it can run as its
+ // own subquery, but physical planning reuses the broadcast of `r` first.
The DPP filter on
+ // `q.f5` has none, so it is only in broadcast and cannot reuse a
broadcast. A bloom filter on
+ // `q.f5` would go into `r`.
+ withTable("bf5big") {
+ sql("select a5 + id * 100 as a5, b5, c5, d5, e5, f5 from bf5part cross
join range(400)")
+ .write.partitionBy("f5").saveAsTable("bf5big")
+ val query = "select /*+ broadcast(r) */ * from bf5big p join (select
q.f5 rf, q.a5 ra " +
+ "from bf5filtered q join bf5part d on q.f5 = d.c5 where d.a5 = 62) r
on p.f5 = r.rf"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "1.0",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+
assert(dynamicPruningFilters(query).contains(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: keep the bloom filter of a key without a dpp filter") {
+ // Pushdown moves the bloom filter on `s.k` into `x`, the broadcast that
the DPP filter on
+ // `p.f5` reuses. The key has no DPP filter, so the rule adds the bloom
filter all the same.
+ val query = "select * from (select x.c2 k, count(*) from bf5part p join
(select c2 from bf2 " +
+ "where a2 = 62) x on p.f5 = x.c2 group by x.c2) s join bf3 on s.k =
bf3.c3 " +
+ "where bf3.a3 = 62"
+ withDppTestConf() {
+ assertRewroteWithBloomFilter(query)
+ }
+ }
+
+ test("SPARK-60702: keep the dropped dpp filter of the key out of the bloom
filter's creation " +
+ "side") {
+ // `q` is too small to take the bloom filter, so `b` takes it, and `q` is
its creation side.
+ // The DPP filter on `q.f5` stays out of the bloom filter's subquery. `q`
is on either side of
+ // the join.
+ withTable("bigapp") {
+ sql("select a5 + id * 100 as k, b5 from bf5part cross join range(400)")
+ .write.saveAsTable("bigapp")
+ Seq(
+ "select * from bf5filtered q join bigapp b on q.f5 = b.k where q.a5 >
50 and b.b5 = 1",
+ "select * from bigapp b join bf5filtered q on b.k = q.f5 where q.a5 >
50 and b.b5 = 1"
+ ).foreach { query =>
+ withDppTestConf(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"20000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ val bloomFilterPlans =
sql(query).queryExecution.optimizedPlan.collect {
+ case Filter(condition, _) => condition.collect {
+ case subquery: ScalarSubquery => subquery.plan
+ }
+ }.flatten
+ assert(bloomFilterPlans.size == 1)
+ val creationSide = bloomFilterPlans.head
+ assert(creationSide.collectLeaves().exists {
+ case LogicalRelationWithTable(_, Some(table)) =>
+ table.identifier.table == "bf5filtered"
+ case _ => false
+ }, creationSide)
+ assert(!creationSide.exists {
+ case Filter(_: DynamicPruningSubquery, _) => true
+ case _ => false
+ }, creationSide)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if physical planning keeps the
dpp filter over the " +
+ "build plan of a null-aware anti join") {
+ // The join is a shuffle join. The `not in` subquery broadcasts the build
plan of its DPP
+ // filter. Without AQE, physical planning keeps the filter, though it
builds its own
+ // broadcast. AQE plans the filter as `true`, so there the key gets the
bloom filter.
+ val query = "select * from bf5part p join (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null) d on p.f5 = d.c2 where p.a5 not in (select c2 from bf2 where a2 =
62 and c2 is not " +
+ "null)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500",
+ SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD.key -> "-1") {
+ withSQLConf(SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "false") {
+ assertRewroteWithBloomFilter(query)
+ }
+ if (conf.adaptiveExecutionEnabled) {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ } else {
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) ==
Seq(classOf[SubqueryBroadcastExec]))
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if the join type rules out the broadcast
the dpp filter " +
+ "needs") {
+ // `bf2` is small enough to broadcast, but a right outer join cannot
broadcast its right side,
+ // so the DPP filter does not count. The aggregate shuffles below the
join, so the join can get
+ // the bloom filter.
+ val query = "select * from (select f5, sum(a5) s from bf5part group by f5)
t " +
+ "right join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter if a shuffle hash hint rules out the
broadcast the dpp " +
+ "filter needs") {
+ // `bf2` is small enough to broadcast, but the hint asks for a shuffle
hash join. The aggregate
+ // shuffles below the join, so the join can get the bloom filter.
+ val query = "select /*+ shuffle_hash(bf2) */ * from (select f5, sum(a5) s
from bf5part " +
+ "group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "5000") {
+ assertRewroteWithBloomFilterForDroppedDpp(query)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter if the dpp filter has a
non-deterministic build " +
+ "plan") {
+ // The build plan of the DPP filter would be the creation side of the
bloom filter. A bloom
+ // filter would evaluate it again, apart from the join, and could drop
rows (SPARK-60808).
+ // Cleanup drops such a DPP filter.
+ val query = "select * from bf5part join (select * from bf2 where a2 = 36
and " +
+ "cast(reflect('java.lang.Math', 'random') as double) < 2.0) d on
bf5part.f5 = d.c2"
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query)
+ assertDidNotRewriteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // The DPP filter on `d.k` has the deterministic build plan `f`, so it is
dropped. `d` is too
+ // small to take a bloom filter, so `f` would take one, built from `d`,
which is not
+ // deterministic (SPARK-60808).
+ withTable("ndd", "ndf") {
+ spark.range(40).selectExpr("id as k", "id as
w").write.partitionBy("k").saveAsTable("ndd")
+ spark.range(40000).selectExpr("id % 40 as k", "id as
v").write.saveAsTable("ndf")
+ sql("analyze table ndd compute statistics")
+ sql("analyze table ndf compute statistics")
+ val nddSize =
spark.table("ndd").queryExecution.optimizedPlan.stats.sizeInBytes
+ val query = "select count(*) from (select k from ndd " +
+ "where cast(reflect('java.lang.Math', 'random') as double) < 0.5) d " +
+ "join ndf f on d.k = f.k where f.v >= 0"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
+ (nddSize + 1).toString) {
+ assert(getNumBloomFilters(sql(query).queryExecution.optimizedPlan) ==
0)
Review Comment:
Done. `bloomFilterKeyNames` replaces the repeated `flatMap`, and that
control uses `assertBloomFilterWithoutDpp`.
--
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]
