[
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: [email protected]
For additional commands, e-mail: [email protected]
