peter-toth commented on code in PR #60176:
URL: https://github.com/apache/spark/pull/60176#discussion_r4495117516
##########
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:
Yes, the placement is intended. The check runs once, on the subquery that is
actually executed, next to the Python UDF check, which stops the search the
same way. In `extract` it would be needed at each leaf that builds a creation
side, on a plan built only for the check. It matters only for a creation side
join whose first side is non-deterministic, and only for performance, so I'd
keep the single check, which also backports as is.
##########
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:
I'd rather not drop conjuncts here. As you note, a dropped conjunct can
guard a later one that fails, so this would need the same can-fail analysis
that SPARK-60980 added to `CombineFilters`. Also, `rand()` keeps the same rows
only while both scans split the input the same way, because `Rand` seeds each
partition with `seed + partitionIndex`. The subquery's scan can get other
partitions, for example when a DPP filter prunes the scan in the join but is
`true` in the subquery. I added the sampling filter case to the user-facing
section of the description.
##########
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:
`safeLineage` decides whether the exact materialized statistics can be used,
and it is stricter on purpose. Relaxing it would change when the materialized
path applies, which is an optimization, not part of this fix. The 4.0 to 4.2
branches don't have that path, so I'd keep it out of this PR, and the backports
stay the same as master.
##########
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:
Added in
[2f630c1](https://github.com/apache/spark/pull/60176/commits/2f630c16fe18d935386e36261e27a1917fbfa11b):
with the same tables and settings, `WHERE k < 100` gets one bloom filter.
##########
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:
Added a sentence in
[2f630c1](https://github.com/apache/spark/pull/60176/commits/2f630c16fe18d935386e36261e27a1917fbfa11b).
--
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]
