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