Uh oh!
There was an error while loading. Please reload this page.
[SPARK-59120][SQL][4.2] Fix ClassCastException in a storage-partitioned join whose partition keys were reduced - #58431
Closed
peter-toth wants to merge 1 commit into
Conversation
…in whose partition keys were reduced `KeyedPartitioning` gains `keyDataTypes`, the schema each `partitionKeys` row was actually written under, and everything that reads those rows takes its types from there instead of from the partition expressions. The two can disagree. With `v2BucketingAllowCompatibleTransforms` a storage-partitioned join reduces one or both sides' keys onto a common key space, and the partitioning keeps reporting the expressions it was built from. Joining an `identity(ts)`-partitioned table to a `years(ts)`-partitioned one leaves `IntegerType` year values under a `TimestampType`-declared expression. A row is only readable at its own schema, so the wrappers are the authority: transient lazy val keyDataTypes: Seq[DataType] = partitionKeys.headOption.map(_.dataTypes).getOrElse(expressionDataTypes) Switched to it: `keyRowOrdering` (and with it `toGrouped` and `createShuffleSpec`'s subset projection), the instance-level `projectKeys` and `reduceKeys`, `GroupPartitionsExec`'s projection base types, and the reduce path in `EnsureRequirements`. `PushDownUtils` changes basis twice, at its wrapper factory and through `keyRowOrdering` at its sort, and both are no-ops today: it handles rows a scan has just reported, and a scan's own partitioning is never reduced. They are switched because those wrappers are compared against the stored keys, which the switch keeps them comparable with. One drive-by in the same reduce path: it built its ordering with `RowOrdering.createNaturalAscendingOrdering` directly, and now calls `KeyedPartitioning.groupedKeyRowOrdering`, which is where that order is defined. The expressions' own types stay right in one place. `ShuffleExchangeExec` builds a `KeyGroupedPartitioner` whose lookup keys come from evaluating the expressions per row, so both sides of that comparison are declared at `expressionDataTypes`. That is also what the second change is about. A reduced partitioning must not be the layout another child is shuffled onto, whatever types it declares, so `KeyedShuffleSpec.canCreatePartitioning` refuses the ones it can detect. The question gets a name next to the two values it relates: transient lazy val expressionsDescribeKeyShape: Boolean = keyDataTypes.corresponds(expressionDataTypes)( DataType.equalsStructurally(_, _, ignoreNullability = true)) Shapes rather than plain equality, which is the test `HashJoin` already applies to its join key types. `createPartitioning` puts the other child's expressions over this side's keys, so a struct key whose two sides name the field differently reaches the gate with no reducer involved, and refusing it would cost a shuffle for nothing. The shape test is only a proxy for the real question, and the comment says so. A reduction that keeps the type, `bucket(12)` and `bucket(8)` both reducing onto `bucket(4)`, passes it and still misroutes rows. SPARK-59045 and SPARK-59121 add the real test. Two of the switched sites are inert today, for the same reason as the `PushDownUtils` no-op: `reduceKeys` and `GroupPartitionsExec`'s projection only see a reduced partitioning when a second reduce lands on a first, which still fails at planning either way, in `KeyedShuffleSpec.reducers`. That reducer is bound to the partition expression and then reads a stored key, so it is a fourth reader of these rows that types cannot fix. It is named in the `keyDataTypes` scaladoc and belongs to SPARK-59121. Both halves fail a query on the `v2BucketingAllowCompatibleTransforms` path. Reading a reduced key at the expression's type unboxes an Integer as a Long, which throws `ClassCastException` at planning. AQE reaches it on its own: `ValidateRequirements` re-runs `createShuffleSpec` on the already reduced children, and with the subset opt-in on that projects and sorts the keys. The first test below is such a query. The second half is independent of the first, and it fails at execution rather than at planning. A reduced partitioning is grouped and does satisfy the distribution, so `EnsureRequirements` offers it as the layout to shuffle another child onto. `ShuffleExchangeExec` then builds the shuffle's key map by re-wrapping the stored keys at the expressions' types, and the same unboxing throws, on the driver as the shuffle is prepared. The gate makes `EnsureRequirements` fall back to shuffling both sides, which returns the right rows. The `EnsureRequirements` reduce path is the third behaviour change. Its `fold` default now reports the keys' own types, so two sides whose reduced key spaces genuinely differ raise `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` where `master` cast one to the other and threw `ClassCastException` in the merge. What the gate does not fix is a reduction that keeps the type. There `KeyGroupedPartitioner` sends a key it cannot find to `hashCode % numPartitions`, and rows are lost silently, both before and after this change. `bucket(12)` joined to `bucket(8)`, then joined to an unpartitioned table, returns 8 of 12 rows on `master` and still does here. The reduce and the reporting of reduced keys under the original expressions arrived in SPARK-47094 (4.0.0). The crashes start in 4.2.0, with the `KeyedPartitioning` / `GroupPartitionsExec` refactor that derives the key ordering and types from the reported expressions. Yes, on the opt-in `v2BucketingAllowCompatibleTransforms` path. Queries that failed with `ClassCastException` at planning, as the shuffle was prepared, or in the shuffle write, now plan and return rows. One error changes: two sides with genuinely incompatible reduced types now say so instead of throwing `ClassCastException`. A connector whose transforms reduce onto a type-compatible key space could previously lose rows to the partitioner's hash fallback and now gets a shuffle instead, so its results change from wrong to right. Five new tests, all five failing on `master`: - `KeyGroupedPartitioningSuite`: `reduced partition keys are read at the types they were built with`, which throws at planning on `master`, in `toGrouped` under `createShuffleSpec` under `ValidateRequirements`. `another child is not shuffled onto reducer-rewritten keys`, which throws on `master` as the shuffle is prepared. And `incompatible reduced key types are reported instead of cast`, which throws `ClassCastException` on `master` and the named error here. The first two also assert the plan the fix is about: no shuffle for the co-partitioned pair, two shuffles for the child that must not be laid out on reduced keys. - `ShuffleSpecSuite`: `createShuffleSpec sorts the projected keys at their built-with types`, the unit-level form of the first one, same exception on `master`. And `canCreatePartitioning: KeyedShuffleSpec requires the declared key shape`, which asserts both directions, the reduced keys refused and the differently named struct fields allowed. The two changes are pinned separately by ablation. Removing the `canCreatePartitioning` clause leaves the first end-to-end test passing and makes the second one throw again, so neither test stands in for the other. Tightening the shape test to plain equality fails the allowed direction alone. 291 tests green across `KeyGroupedPartitioningSuite`, `KeyGroupedPartitioningCatalystRuntimeFilterSuite`, `EnsureRequirementsSuite`, `ValidateRequirementsSuite`, `ProjectedOrderingAndPartitioningSuite`, `PlannerSuite`, `ShuffleSpecSuite` and `DistributionSuite`. `dev/lint-scala` clean. Generated-by: Claude Code Closesapache#58420 from peter-toth/SPARK-59120-keyed-partitioning-key-data-types. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Peter Toth <peter.toth@gmail.com> (cherry picked from commit 7f12c2d)
peter-toth
commented
Aug 31, 2026
ContributorAuthor
ulysses-you
approved these changes
Aug 31, 2026
peter-toth added a commit
that referenced
this pull request
Aug 31, 2026
…ed join whose partition keys were reduced ### What changes were proposed in this pull request? Backport of #58420 to `branch-4.2`. `KeyedPartitioning` gains `keyDataTypes`, the schema each `partitionKeys` row was actually written under, and everything that reads those rows takes its types from there instead of from the partition expressions. `KeyedShuffleSpec.canCreatePartitioning` additionally refuses a partitioning whose expressions no longer describe the shape of its keys, so another child is not shuffled onto a reducer-rewritten layout. Tailored for this branch in four places: - `KeyedPartitioning.keyRowOrdering` and the reduce path in `EnsureRequirements` keep building their ordering with `RowOrdering.createNaturalAscendingOrdering`. `KeyedPartitioning.groupedKeyRowOrdering`, which `master` calls there, arrived with SPARK-59027 and is not on this branch. - The `PushDownUtils` hunk is dropped. `replanWithRuntimeFilters` and the runtime-filter SPJ path it belongs to are not on this branch, and on `master` that hunk was a no-op anyway. - The config is spelled `V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS` here, the `JOIN_` was dropped from the name later. - Two of the five tests are omitted, see the testing section. ### Why are the changes needed? Both halves fail a query on the `v2BucketingAllowCompatibleTransforms` path. Reading a reduced key at the expression's type unboxes an Integer as a Long, which throws `ClassCastException`. A reduced partitioning is also grouped and does satisfy the distribution, so `EnsureRequirements` offers it as the layout to shuffle another child onto. `ShuffleExchangeExec` then builds the shuffle's key map by re-wrapping the stored keys at the expressions' types, and the same unboxing throws on the driver as the shuffle is prepared. The gate makes `EnsureRequirements` fall back to shuffling both sides, which returns the right rows. The `EnsureRequirements` reduce path is the third behaviour change. Its `fold` default now reports the keys' own types, so two sides whose reduced key spaces genuinely differ raise `STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES` where this branch cast one to the other and threw `ClassCastException` in the merge. What the gate does not fix is a reduction that keeps the type. There `KeyGroupedPartitioner` sends a key it cannot find to `hashCode % numPartitions`, and rows are lost silently, both before and after this change. That half is tracked in SPARK-59121. The reduce and the reporting of reduced keys under the original expressions arrived in SPARK-47094 (4.0.0). The crashes start in 4.2.0, with the `KeyedPartitioning` / `GroupPartitionsExec` refactor that derives the key ordering and types from the reported expressions, which is why this branch is affected. ### Does this PR introduce _any_ user-facing change? Yes, on the opt-in `v2BucketingAllowCompatibleTransforms` path. Queries that failed with `ClassCastException` as the shuffle was prepared now plan and return rows. One error changes: two sides with genuinely incompatible reduced types now say so instead of throwing `ClassCastException`. ### How was this patch tested? Three tests, each measured failing on `branch-4.2` at `149b29aac1d` and passing here: - `SPARK-59120: canCreatePartitioning: KeyedShuffleSpec requires the declared key shape` - `SPARK-59120: incompatible reduced key types are reported instead of cast`, `ClassCastException` on the base - `SPARK-59120: another child is not shuffled onto reducer-rewritten keys`, `ClassCastException` on the base Two tests from #58420 are omitted rather than carried over green. Both were measured **passing** on this branch's base, so they would pin nothing here: `createShuffleSpec sorts the projected keys at their built-with types` and `reduced partition keys are read at the types they were built with`. Both reach the crash through the sort that `createShuffleSpec` applies on `master`, and that sort arrived with SPARK-59022, which is not on this branch. 171 tests green across `ShuffleSpecSuite`, `KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`, `EnsureRequirementsSuite`, `ValidateRequirementsSuite` and `ProjectedOrderingAndPartitioningSuite`. `dev/lint-scala` clean. ### Was this patch authored or co-authored using generative AI tooling? Yes. Generated-by: Claude Code. Closes#58431 from peter-toth/SPARK-59120-keyed-partitioning-key-data-types-4.2. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Peter Toth <peter.toth@gmail.com>
peter-toth
commented
Aug 31, 2026
ContributorAuthor
Merge Summary:
Posted by |
peter-toth
commented
Aug 31, 2026
ContributorAuthor
Thank you @ulysses-you for the review. |
ulysses-you added a commit
that referenced
this pull request
Sep 2, 2026
…es partition key data type ### 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 #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. Closes#58451 from ulysses-you/spj-reducer-4.2. Authored-by: Xiduo You <ulyssesyou18@gmail.com> Signed-off-by: Xiduo You <ulyssesyou@apache.org>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Backport of #58420 to
branch-4.2.KeyedPartitioninggainskeyDataTypes, the schema eachpartitionKeysrow was actually written under, and everything that reads those rows takes its types from there instead of from the partition expressions.KeyedShuffleSpec.canCreatePartitioningadditionally refuses a partitioning whose expressions no longer describe the shape of its keys, so another child is not shuffled onto a reducer-rewritten layout.Tailored for this branch in four places:
KeyedPartitioning.keyRowOrderingand the reduce path inEnsureRequirementskeep building their ordering withRowOrdering.createNaturalAscendingOrdering.KeyedPartitioning.groupedKeyRowOrdering, whichmastercalls there, arrived with SPARK-59027 and is not on this branch.PushDownUtilshunk is dropped.replanWithRuntimeFiltersand the runtime-filter SPJ path it belongs to are not on this branch, and onmasterthat hunk was a no-op anyway.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYShere, theJOIN_was dropped from the name later.Why are the changes needed?
Both halves fail a query on the
v2BucketingAllowCompatibleTransformspath.Reading a reduced key at the expression's type unboxes an Integer as a Long, which throws
ClassCastException. A reduced partitioning is also grouped and does satisfy the distribution, soEnsureRequirementsoffers it as the layout to shuffle another child onto.ShuffleExchangeExecthen builds the shuffle's key map by re-wrapping the stored keys at the expressions' types, and the same unboxing throws on the driver as the shuffle is prepared. The gate makesEnsureRequirementsfall back to shuffling both sides, which returns the right rows.The
EnsureRequirementsreduce path is the third behaviour change. Itsfolddefault now reports the keys' own types, so two sides whose reduced key spaces genuinely differ raiseSTORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPESwhere this branch cast one to the other and threwClassCastExceptionin the merge.What the gate does not fix is a reduction that keeps the type. There
KeyGroupedPartitionersends a key it cannot find tohashCode % numPartitions, and rows are lost silently, both before and after this change. That half is tracked in SPARK-59121.The reduce and the reporting of reduced keys under the original expressions arrived in SPARK-47094 (4.0.0). The crashes start in 4.2.0, with the
KeyedPartitioning/GroupPartitionsExecrefactor that derives the key ordering and types from the reported expressions, which is why this branch is affected.Does this PR introduce any user-facing change?
Yes, on the opt-in
v2BucketingAllowCompatibleTransformspath. Queries that failed withClassCastExceptionas the shuffle was prepared now plan and return rows. One error changes: two sides with genuinely incompatible reduced types now say so instead of throwingClassCastException.How was this patch tested?
Three tests, each measured failing on
branch-4.2at149b29aac1dand passing here:SPARK-59120: canCreatePartitioning: KeyedShuffleSpec requires the declared key shapeSPARK-59120: incompatible reduced key types are reported instead of cast,ClassCastExceptionon the baseSPARK-59120: another child is not shuffled onto reducer-rewritten keys,ClassCastExceptionon the baseTwo tests from #58420 are omitted rather than carried over green. Both were measured passing on this branch's base, so they would pin nothing here:
createShuffleSpec sorts the projected keys at their built-with typesandreduced partition keys are read at the types they were built with. Both reach the crash through the sort thatcreateShuffleSpecapplies onmaster, and that sort arrived with SPARK-59022, which is not on this branch.171 tests green across
ShuffleSpecSuite,KeyGroupedPartitioningSuite,GroupPartitionsExecSuite,EnsureRequirementsSuite,ValidateRequirementsSuiteandProjectedOrderingAndPartitioningSuite.dev/lint-scalaclean.Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: Claude Code.