[jira] [Updated] (SPARK-58783) SPJ with runtime filtering fails when join keys are a subset of partition keys

["Peter Toth (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58783?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Peter Toth updated SPARK-58783:
-------------------------------
    Affects Version/s: 4.1.0
                       4.0.0
                           (was: 5.0.0)

> SPJ with runtime filtering fails when join keys are a subset of partition keys
> ------------------------------------------------------------------------------
>
>                 Key: SPARK-58783
>                 URL: https://issues.apache.org/jira/browse/SPARK-58783
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 4.1.0, 4.0.0
>            Reporter: Peter Toth
>            Priority: Major
>
> On branch-4.1 and lower, a storage-partitioned join and runtime filtering cannot coexist on the same scan when the join keys are a subset of the partition keys. The query fails on the driver before a single task runs:
> {code}
> org.apache.spark.SparkException: During runtime filtering, data source must not report new partition values that are not present in the original partitioning.
> {code}
> BatchScanExec.outputPartitioning projects the partition expressions and values down to the join keys when SPJ params were pushed down (spjParams.joinKeyPositions), but filteredPartitions validates the re-planned partitions against that projected partitioning while wrapping them with their full, unprojected HasPartitionKey.partitionKey(). The two sides of the comparison are in different spaces.
> It only fails when the join key is not the first partition key. InternalRowComparableWrapper compares only as many fields as it has data types, so with the join key at position 0 it happens to compare the right column and the mismatch goes unnoticed.
> Three things have to come together, and removing any one of them makes the same query work:
>  * SPJ with spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true, so Spark projects the scan partitioning down to the join keys
>  * a scan reporting KeyGroupedPartitioning with more partition keys than the join key, where the join key is not the first one
>  * a runtime filter on that same scan
> A copy-on-write MERGE INTO hits this by default, because the row-level operation group filter (spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled) is on: the target is partitioned by e.g. months(created), bucket(4, id) and the merge joins on id. The only workaround is to turn the group filter off, which loses runtime file pruning for every row-level operation in the session.
> The fix is to validate against the unprojected partitioning. Everything after the check already works in the full partition key space: groupPartitions groups by the unprojected keys, and the projection, the reducers, the common partition values and the empty-partition padding all happen later, in inputRDD.
> master and branch-4.2 are not affected. SPARK-55535 / SPARK-55092 moved the join key projection out of the scan into GroupPartitionsExec, so there both sides of the comparison are already unprojected. That refactor is too large for a maintenance branch, hence this targeted fix for branch-4.1 (and branch-4.0).



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org