peter-toth opened a new pull request, #58013:
URL: https://github.com/apache/spark/pull/58013
### What changes were proposed in this pull request?
This PR makes `BatchScanExec.filteredPartitions` validate the partitions a
data source reports after runtime filtering against the **unprojected**
partitioning (`super.outputPartitioning`) instead of `outputPartitioning`.
`BatchScanExec.outputPartitioning` projects the partition expressions and
values down to the join keys when SPJ params were pushed down
(`spjParams.joinKeyPositions`), and it also replaces the partition values with
the common partition values of the join. But `filteredPartitions` wraps the
re-planned partitions with their full, unprojected
`HasPartitionKey.partitionKey()`, so the two sides of the comparison are in
different spaces.
Nothing else needs to change, because 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`.
This PR targets `branch-4.1` (and later `branch-4.0`) only. `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.
### Why are the changes needed?
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:
```
org.apache.spark.SparkException: During runtime filtering, data source must
not report new partition values that are not present in the original
partitioning.
at
org.apache.spark.sql.execution.datasources.v2.BatchScanExec.filteredPartitions$lzycompute(BatchScanExec.scala:100)
at
org.apache.spark.sql.execution.datasources.v2.BatchScanExec.inputRDD$lzycompute(BatchScanExec.scala:150)
```
Three things have to come together, and removing any one of them makes the
same query work:
1. SPJ with
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true`,
so Spark projects the scan partitioning down to the join keys.
2. A scan reporting `KeyGroupedPartitioning` with more partition keys than
the join key, where the join key is **not** the first partition key. If it is
the first one the mismatch goes unnoticed, because
`InternalRowComparableWrapper` compares only as many fields as it has data
types, so it happens to compare the right column.
3. A runtime filter on that same scan.
A copy-on-write `MERGE INTO` hits all three 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`. This makes SPJ unusable for `MERGE`, which is the case it is
most worth having. The only workaround is to turn the group filter off, which
loses runtime file pruning for every row-level operation in the session.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes the bug above: a query that failed with the exception now
runs. There is no change for queries that don't combine SPJ, a join key that is
a subset of the partition keys, and a runtime filter on the same scan. This is
a change compared to released 4.0.x and 4.1.x; `master` and `branch-4.2`
already behave this way.
### How was this patch tested?
Added a test to `KeyGroupedPartitioningSuite` that joins on the second of
two partition keys with a dynamic partition filter on that scan, over the
`pushDownValues` x `partiallyClustered` matrix. It fails with the exception
above without the fix and passes with it.
Also ran the full `KeyGroupedPartitioningSuite` (71 tests, covering the
partially-clustered and dynamic-filtering combinations that exercise the common
partition values path), plus `GroupBasedMergeIntoTableSuite`,
`GroupBasedDeleteFromTableSuite`, `GroupBasedUpdateTableSuite` and
`DataSourceV2Suite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
