This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 483b0211bad2 [SPARK-59120][4.2][SQL] Fix ClassCastException in a
storage-partitioned join whose partition keys were reduced
483b0211bad2 is described below
commit 483b0211bad21402b5ca14d8df8960ff1bfc046c
Author: Peter Toth <[email protected]>
AuthorDate: Mon Aug 31 17:34:29 2026 +0200
[SPARK-59120][4.2][SQL] Fix ClassCastException in a storage-partitioned
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` fal [...]
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 <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
---
.../sql/catalyst/plans/physical/partitioning.scala | 55 ++++++++++++++++--
.../spark/sql/catalyst/ShuffleSpecSuite.scala | 23 +++++++-
.../datasources/v2/GroupPartitionsExec.scala | 2 +-
.../execution/exchange/EnsureRequirements.scala | 4 +-
.../connector/KeyGroupedPartitioningSuite.scala | 67 ++++++++++++++++++++++
5 files changed, 143 insertions(+), 8 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
index 6fb2ae3821e0..b53648bfd611 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
@@ -437,10 +437,51 @@ case class KeyedPartitioning(
newChildren: IndexedSeq[Expression]): KeyedPartitioning =
copy(expressions = newChildren)
+ /** Need not be what the `partitionKeys` rows hold. See `keyDataTypes`. */
@transient lazy val expressionDataTypes: Seq[DataType] =
expressions.map(_.dataType)
+ /**
+ * The types the `partitionKeys` rows were built with. Anything reading
those rows should take its
+ * types from here. It is a driver-side value, since `partitionKeys` is
`@transient`.
+ *
+ * They are the `expressionDataTypes` unless a reducer has rewritten the
keys. With
+ * `v2BucketingAllowCompatibleTransforms` a storage-partitioned join reduces
one or both sides'
+ * keys onto a common key space, while 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.
With no key at all the
+ * expressions are all there is, and there is no row to read or to place.
+ *
+ * `ShuffleExchangeExec` is the one reader that stays on
`expressionDataTypes`. It evaluates the
+ * expressions to place the other child's rows, and it runs on executors,
where this value is not
+ * available. `expressionsDescribeKeyShape` is what keeps that site sound.
+ *
+ * Only the first key's types are read, and nothing enforces that the rest
match. Reduced keys
+ * also break three readers in ways no choice of types can fix, and
SPARK-59121 covers all three:
+ *
+ * - `EnsureRequirements`' `OrderedDistribution` arm orders these rows by
the partition
+ * attributes.
+ * - the `UnionExec` key merge concatenates several children's keys.
+ * - `KeyedShuffleSpec.reducers` binds its reducer to the partition
expression, then reads a
+ * stored key with it.
+ */
+ @transient lazy val keyDataTypes: Seq[DataType] =
+ partitionKeys.headOption.map(_.dataTypes).getOrElse(expressionDataTypes)
+
+ /**
+ * Whether the `partitionKeys` rows still read the same at
`expressionDataTypes`, which is what
+ * `ShuffleExchangeExec` declares them at.
+ *
+ * Shapes rather than plain equality, the test `HashJoin` applies to its
join key types.
+ * `KeyedShuffleSpec.createPartitioning` puts the other child's expressions
over these keys, so a
+ * struct field name can differ here with no reducer involved.
+ */
+ @transient lazy val expressionsDescribeKeyShape: Boolean =
+ keyDataTypes.corresponds(expressionDataTypes)(
+ DataType.equalsStructurally(_, _, ignoreNullability = true))
+
+ /** Driver-side, like the `keyDataTypes` it comes from. */
@transient lazy val keyRowOrdering =
- RowOrdering.createNaturalAscendingOrdering(expressionDataTypes)
+ RowOrdering.createNaturalAscendingOrdering(keyDataTypes)
@transient lazy val keyOrdering = keyRowOrdering.on((t:
InternalRowComparableWrapper) => t.row)
@@ -455,7 +496,7 @@ case class KeyedPartitioning(
* Returns the projected expressions and their data types together with the
projected keys.
*/
def projectKeys(positions: Seq[Int]): (Seq[DataType],
Seq[InternalRowComparableWrapper]) =
- KeyedPartitioning.projectKeys(partitionKeys, expressionDataTypes,
positions)
+ KeyedPartitioning.projectKeys(partitionKeys, keyDataTypes, positions)
/**
* Reduces this partitioning's partition keys by applying the given reducers.
@@ -463,7 +504,7 @@ case class KeyedPartitioning(
*/
def reduceKeys(
reducers: Seq[Option[Reducer[_, _]]]): (Seq[DataType],
Seq[InternalRowComparableWrapper]) =
- KeyedPartitioning.reduceKeys(partitionKeys, expressionDataTypes, reducers)
+ KeyedPartitioning.reduceKeys(partitionKeys, keyDataTypes, reducers)
override def satisfies0(required: Distribution): Boolean = {
nonGroupedSatisfies(required) || (isGrouped && groupedSatisfies(required))
@@ -1070,7 +1111,13 @@ case class KeyedShuffleSpec(
!SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled &&
partitioning.expressions.forall { e =>
e.isInstanceOf[AttributeReference] ||
e.isInstanceOf[TransformExpression]
- }
+ } &&
+ // A reducer leaves keys that the expressions no longer describe, and
`ShuffleExchangeExec`
+ // would then compare the two at different types as it builds the
shuffle's key map. Matching
+ // shapes are only a proxy for that. `bucket(12)` and `bucket(8)`
reducing onto `bucket(4)`
+ // keep the type, pass here, and still misroute rows. SPARK-59045 and
SPARK-59121 add the real
+ // test.
+ partitioning.expressionsDescribeKeyShape
override def createPartitioning(clustering: Seq[Expression]): Partitioning =
{
assert(clustering.size == distribution.clustering.size,
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala
index 85d285aa76c0..384c2bb05bca 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala
@@ -19,10 +19,11 @@ package org.apache.spark.sql.catalyst
import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException}
import org.apache.spark.sql.catalyst.dsl.expressions._
-import org.apache.spark.sql.catalyst.expressions.DirectShufflePartitionID
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
DirectShufflePartitionID}
import org.apache.spark.sql.catalyst.plans.SQLHelper
import org.apache.spark.sql.catalyst.plans.physical._
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, StructType}
class ShuffleSpecSuite extends SparkFunSuite with SQLHelper {
private val passThrough_a_10 =
ShufflePartitionIdPassThrough(DirectShufflePartitionID($"a"), 10)
@@ -424,6 +425,26 @@ class ShuffleSpecSuite extends SparkFunSuite with
SQLHelper {
assert(!RangeShuffleSpec(10, distribution).canCreatePartitioning)
}
+ test("SPARK-59120: canCreatePartitioning: KeyedShuffleSpec requires the
declared key shape") {
+ // A partitioning whose keys were built with types the expressions no
longer declare, which is
+ // what a reducer leaves behind.
+ def spec(builtWith: Attribute, declared: Attribute, key: InternalRow):
KeyedShuffleSpec =
+ KeyedShuffleSpec(
+ KeyedPartitioning(Seq(builtWith), Seq(key)).copy(expressions =
Seq(declared)),
+ ClusteredDistribution(Seq(declared)))
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ assert(!spec($"a".int, $"a".timestamp,
InternalRow(2020)).canCreatePartitioning,
+ "`IntegerType` keys cannot be looked up by evaluating a
`TimestampType` expression")
+
+ // The two sides can name a struct field differently with no reducer
involved, see
+ // `expressionsDescribeKeyShape`.
+ def struct(field: String): Attribute = $"a".struct(new
StructType().add(field, IntegerType))
+ assert(spec(struct("a"), struct("b"),
InternalRow(InternalRow(1))).canCreatePartitioning,
+ "a key row reads the same either way, so the field names must not
matter")
+ }
+ }
+
test("createPartitioning: HashShuffleSpec") {
checkCreatePartitioning(
HashShuffleSpec(HashPartitioning(Seq($"a"), 10),
ClusteredDistribution(Seq($"a", $"b"))),
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
index 57951a228531..494fb38a305e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
@@ -144,7 +144,7 @@ case class GroupPartitionsExec(
// Project partition keys if join key positions are specified
val (projectedDataTypes, projectedKeys) =
joinKeyPositions.fold(
- (keyedPartitioning.expressionDataTypes,
keyedPartitioning.partitionKeys)
+ (keyedPartitioning.keyDataTypes, keyedPartitioning.partitionKeys)
)(keyedPartitioning.projectKeys)
// Reduce keys if reducers are specified
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
index aefc947d6c63..b5114aa0be90 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
@@ -552,10 +552,10 @@ case class EnsureRequirements(
val leftReducers = leftSpec.reducers(rightSpec)
val rightReducers = rightSpec.reducers(leftSpec)
val (leftReducedDataTypes, leftReducedKeys) = leftReducers.fold(
- (leftPartitioning.expressionDataTypes,
leftPartitioning.partitionKeys)
+ (leftPartitioning.keyDataTypes, leftPartitioning.partitionKeys)
)(leftPartitioning.reduceKeys)
val (rightReducedDataTypes, rightReducedKeys) = rightReducers.fold(
- (rightPartitioning.expressionDataTypes,
rightPartitioning.partitionKeys)
+ (rightPartitioning.keyDataTypes, rightPartitioning.partitionKeys)
)(rightPartitioning.reduceKeys)
val reducedDataTypes = if (leftReducedDataTypes ==
rightReducedDataTypes) {
leftReducedDataTypes
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index b7094731bb95..e9dc7ee56e06 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -4195,4 +4195,71 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
}
+
+ private def createTsTable(name: String, partitions: Array[Transform]): Unit
= {
+ createTable(name, columns, partitions)
+ sql(s"INSERT INTO testcat.ns.$name VALUES " +
+ s"(1, 'aa', cast('2020-01-01' as timestamp)), " +
+ s"(2, 'bb', cast('2021-06-01' as timestamp))")
+ }
+
+ test("SPARK-59120: incompatible reduced key types are reported instead of
cast") {
+ // The first join reduces the identity side onto the year key space. The
third table does not
+ // reduce at all, so the two key spaces genuinely differ, and reading each
side's keys at their
+ // own types is what lets `EnsureRequirements` say so. On `master` the
merge cast one to the
+ // other and threw `ClassCastException`.
+ withTable("t_identity", "t_years", "t_identity2") {
+ createTsTable("t_identity", Array(identity("ts")))
+ createTsTable("t_years", Array(years("ts")))
+ createTsTable("t_identity2", Array(identity("ts")))
+
+ val query =
+ """SELECT /*+ MERGE(a, b), MERGE(a, c) */ a.id, c.data
+ |FROM testcat.ns.t_identity a JOIN testcat.ns.t_years b ON a.ts =
b.ts
+ |JOIN testcat.ns.t_identity2 c ON a.ts = c.ts
+ |""".stripMargin
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ checkError(
+ exception = intercept[SparkException](sql(query).collect()),
+ condition = "STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES",
+ parameters = Map(
+ "leftReducers" -> "[]",
+ "leftReducedDataTypes" -> "[\"INT\"]",
+ "rightReducers" -> "[]",
+ "rightReducedDataTypes" -> "[\"TIMESTAMP\"]"))
+ }
+ }
+ }
+
+ test("SPARK-59120: another child is not shuffled onto reducer-rewritten
keys") {
+ // `canCreatePartitioning` refuses the reduced side as a layout for the
second join, so both of
+ // that join's sides are shuffled. Without the gate the driver dies while
assembling that
+ // shuffle's key map, where the stored keys are re-wrapped at the
expressions' types.
+ withTable("t_identity", "t_years", "t_plain") {
+ createTsTable("t_identity", Array(identity("ts")))
+ createTsTable("t_years", Array(years("ts")))
+ createTsTable("t_plain", Array.empty[Transform])
+
+ val query =
+ """SELECT /*+ MERGE(a, b), MERGE(c) */ a.id, c.data
+ |FROM testcat.ns.t_identity a JOIN testcat.ns.t_years b ON a.ts =
b.ts
+ |JOIN testcat.ns.t_plain c ON a.ts = c.ts
+ |""".stripMargin
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ val df = sql(query)
+ checkAnswer(df, Seq(Row(1, "aa"), Row(2, "bb")))
+ assert(collectAllShuffles(df.queryExecution.executedPlan).size == 2,
+ "the second join shuffles both of its sides")
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
