[
https://issues.apache.org/jira/browse/SPARK-60752?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth reassigned SPARK-60752:
----------------------------------
Assignee: Peter Toth
> Storage-partitioned join fails when a DPP subquery's build side prunes the
> Scan it shares with the join
> -------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60752
> URL: https://issues.apache.org/jira/browse/SPARK-60752
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.3.0, 4.2.1, 4.0.5, 4.1.4
> Reporter: Peter Toth
> Assignee: Peter Toth
> Priority: Major
> Labels: pull-request-available
>
> A storage-partitioned join can fail with {{Can't zip RDDs with unequal
> numbers of partitions}} when the build side of a DPP subquery has its own
> runtime filter.
> A DPP subquery that does not reuse a broadcast plans its build side again
> from the logical plan. Its {{BatchScanExec}} wraps the same {{Scan}} instance
> as the join's scan on that side. The steps:
> # The build side has a runtime filter of its own, a scalar subquery filter
> routed there by SPARK-56467. The DPP subquery runs it first.
> # The connector prunes its planned partitions in {{filter()}}, which the
> runtime filtering contracts allow.
> # The join's {{BatchScanExec}} is a copy made after planning, since
> {{ReusedSubquery}} replaced the scalar subquery in its runtime filter. So it
> plans its input partitions only at execution, from the already pruned
> {{Scan}}.
> # Its keyed partitioning then has fewer partitions than the other side of the
> join.
> With {{InMemoryTableWithV2FilterCatalog}} tables {{items}} partitioned by
> {{id}} and {{purchases}} partitioned by {{item_id}}, ids 1 to 7 with one
> split per key, AQE off, DPP on,
> {{spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=false}},
> {{spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio=10}} and
> {{spark.sql.autoBroadcastJoinThreshold=-1}}:
> {code:sql}
> SELECT i.id, i.name, p.price FROM items i JOIN purchases p ON i.id = p.item_id
> WHERE i.id = (SELECT max(item_id) FROM purchases WHERE price > 4.0)
> {code}
> fails with {{Can't zip RDDs with unequal numbers of partitions: List(7, 1)}}.
> * {{InMemoryCatalystRuntimeFilterCatalog}} fails the same way, also with
> {{i.id >= (SELECT min(item_id) ...)}}. There the join's {{purchases}} scan
> reports 3 partitions and the DPP subquery's {{purchases}} scan 7, and both
> wrap the same {{Scan}}.
> * {{InMemoryTableCatalog}} prunes only on {{IN}}, so it does not reproduce
> there.
> * Reading the scans' {{outputPartitioning}} before execution avoids the
> failure.
> * With AQE on, the query returns the right rows.
> * Without the scalar subquery, e.g. {{WHERE p.price > 4.0}}, it returns the
> right rows.
> Reproduced on master, branch-4.x and branch-4.2. Not reproduced on branch-4.1
> and branch-4.0, which do not route scalar subquery filters into runtime
> filtering. Found while reviewing SPARK-60260
> (https://github.com/apache/spark/pull/59789).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
