ammarchalifah opened a new pull request, #59410: URL: https://github.com/apache/spark/pull/59410
### What changes were proposed in this pull request? This PR makes `KeyGroupedPartitionedScan.getOutputKeyGroupedPartitioning` project `originalPartitionValues` down to the join keys whenever `spjParams.joinKeyPositions` is set, using the same projection already applied to `partitionValues`. When SPJ pushes join-key-subset parameters into a scan, the method already trims `expressions` and `partitionValues` to the join keys but leaves `originalPartitionValues` as the full partition row from the data source. Later, during AQE planning, `KeyGroupedPartitioning.createShuffleSpec` calls `KeyGroupedPartitioning.apply(...)`, which projects `originalPartitionValues` using the already-trimmed partition expressions. When the join keys are not a leading prefix of the partition keys (for example, `identity(date), bucket(id), bucket(data)` joined only on `id`), position 0 of the full row (a string) is read with the bucket column's integer type, which fails with a type mismatch. This is a follow-up to SPARK-58783 (https://github.com/apache/spark/pull/58013), which fixed runtime-filter validation in `BatchScanExec.filteredPartitions` but did not update `originalPartitionValues` in the scan's output partitioning. The user-facing report is apache/iceberg#15349 (https://github.com/apache/iceberg/issues/15349): copy-on-write `MERGE INTO` on Iceberg tables partitioned by an identity column followed by bucket columns, with SPJ and `spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true`. This PR targets `branch-4.1` only. `master` and `branch-4.2` are not affected: SPARK-55535 / SPARK-55092 moved join-key projection out of the scan into `GroupPartitionsExec`, and `originalPartitionValues` is no longer used on that path. ### Why are the changes needed? Storage-partitioned joins with join keys that are a non-prefix subset of the partition keys fail during driver-side planning when AQE is enabled (including skew-join handling). The query never reaches task execution. Typical trigger (also described in apache/iceberg#15349): 1. SPJ with `spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled=true`. 2. A table partitioned by a leading identity column plus bucket transforms, where the join uses only the bucket columns. 3. AQE enabled (`spark.sql.adaptive.enabled=true`). Without this change, `getOutputKeyGroupedPartitioning` returns a `KeyGroupedPartitioning` whose expressions and `partitionValues` are trimmed but whose `originalPartitionValues` still hold the full Iceberg partition row. `KeyGroupedPartitioning.createShuffleSpec` then mis-projects those rows and throws, for example: ``` IllegalArgumentException: Wrong class, expected java.lang.CharSequence, but was java.lang.Integer ``` or, depending on column order and types: ``` SparkException: Unknown type for int field ``` SPARK-58783 (https://github.com/apache/spark/pull/58013) addressed a related failure in `BatchScanExec.filteredPartitions` when runtime filtering is combined with join-key projection. That fix does not cover the `originalPartitionValues` path exercised by AQE's call to `createShuffleSpec`. The same misalignment also affects `EnsureRequirements` logic for partially clustered joins (~L643), which reads `spec.partitioning.originalPartitionValues`; trimming at the scan makes those values consistent with the projected partition expressions, so no separate change is required there. ### Does this PR introduce _any_ user-facing change? Yes. It fixes a bug: queries that failed during AQE planning now run. There is no change for queries that do not combine SPJ, join keys that are a non-prefix subset of the partition keys, and AQE. This is a change compared to released and current unreleased 4.1.x; `master` and `branch-4.2` already behave correctly after the SPJ refactor. Before (4.1.x without this fix): `MERGE INTO` / join on bucket columns only, with a leading identity partition column, fails at planning time with a type mismatch in partition-key handling. After: The same query completes and SPJ avoids a shuffle on the scan side. ### How was this patch tested? Added one test in `KeyGroupedPartitioningSuite`, modeled on the SPARK-58783 cases but with AQE and skew join enabled (the path those tests deliberately disable). The test covers a leading `identity(date)` partition with a join on trailing `bucket(32, id)` only; on branch-4.1 without this change that shape fails with a type mismatch during planning in `KeyGroupedPartitioning.createShuffleSpec`. ### Was this patch authored or co-authored using generative AI tooling? Yes: Grok 4.7 -- 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]
