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