[SPARK-59045][4.2][SQL] Fix SPJ ClassCastException when reducer changes partition key data type

[ulysses-you]

What changes were proposed in this pull request?

Backport of #58335 to branch-4.2.

KeyedShuffleSpec.reducersBothWays now pairs each Reducer with the reduced partition expression it produces (KeyReducer), and GroupPartitionsExec.outputPartitioning reports the reduced expression instead of the original partition expressions when reducers are applied. The stored expression is re-targeted at each KeyedPartitioning's own key attribute at the use site via the new TransformExpression.withReference, so a chained storage-partitioned join keeps every side's partitioning intact.

GroupPartitionsExec.doCanonicalize additionally normalizes the exprIds inside KeyReducer - plan canonicalization does not reach into the plain case class - and the reducer applied for an identity-vs-transform pair is a named IdentityReducer case class, so structurally identical SPJ subtrees with value-equal reducers still compare equal and exchange/subquery reuse keeps deduplicating them.

Tailored for this branch in four places:

• GroupPartitionsExec.outputPartitioning keeps this branch's groupedPartitions/isGrouped reporting and its multi-KeyedPartitioning partition-keys assertion. The PartitionGrouping/isCollapsed refactor, which replaced both on master, is not on this branch.
• The config is spelled V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS here, the JOIN_ was dropped from the name later.
• The canonicalization test builds LocalTableScanExec with this branch's three-argument constructor.
• The subset-join-key test additionally sets REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION to false. SPARK-58558, which relaxed createKeyedShuffleSpec from an exact key match to a key-coverage check so the subset shape plans under the default, is not on this branch; the branch's other subset tests set the config the same way.

One of the three SPARK-59120 tests from #58335 is not carried over: SPARK-59120: reduced partition keys are read at the types they were built with was already omitted from the #58420 backport (#58431), because its failure mode runs through the sort that createShuffleSpec applies, which arrived with SPARK-59022 and is not on this branch. The other two SPARK-59120 tests are carried over verbatim; like on master, they replace the two #58431 tests and pin the combined behavior.

Why are the changes needed?

In a storage-partitioned join with compatible transforms whose result types differ (e.g. identity(id) on one side and bucket(N, id) on the other), the reducer maps the partition keys to the other side's value type. GroupPartitionsExec.outputPartitioning used to report the original expressions with the reduced keys, so the two had different data types and computing the key ordering threw:

java.lang.ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long

This fix covers the reducers where the reduced keys equal a single transform applied to one side (identity-vs-transform and single-side-transform). When both sides of a compatible-transform join reduce their keys, the reduced keys are not expressible as a single transform; that shape is a known gap tracked in SPARK-59121.

Does this PR introduce any user-facing change?

No by default. Under spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled, a storage-partitioned join whose reducer changes the partition key data type previously threw ClassCastException and now succeeds.

How was this patch tested?

Added regression tests in KeyGroupedPartitioningSuite, mirroring #58335. Measured failing on this branch at 0e37b68e742 (with #58431 on it):

• SPARK-59045: compatible transforms reduce multiple times fails with ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long - the second reduce's reducer is derived from the stale reported expression, which no type fix reaches.
• The two carried-over SPARK-59120 tests fail: the second-join one raises STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES (the behavior [SPARK-59120][SQL][4.2] Fix ClassCastException in a storage-partitioned join whose partition keys were reduced #58431 pinned on this branch), and the another-child one plans two shuffles instead of one. They now pin the combined behavior.
• The shipments extension of SPARK-56046: Reducers with same result types fails when the third table reduces onto the first join's stale reported expression and that reduce evaluates the expression (SCALAR_FUNCTION_NOT_FULLY_IMPLEMENTED).

Measured passing on that base, as on master: the identity-vs-bucket reducer, the subset-join-key, and the per-KeyedPartitioning retargeting tests - they pin this change's reporting, and the retargeting test regresses together with the multi-reduce test if the use-site re-targeting is dropped. The canonicalization test arrived with KeyReducer in this change. The first three also assert the storage-partitioned join introduces no shuffle. All tests pass here.

Ran KeyGroupedPartitioningSuite, GroupPartitionsExecSuite, EnsureRequirementsSuite, ProjectedOrderingAndPartitioningSuite, ShuffleSpecSuite, ValidateRequirementsSuite, and DistributionSuite.

Was this patch authored or co-authored using generative AI tooling?

Yes. Generated-by: Claude Code.

[ulysses-you]

Merge Summary:

• merged into branch-4.2 773f49a

Posted by merge_spark_pr.py