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]

Reply via email to