[
https://issues.apache.org/jira/browse/SPARK-60808?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dongjoon Hyun updated SPARK-60808:
----------------------------------
Fix Version/s: 4.2.2
> Runtime bloom filter built from a non-deterministic creation side can drop
> join rows
> ------------------------------------------------------------------------------------
>
> Key: SPARK-60808
> URL: https://issues.apache.org/jira/browse/SPARK-60808
> Project: Spark
> Issue Type: Sub-task
> Components: SQL
> Affects Versions: 3.5.8, 4.3.0, 4.2.1, 4.0.5, 4.1.4
> Reporter: Peter Toth
> Assignee: Peter Toth
> Priority: Major
> Labels: correctness, pull-request-available
> Fix For: 4.4.0, 4.2.2, 4.3.1
>
>
> {{InjectRuntimeFilter}} can build a runtime bloom filter from a creation side
> with a non-deterministic filter. The bloom filter's subquery evaluates that
> side a second time, apart from the join, and the two evaluations can keep
> different rows. So the bloom filter can drop application side rows that match
> a creation side row that the join sees, and the query loses rows.
> {{extractSelectiveFilterOverScan}} accepts any {{Filter}} whose condition
> passes {{isSimpleExpression}}, which does not look at determinism. Only its
> materialized cache path checks it.
> Example, where every {{dim.k}} is in {{fact}}, so {{lost}} must be 0:
> {code:sql}
> CREATE TABLE dim USING parquet AS SELECT cast(id AS INT) k FROM range(200);
> CREATE TABLE fact USING parquet AS SELECT cast(id % 200 AS INT) k, id v FROM
> range(20000);
> SET spark.sql.autoBroadcastJoinThreshold=-1;
> SET
> spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1;
> SELECT count(*) total, count_if(f.k IS NULL) lost
> FROM (SELECT k FROM dim WHERE cast(reflect('java.lang.Math', 'random') AS
> DOUBLE) < 0.5) d
> LEFT JOIN fact f ON d.k = f.k;
> {code}
> Three runs gave {{[5042,42]}}, {{[5253,53]}} and {{[4448,48]}}. With
> {{spark.sql.optimizer.runtime.bloomFilter.enabled=false}}, {{lost}} is 0.
> DPP does not have this problem. A non-deterministic build plan makes the
> {{DynamicPruningSubquery}} non-deterministic, and
> {{CleanupDynamicPruningFilters}} replaces it with {{true}}.
> {{PartitionPruning}} still inserts such a filter and relies on that cleanup.
> Found by dongjoon-hyun while reviewing SPARK-60702:
> https://github.com/apache/spark/pull/59916#discussion_r4437009548
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
