[
https://issues.apache.org/jira/browse/SPARK-60842?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-60842:
-------------------------------
Affects Version/s: 4.0.5
4.1.4
> Runtime bloom filter creation side can read partitions that DPP prunes in the
> join and fail the query
> -----------------------------------------------------------------------------------------------------
>
> Key: SPARK-60842
> URL: https://issues.apache.org/jira/browse/SPARK-60842
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.1, 4.0.5, 4.1.4
> Reporter: Peter Toth
> Priority: Major
>
> The creation side of a runtime bloom filter can keep the DPP filter of
> another join key. In the main plan, that DPP filter reuses a broadcast, so
> the join reads only the partitions it keeps. The bloom filter's subquery is
> planned on its own, where an {{onlyInBroadcast}} DPP filter finds no
> broadcast and is planned as {{true}}. So the subquery reads every partition,
> also those that the join never reads. A creation side key or filter that can
> fail, like an ANSI cast, then fails the query.
> Example, with ANSI on (the default):
> {code:sql}
> CREATE TABLE ka (v INT, x INT, b INT, a INT) USING parquet;
> INSERT INTO ka SELECT cast(id AS INT), 1, 10, 1 FROM range(2000);
> INSERT INTO ka SELECT cast(id AS INT), 1, 20, 2 FROM range(2000);
> CREATE TABLE kc (s STRING, flag INT, q INT) USING parquet PARTITIONED BY (q);
> INSERT INTO kc VALUES ('1', 1, 10), ('2', 1, 20), ('abc', 1, 99);
> CREATE TABLE kx (k1 INT, k2 INT) USING parquet;
> INSERT INTO kx VALUES (1, 10), (2, 20);
> SET spark.sql.adaptive.enabled=false;
> SET spark.sql.autoBroadcastJoinThreshold=-1;
> SET
> spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1;
> SELECT count(*) FROM (
> SELECT f.v FROM ka f JOIN kc c ON f.a = cast(c.s AS INT) AND f.b = c.q
> WHERE f.x = 1 AND c.flag = 1
> UNION ALL
> SELECT /*+ BROADCAST(g) */ g.v FROM kx JOIN (SELECT * FROM ka WHERE x = 1) g
> ON kx.k1 = g.a AND kx.k2 = g.b);
> {code}
> The DPP filter on {{c.q}} reuses the broadcast of {{g}}, so the join reads
> only the partitions {{q = 10}} and {{q = 20}}. The bloom filter on {{f.a}} is
> built from {{kc}} in its own subquery, where that DPP filter is {{true}}. So
> the subquery reads {{q = 99}} too, and the query fails with
> {{CAST_INVALID_INPUT}} for {{'abc'}}. With
> {{spark.sql.optimizer.runtime.bloomFilter.enabled=false}}, it returns 8000.
> dongjoon-hyun found this for a key with a dropped DPP filter while reviewing
> SPARK-60702
> (https://github.com/apache/spark/pull/59916#discussion_r4448448570). The
> example above has no DPP filter on {{f.a}}, so it takes the path that master
> already has.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
