dongjoon-hyun commented on code in PR #60176:
URL: https://github.com/apache/spark/pull/60176#discussion_r4494935437
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -88,6 +88,11 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
if (aggregate.containsPattern(PYTHON_UDF)) {
return filterApplicationSidePlan
}
+ // The subquery evaluates the creation side again, apart from the join. A
non-deterministic
+ // filter or key can give it different keys than the join sees.
+ if (!aggregate.deterministic) {
Review Comment:
**3. [Nit] `safeLineage` still checks determinism on the unpruned plan**
The materialized path requires `currentPlan.deterministic` before pruning
([L200](https://github.com/apache/spark/blob/8d3c09ffe314a66be4e58c8c6af7739fe3b01783/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala#L200)),
while this check runs after pruning. So a non-deterministic column that the
bloom filter does not read still disables the materialized path, for example
`SELECT k, rand() r FROM cached_dim` over a materialized cache, although the
same plan passes this check. A creation side that qualifies only through the
materialized statistics, with no selective predicate, then gets no bloom
filter. Now that this check guards correctness on both paths, could
`safeLineage` drop its determinism term, or check the pruned plan, so that the
two paths agree?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -88,6 +88,11 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
if (aggregate.containsPattern(PYTHON_UDF)) {
return filterApplicationSidePlan
}
+ // The subquery evaluates the creation side again, apart from the join. A
non-deterministic
+ // filter or key can give it different keys than the join sees.
+ if (!aggregate.deterministic) {
+ return filterApplicationSidePlan
Review Comment:
**1. [Low] A non-deterministic candidate stops the search for a
deterministic one**
This check runs after `extractSelectiveFilterOverScan` has already picked
one candidate. When that candidate is non-deterministic, the rule returns here,
and the alternatives that `extract` would have tried are never tried. For
example, the transitive branch
([L159-L175](https://github.com/apache/spark/blob/8d3c09ffe314a66be4e58c8c6af7739fe3b01783/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala#L159-L175))
runs only when `extract(left, ...)` returns `None`:
```sql
SELECT * FROM fact f JOIN (
SELECT l.k FROM (SELECT * FROM dim1 WHERE a = 1 AND rand() < 0.5) l
JOIN (SELECT * FROM dim2 WHERE b = 2) r ON l.k = r.k) t
ON f.k = t.k
```
`extract(left)` returns the non-deterministic `l`, so `r.k` is never tried,
and `fact` gets no bloom filter. `r` is deterministic and selective, and for
this inner join its keys are a superset of the keys the join sees. This costs
only performance. If the check ran inside `extract` at the leaf, on the pruned
plan, a rejected side would return `None` and the search would go on. Is the
current placement intended, for example to keep the backport small?
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1108,6 +1108,37 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
}
}
+ test("SPARK-60808: do not add bloom filter whose creation side is not
deterministic") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val queries = Seq(
+ "select * from bf2 where a2 = 62 and rand() < 0.5",
Review Comment:
**2. [Low] One non-deterministic conjunct drops the whole bloom filter**
In this case, `a2 = 62` alone is a selective, deterministic creation side,
but `rand() < 0.5` makes the rule drop the bloom filter. Dropping a filter
conjunct can only widen the creation side. So a bloom filter built without the
non-deterministic conjuncts still holds every key the join sees, as long as the
key expression is deterministic.
That would keep the pruning for a common shape: a sampling filter such as
`rand() < 0.1` on a dimension table. As the comment at L1131 says, `rand()`
picks the same rows in each scan of the same files, so such queries were
correct in practice, and with this PR they only lose the bloom filter.
One caveat: a dropped conjunct can guard a later conjunct that fails on the
rows it filters out, the case that SPARK-60980 fixed in `CombineFilters`. Would
this be worth doing here, or noting in the description as a possible
improvement?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -88,6 +88,11 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
if (aggregate.containsPattern(PYTHON_UDF)) {
return filterApplicationSidePlan
}
+ // The subquery evaluates the creation side again, apart from the join. A
non-deterministic
Review Comment:
**5. [Nit] The comment could say what the check does not cover**
The description lists a known limit: a leaf that returns other rows when it
is read again, such as a cache whose plan is not repeatable, a V2 scan with a
pushed limit, top-N or sample, or a `LogicalRDD` that is not a checkpoint.
Readers of this code will not see that, and may take `deterministic` as the
full guarantee. Could this comment add a sentence about it, like the comment on
`PartitionPruning.isCheaplyRecomputableMaterializedPlan`?
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -1108,6 +1108,37 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
}
}
+ test("SPARK-60808: do not add bloom filter whose creation side is not
deterministic") {
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ val queries = Seq(
+ "select * from bf2 where a2 = 62 and rand() < 0.5",
+ "select if(rand() < 0.5, c2, null) c2 from bf2 where a2 = 62",
+ // The bloom filter does not read `r`.
+ "select c2, rand() r from bf2 where a2 = 62"
+ ).map(creationSide => s"select * from bf1 join ($creationSide) t on
bf1.c1 = t.c2")
+ assert(queries.map(q =>
getNumBloomFilters(sql(q).queryExecution.optimizedPlan)) ==
+ Seq(0, 0, 1))
+ }
+ }
+
+ test("SPARK-60808: a non-deterministic creation side does not lose join
rows") {
+ withTable("dim", "fact") {
+ sql("CREATE TABLE dim USING parquet AS SELECT cast(id AS INT) k FROM
range(200)")
+ sql("CREATE TABLE fact USING parquet AS SELECT cast(id % 200 AS INT) k
FROM range(20000)")
+
withSQLConf(SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key
-> "1",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // Every `dim.k` is in `fact`. `rand()` would pick the same rows in
each scan of the same
+ // files, so the filter calls `java.lang.Math.random` instead.
+ checkAnswer(sql(
Review Comment:
**4. [Nit] The end-to-end test does not check that a bloom filter is
planned**
This test checks only `lost = 0`, which also holds when no bloom filter is
planned for some other reason. For example, `reflect` is marked `expensive`. If
the rule ever stops treating such a filter as simple or selective, this test
also passes on the old code. Could it also assert, with the same tables and
settings, that a deterministic creation side such as `WHERE k < 100` gets one
bloom filter (`getNumBloomFilters`)? Then the test keeps covering the case it
was written for.
--
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]
