[
https://issues.apache.org/jira/browse/SPARK-60809?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136639#comment-18136639
]
Peter Toth commented on SPARK-60809:
------------------------------------
The same failure also happens without DPP or a bloom filter when the key is an
alias. {{PruneFileSourcePartitions}} evaluates a filter on the key, which the
optimizer infers from the join condition, on every partition at planning time:
{code:sql}
CREATE TABLE dim (k INT, a INT) USING parquet;
INSERT INTO dim VALUES (1, 1), (2, 1), (3, 0);
CREATE TABLE fact (v INT, x INT, p INT) USING parquet PARTITIONED BY (p);
INSERT INTO fact SELECT cast(id AS INT), 1, 50 FROM range(100);
INSERT INTO fact SELECT cast(id AS INT), 1, 100 FROM range(100);
INSERT INTO fact SELECT cast(id AS INT), 0, 0 FROM range(100);
SELECT count(*) FROM (SELECT 100 div p AS k, x FROM fact) f JOIN dim d ON f.k =
d.k WHERE d.a = 1 AND f.x > 0;
{code}
With ANSI on, this fails with {{DIVIDE_BY_ZERO}} in
{{PruneFileSourcePartitions}}, with DPP on or off, and with runtime bloom
filters off. Every row of the partition {{p = 0}} has {{x <= 0}}, so the join
itself never computes {{100 div 0}}.
> DPP and runtime bloom filters can fail a query by evaluating a throwable
> pruning key on partitions that the data filters exclude
> --------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60809
> URL: https://issues.apache.org/jira/browse/SPARK-60809
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.1, 4.0.5, 4.1.4
> Reporter: Peter Toth
> Priority: Major
>
> A DPP filter or a runtime bloom filter on a partition column becomes a
> partition filter. It is evaluated for every partition, before any data
> filter. When its key can throw, like an ANSI cast, it can fail on a partition
> whose rows never reach the join. The same query without the runtime filter
> succeeds.
> Example, with ANSI on (the default):
> {code:sql}
> CREATE TABLE dim (k INT, a INT) USING parquet;
> INSERT INTO dim VALUES (1, 1), (2, 1), (3, 0);
> CREATE TABLE fact (v INT, x INT, p STRING) USING parquet PARTITIONED BY (p);
> INSERT INTO fact SELECT cast(id AS INT), 1, '1' FROM range(1000);
> INSERT INTO fact SELECT cast(id AS INT), 1, '2' FROM range(1000);
> INSERT INTO fact SELECT cast(id AS INT), 0, 'abc' FROM range(1000);
> SELECT count(*) FROM fact f JOIN dim d ON cast(f.p AS INT) = d.k WHERE d.a =
> 1 AND f.x > 0;
> {code}
> Every row of the partition {{p = 'abc'}} has {{x <= 0}}, so the join itself
> never casts {{'abc'}}.
> * With the default configs, {{dim}} is broadcast and the DPP filter runs. The
> query fails with {{CAST_INVALID_INPUT}} for {{'abc'}}.
> * With {{spark.sql.autoBroadcastJoinThreshold=-1}},
> {{spark.sql.optimizer.dynamicPartitionPruning.enabled=false}} and
> {{spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1}},
> the runtime bloom filter on the same key fails the same way.
> * With {{spark.sql.autoBroadcastJoinThreshold=-1}} and
> {{spark.sql.optimizer.runtime.bloomFilter.enabled=false}}, the DPP filter is
> {{true}}, and the query returns 2000.
> SPARK-60602 and SPARK-60633 fixed the same kind of problem for filter
> pushdown.
> Found by dongjoon-hyun while reviewing SPARK-60702:
> https://github.com/apache/spark/pull/59916#discussion_r4437009558
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
