Peter Toth created SPARK-60842:
----------------------------------
Summary: 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
Reporter: Peter Toth
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]
