This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7f12c2da0b4e [SPARK-59120][SQL] Fix ClassCastException in a
storage-partitioned join whose partition keys were reduced
7f12c2da0b4e is described below
commit 7f12c2da0b4e419c9fa1bb8a059881ea54c55451
Author: Peter Toth <[email protected]>
AuthorDate: Mon Aug 31 12:03:57 2026 +0200
[SPARK-59120][SQL] Fix ClassCastException in a storage-partitioned join
whose partition keys were reduced
### What changes were proposed in this pull request?
`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 becau [...]
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 [...]
### 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` 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 [...]
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.
### Does this PR introduce _any_ user-facing change?
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, [...]
### How was this patch tested?
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 [...]
- `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.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #58420 from peter-toth/SPARK-59120-keyed-partitioning-key-data-types.
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 | 41 +++++++++-
.../datasources/v2/GroupPartitionsExec.scala | 2 +-
.../execution/datasources/v2/PushDownUtils.scala | 2 +-
.../execution/exchange/EnsureRequirements.scala | 9 +-
.../connector/KeyGroupedPartitioningSuite.scala | 95 ++++++++++++++++++++++
6 files changed, 192 insertions(+), 12 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 688968bda16b..6c585e043432 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
@@ -554,10 +554,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 =
- KeyedPartitioning.groupedKeyRowOrdering(expressionDataTypes)
+ KeyedPartitioning.groupedKeyRowOrdering(keyDataTypes)
@transient lazy val keyOrdering = keyRowOrdering.on((t:
InternalRowComparableWrapper) => t.row)
@@ -572,7 +613,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.
@@ -580,7 +621,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))
@@ -1434,7 +1475,13 @@ case class KeyedShuffleSpec(
partitioning.isGrouped &&
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 35f06a722f19..a76ba44b6803 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)
@@ -444,6 +445,44 @@ class ShuffleSpecSuite extends SparkFunSuite with
SQLHelper {
}
}
+ 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("SPARK-59120: createShuffleSpec sorts the projected keys at their
built-with types") {
+ val a = $"a".timestamp
+ val b = $"b".int
+ // A reducer left `(year, bucket)` values, both `IntegerType`, under
expressions that still
+ // declare `(TimestampType, IntegerType)`. Projecting to the `a` position
sorts them. An
+ // ordering built from the declared types reads the year as a Long and
throws.
+ val reduced =
+ KeyedPartitioning(Seq($"a".int, b), Seq(InternalRow(2021, 1),
InternalRow(2020, 0)))
+ .copy(expressions = Seq(a, b))
+
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key
-> "true") {
+ val spec = reduced.createShuffleSpec(ClusteredDistribution(Seq(a)))
+ .asInstanceOf[KeyedShuffleSpec]
+ assert(spec.joinKeyPositions === Some(Seq(0)))
+ assert(spec.partitioning.partitionKeys.map(_.row.getInt(0)) ===
Seq(2020, 2021))
+ }
+ }
+
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 6f671bfcb80b..22e7649886ed 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
@@ -136,7 +136,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/datasources/v2/PushDownUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
index fa330212370f..9a1d00a96a68 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
@@ -313,7 +313,7 @@ object PushDownUtils extends Logging {
val inputMap =
k.partitionKeys.groupBy(identity).view.mapValues(_.size)
val comparableKeyWrapperFactory = InternalRowComparableWrapper
- .getInternalRowComparableWrapperFactory(k.expressionDataTypes)
+ .getInternalRowComparableWrapperFactory(k.keyDataTypes)
val filteredMap = newPartitions.groupBy(
p =>
comparableKeyWrapperFactory(p.asInstanceOf[HasPartitionKey].partitionKey())
)
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 c2051ac91b06..36857539fb78 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
@@ -557,10 +557,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
@@ -572,9 +572,8 @@ case class EnsureRequirements(
rightReducedDataTypes = rightReducedDataTypes)
}
- val reducedKeyRowOrdering =
RowOrdering.createNaturalAscendingOrdering(reducedDataTypes)
- val reducedKeyOrdering =
- reducedKeyRowOrdering.on((t: InternalRowComparableWrapper) => t.row)
+ val reducedKeyOrdering =
KeyedPartitioning.groupedKeyRowOrdering(reducedDataTypes)
+ .on((t: InternalRowComparableWrapper) => t.row)
// merge values on both sides
var mergedPartitionKeys =
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 948de9995954..9c2aa8afab5e 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
@@ -4946,6 +4946,101 @@ class KeyGroupedPartitioningSuite
}
}
}
+
+ 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: reduced partition keys are read at the types they were
built with") {
+ // The join reduces the identity side onto the year key space, so its keys
become `IntegerType`
+ // years while the partitioning still reports `identity(ts)`, declaring
`TimestampType`. With
+ // the subset opt-in on, AQE re-runs `createShuffleSpec` on the already
reduced children through
+ // `ValidateRequirements`, which projects and sorts those keys. The
mechanism is in
+ // `ShuffleSpecSuite`'s "createShuffleSpec sorts the projected keys at
their built-with types".
+ withTable("t_identity", "t_years") {
+ createTsTable("t_identity", Array(identity("ts")))
+ createTsTable("t_years", Array(years("ts")))
+
+ val query =
+ s"""${selectWithMergeJoinHint("a", "b")} a.id, b.data
+ |FROM testcat.ns.t_identity a JOIN testcat.ns.t_years b ON a.ts =
b.ts
+ |""".stripMargin
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true") {
+ val df = sql(query)
+ checkAnswer(df, Seq(Row(1, "aa"), Row(2, "bb")))
+ assert(collectShuffles(df.queryExecution.executedPlan).isEmpty,
+ "the two sides are reduced onto one key space, so they stay
co-partitioned")
+ }
+ }
+ }
+
+ 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]
