dongjoon-hyun commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4486187844
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,99 @@ 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, bottom-up, two
wrappers that cannot
+ // fail either over a child that it takes: a numeric up-cast, and the
normalization that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. A
stand-in keeps the type of
+ // what it replaces, so each wrapper is checked against a child of its own
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.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) =>
Review Comment:
**2. [Nit] The accepting half of the normalization guard is pinned only with
an attribute child**
Since the rewrite is bottom-up again, the guard on this line can see a child
that is already a stand-in. Every test that accepts a floating point key gives
it an `AttributeReference`: the test at `InjectRuntimeFilterSuite.scala`
L1444-L1452 normalizes the `upCast` stand-in of `cast(f5 as double)` and the
attribute `k`. So a narrower arm, `case n @
KnownFloatingPointNormalized(NormalizeNaNAndZero(_: AttributeReference)) =>`,
passes every test, though it turns away a normalized key over a `Coalesce` or a
struct field, which cannot fail either. Under the top-down rewrite of 1263713,
the query at `InjectRuntimeFilterSuite.scala` L1447 caught that arm, since the
guard still saw the cast there. [Round-6 item
3](https://github.com/apache/spark/pull/59916#discussion_r4484533080) pinned
only the rejecting half.
A query whose normalized key is not an attribute would pin it, for example
as a second query of that test:
```scala
"select * from bf5part join (select cast(c2 as double) k, a2 from bf2) d " +
"on coalesce(cast(bf5part.f5 as double), 0D) = d.k where d.a2 = 36"
```
Here the guard sees `coalesce(upCast, 0.0)`. The explicit cast matters: with
`coalesce(bf5part.f5, 0)`, type coercion casts the whole `coalesce`, which
turns into the stand-in again. The narrower arm would only fall back to
master's outcome, hence a Nit. 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,99 @@ 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, bottom-up, two
wrappers that cannot
+ // fail either over a child that it takes: a numeric up-cast, and the
normalization that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. A
stand-in keeps the type of
+ // what it replaces, so each wrapper is checked against a child of its own
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.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)()
+ })
+ }
+
+ // `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 {
+ 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
Review Comment:
**1. [Low] A scan with a pushed LIMIT or top-N counts as giving the same
rows again**
This arm takes every other leaf as one that gives the same rows again. A
`DataSourceV2ScanRelation` that took a fully pushed LIMIT, top-N or OFFSET does
not: the source picks the rows, and a second read can pick others. Unlike a
`LogicalRDD`, catalyst can see this. `onlyRowWisePushdowns`
(`DataSourceV2Relation.scala` L191-L197) is false for such a scan
(`V2ScanRelationPushDown.scala` L914-L918 and L1068), and
`DataSourceV2Strategy.scala` L142 already keeps storage filters away from such
a scan, since the pushed operation decides which rows it returns.
For example, take a JDBC catalog table `pg.s.d` read as one partition,
`spark.sql.autoBroadcastJoinThreshold=-1`, `spark.sql.defaultSizeInBytes=1m` (a
JDBC scan reports no statistics), and `f` partitioned by `k` and over the
application-side threshold:
```sql
SELECT * FROM f JOIN (SELECT * FROM (SELECT * FROM pg.s.d ORDER BY v LIMIT
100) WHERE flag = 1) x
ON f.k = x.k
```
`flag = 1` sits above the limit when `pushDownFilters` runs, which is before
`pushDownLimitAndOffset` (`V2ScanRelationPushDown.scala` L58 and L65), so it
stays in Spark above the scan. With one partition, the top-N is pushed fully
(`JDBCScanBuilder.scala` L341-L353), so the `Sort` and the `Limit` go away
(`V2ScanRelationPushDown.scala` L1162-L1170 and L1241). `extract` takes
`Filter(flag = 1, scan)` as the creation side (L236-L240), and this line
accepts it. Nothing broadcasts, so the `onlyInBroadcast` DPP filter on `f.k` is
dropped, and `f.k` gets a bloom filter whose subquery sends `ORDER BY v LIMIT
100` to the database a second time. If `v` has ties at the cut, that read can
return other rows. The bloom filter, a partition filter on `f`, then drops `f`
rows that the join's own read matches. A `LIMIT` without `ORDER BY` is the
same. On master, the DPP filter keeps the bloom filter off `f.k`.
In-tree, only JDBC pushes a limit fully, and it reaches this only with a
lowered `spark.sql.defaultSizeInBytes`, since it reports no size. A connector
that pushes a limit fully and reports a small size reaches it with the
defaults. The cache check also disagrees: `hasRepeatableLogicalPlan`, behind
`isOutputRepeatable`, trusts no V2 scan (`InMemoryRelation.scala` L519-L530),
so a cache of the same scan gives no bloom filter, while the scan itself gets
one.
An arm before this one, as for the CTE reference, would turn such scans away:
```scala
case r: DataSourceV2ScanRelation if !r.onlyRowWisePushdowns => true
```
It also turns away a pushed aggregate or join, which gives master's outcome.
`StorageFilterTestSource` with `push=topN` pushes fully and reports 10 MB
(`StorageFilterTestSource.scala` L257-L264 and L281), so it could pin this in a
plan test. Otherwise the Known limits could name it next to `LogicalRDD`. 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,99 @@ 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, bottom-up, two
wrappers that cannot
+ // fail either over a child that it takes: a numeric up-cast, and the
normalization that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. A
stand-in keeps the type of
+ // what it replaces, so each wrapper is checked against a child of its own
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.transformUp {
Review Comment:
**3. [Nit] No test covers the cost fix of round-6 item 2**
This rewrite gives the same answer as 1263713 for every input, so no
assertion can tell them apart. And the deepest nesting that reaches
`cannotFail` in the suite is a cast over a cast under the normalization
(`InjectRuntimeFilterSuite.scala` L1405), where the cost does not show. So the
exponential code of 1263713 still passes every test. The 28-level query of
[round-6 item
2](https://github.com/apache/spark/pull/59916#discussion_r4484533059) plans in
about 0.1 s now, as you measured, so it could go into the `withDppTestConf()`
block of the test at `InjectRuntimeFilterSuite.scala` L1462 at almost no cost:
```scala
val nested = (1 to 28).foldLeft("b2 + 1") { (acc, k) =>
s"if(d2 > $k, $acc, cast(b2 as decimal(${10 + k}, 0)))"
}
assert(bloomFilterKeys("select * from bf5part join bf2 on bf5part.f5 =
bf2.c2 " +
s"where bf2.a2 = 36 and $nested > 0").isEmpty)
```
A regression would show as a hang rather than a quick failure, hence a Nit.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +304,67 @@ 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()
Review Comment:
**5. [Nit] `dynamicPruningFilters` runs the query also with AQE off**
With AQE off, `executedPlan` holds the planned DPP filters before anything
runs: `PlanDynamicPruningFilters` and `ReuseExchangeAndSubquery` are
preparation rules (`QueryExecution.scala` L841 and L886), and
`InSubqueryExec.plan` does not change at run time. So this `collect()` changes
nothing in what the helper returns there. Only AQE plans the filters as the
query runs. By my count, 38 of the 72 calls in the new and changed tests run
with AQE off, and 19 of those come right after a helper that already ran the
same query three times.
`if (conf.adaptiveExecutionEnabled) df.collect()`, with the doc at L324
updated, would skip those runs. One call needs its run, though: `withoutHint`
at L1430 gets this PR's bloom filter and has no answer check, so its AQE-off
run is the only one that executes that bloom filter with AQE off. A
`sql(withoutHint).collect()` there would keep it.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1097,681 @@ 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)
Review Comment:
**4. [Nit] One `assertDidNotRewriteWithBloomFilter` call is left**
8124920 replaced 15 of the 16 calls that [round-6 item
4](https://github.com/apache/spark/pull/59916#discussion_r4484533088) counted,
and this is the 16th. The test comes from master, but the PR runs it under
`withDppTestConf`, in both AQE modes, and adds `dynamicPruningFilters` at
L1103, so the query now runs 8 times where master ran it 3 times. With
`reuseBroadcastOnly` off, the DPP filter is not `onlyInBroadcast`, so `isKept`
(`InjectRuntimeFilter.scala` L525-L527) keeps the bloom filter off, and both
settings of the answer check give the same plan.
`assert(bloomFilterKeys(query).isEmpty)` would catch the same mutant, and
`dynamicPruningFilters` still runs the query.
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1034,19 +1097,681 @@ 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)
+ }
+ assert(bloomFilterKeys(query).isEmpty)
+ 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)
+ assert(bloomFilterKeys(query).isEmpty)
+ // 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)
+ assert(bloomFilterKeys(query).isEmpty)
+
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)
+ assert(bloomFilterKeys(query).isEmpty)
+
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)
+ assert(bloomFilterKeys(query).isEmpty)
+
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 {
+ assert(bloomFilterKeys(query).isEmpty)
+ 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)
+ assert(bloomFilterKeys(query).isEmpty)
+ assert(dynamicPruningFilters(query).isEmpty)
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side is a
non-deterministic pruned " +
+ "side") {
+ // 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).
+ 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 or under the normalization of a
floating point key.
+ 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,
+ "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
+ ).foreach { case (query, numBloomFilters) =>
+ withDppTestConf() {
+ assertBloomFilterWithoutDpp(query, numBloomFilters)
+ assert(bloomFilterKeys(query).isEmpty)
+ 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)
+ assert(bloomFilterKeys(narrow).isEmpty)
+ 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") {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter under a " +
+ "decimal to bigint cast") {
+ // `Cast.canUpCast` takes `DECIMAL(19,0)` to `BIGINT`, but
`9999999999999999999` overflows. The
+ // query casts it only on rows with `v > 0`, which that partition does not
have.
+ withTable("fd") {
+ sql("create table fd (v int, d decimal(19,0)) using parquet partitioned
by (d)")
+ sql("insert into fd select cast(id as int) + 1, 36 from range(100)")
+ sql("insert into fd select -cast(id as int), cast('9999999999999999999'
as decimal(19,0)) " +
+ "from range(100)")
+ val query = "select * from fd join bf2 on cast(fd.d as bigint) = bf2.c2
" +
+ "where fd.v > 0 and bf2.a2 = 36"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "1") {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ sql(query).collect()
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose key can fail in a union
branch the lineage " +
+ "does not reach") {
+ // The lineage of `u.k` stops at the inner union of the second branch, but
pushdown moves the
+ // bloom filter into both inner branches, where it would cast `p` on every
partition. Each
+ // row of `p = 'abc'` has `v <= 0`, so the query itself never casts
`'abc'`.
+ withTable("sp") {
+ sql("create table sp (v int, p string) using parquet partitioned by (p)")
+ sql("insert into sp select cast(id as int) + 1, '36' from range(100)")
+ sql("insert into sp select -cast(id as int), 'abc' from range(100)")
+ val query = "select * from (select f5 as k, a5 as v from bf5part union
all " +
+ "select coalesce(cast(x as int), 0) as k, sum(v) as v from (select p
as x, v from sp " +
+ "where v > 0 union all select p, v from sp where v > 0) group by x) u
" +
+ "join bf2 on u.k = bf2.c2 where bf2.a2 = 36"
+ withDppTestConf() {
+ // Without DPP, pushdown puts a bloom filter into each of the three
branches.
+ assertBloomFilterWithoutDpp(query, 3)
+ assert(bloomFilterKeys(query).isEmpty)
+ sql(query).collect()
+ }
+ }
+ }
+
+ test("SPARK-60702: add bloom filter whose creation side has a
non-deterministic column that " +
+ "the bloom filter does not read") {
+ // The bloom filter's subquery reads only `c2` of `d`, so `src` does not
change its keys. The
+ // DPP filter is not deterministic, so cleanup removes it.
+ val query = "select * from bf5part p join (select c2, input_file_name()
src from bf2 " +
+ "where a2 = 36) d on p.f5 = d.c2"
+ withDppTestConf() {
+ assert(bloomFilterKeyNames(query) == Seq("f5"))
+ val expected = withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key ->
"false") {
+ sql(query).collect()
+ }
+ checkAnswer(sql(query), expected)
+ }
+ }
+
+ test("SPARK-60702: add bloom filter from a cache only if spark can tell that
scanning it " +
+ "again gives the same rows") {
+ // A cache that is not materialized yet can compute a partition twice,
once for the join and
+ // once for the bloom filter's subquery. So can a cache whose plan Spark
cannot show to be
+ // repeatable. `bf2c` would give the same rows again, but Spark can tell
only once it is
+ // materialized.
+ def query(cache: String): String =
+ s"select * from bf5part p join $cache d on p.f5 = d.c2 where d.a2 = 36"
+ withTempView("bf2c", "bf2r") {
+ withCache("bf2c", "bf2r") {
+ spark.table("bf2").createOrReplaceTempView("bf2c")
+ spark.catalog.cacheTable("bf2c")
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ assertBloomFilterWithoutDpp(query("bf2c"))
+ assert(bloomFilterKeys(query("bf2c")).isEmpty)
+ spark.table("bf2c").count()
+ assert(bloomFilterKeyNames(query("bf2c")) == Seq("f5"))
+ spark.table("bf2").where("cast(reflect('java.lang.Math', 'random')
as double) < 2.0")
+ .createOrReplaceTempView("bf2r")
+ spark.catalog.cacheTable("bf2r")
+ spark.table("bf2r").count()
+ assertBloomFilterWithoutDpp(query("bf2r"))
+ assert(bloomFilterKeys(query("bf2r")).isEmpty)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter for a key with a dropped dpp
filter of a collated " +
+ "string type") {
+ // The bloom filter hashes the bytes of the key, so `'abc'` and `'ABC'`
would not match.
+ withSQLConf("spark.sql.catalog.testcat" ->
classOf[InMemoryTableCatalog].getName) {
+ withTable("testcat.cf", "testcat.cd") {
+ sql("create table testcat.cf (k string collate utf8_lcase, v int)
partitioned by (k)")
+ sql("insert into testcat.cf values ('abc', 1)")
+ sql("create table testcat.cd (k string collate utf8_lcase, flag int)")
+ sql("insert into testcat.cd values ('ABC', 1), ('zzz', 0)")
+ val query = "select v from testcat.cf where k in (select k from
testcat.cd where flag = 1)"
+ withDppTestConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "0") {
+ assertBloomFilterWithoutDpp(query)
+ assert(bloomFilterKeys(query).isEmpty)
+ checkAnswer(sql(query), Row(1))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60702: do not add bloom filter whose creation side reads a cte")
{
+ // The reference to `d` hides that its plan is not deterministic. With
exchange reuse off, the
+ // bloom filter's subquery would draw a second sample of `d`.
+ withTable("cdim", "cfact") {
+ spark.range(200).selectExpr("cast(id as int) k", "cast(id % 2 as int) x")
+ .write.saveAsTable("cdim")
+ spark.range(20000).selectExpr("cast(id % 200 as int) k", "id v")
Review Comment:
**6. [Nit] The CTE test writes a 200-partition table that it never reads**
The test only optimizes its query (L1630-L1631), but it writes `cfact` as
200 partitions, about 400 files with two tasks, and analyzes them. It needs two
things from `cfact`: a size from `analyze`, without which the application-side
check counts it as 0, and a size large enough that `PartitionPruning` sees a
benefit with exchange reuse off, that is, half of it over the size of the scans
on the other side (`PartitionPruning.scala` L205-L206). The other side is the
CTE reference, which reports the size of `cdim`, a few KB. So `id % 20` should
keep the DPP filter with a wide margin, at a tenth of the files. The sizes come
from the [round-2
reproducer](https://github.com/apache/spark/pull/59916#discussion_r4448448541),
which ran the query.
--
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]
