dongjoon-hyun commented on code in PR #58501:
URL: https://github.com/apache/spark/pull/58501#discussion_r3935218535
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -96,7 +96,8 @@ case class GroupPartitionsExec(
case None => projectedExpressions
}
KeyedPartitioning(
- effectiveExpressions, partitionKeys, grouping.isGrouped,
grouping.isCollapsed)
+ effectiveExpressions, partitionKeys, grouping.keyDataTypes,
grouping.isGrouped,
Review Comment:
This looks like a regression. `grouping` takes `keyDataTypes` from the
member found by `collectFirst` (line ~177) and this line stamps it on every
member, while `EnsureRequirements.createKeyedShuffleSpec` builds
`expectedPartitionKeys` from the first member that `satisfies` the
distribution. Members of a `PartitioningCollection` with empty key lists can
carry different `keyDataTypes` (nothing checks or normalizes them, see my
comment on `checkKeyedPartitioningInvariant`), so the two can disagree and the
new constructor `require` fires.
Reachable shape, with `pushPartValues`, `partitionFilter` and
`allowCompatibleTransforms` on:
1. `leg1 = t1 JOIN t2`, both identity-partitioned on `a: string` with
disjoint keys -> partition filter intersects to nothing, two members `KP(a, [],
[String])`.
2. `leg2 = t3 JOIN t4`, both `bucket(4, b: string)`, disjoint -> two members
`KP(bucket(4,b), [], [Int])`.
3. `leg1 JOIN leg2 ON a = b`: `KeyedShuffleSpec.isCompatibleWith` is true
(`Nil == Nil`, `numPartitions 0 == 0`, attribute vs transform is compatible via
`canReduceKeys`), so the push-down branch and its type check are skipped and
`fromPartitionings` produces a collection mixing `[String]` and `[Int]`.
4. `... FULL OUTER JOIN t5(bucket(4, c)) ON b = c`: `EnsureRequirements`
picks the `bucket(4,b)` member, merged keys are `Int` rows. `grouping` picks
member `a`, so `reducedDataTypes = [String]`, and this line builds
`KeyedPartitioning(bucket(4,b), <Int keys>, [String], ...)` ->
`IllegalArgumentException` from the `require` at partitioning.scala:603.
Before this PR the types were read from the key rows, so the query ran. The
inner-join variant passes the `require` (merged keys are empty) but carries
`[String]` on the bucket member, and a later push-down join then throws
`STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1074,6 +1093,12 @@ case class PartitioningCollection(partitionings:
Seq[Partitioning])
* unmarked one. They cannot disagree. The members share one key list, so
they describe one
* reduce, and a reduce that marks one side's expressions marks the other's
in the same step,
* while a one-side reduce marks neither.
+ *
+ * `keyDataTypes` is not in it either, for a different reason. The members
share one key list, and
Review Comment:
I don't think this premise holds. An empty-key member's `keyDataTypes` is
read: by `GroupPartitionsExec.grouping` (via `collectFirst`), by the
reduced-types comparison in `EnsureRequirements`, and by `PushDownUtils`. Those
readers pick a member by different rules (`collectFirst` vs. the first member
that `satisfies`), so if members disagree the answer depends on which one is
consulted. That is what produces the `GroupPartitionsExec` failure I described
above.
A one-line `require(rep.keyDataTypes == first.keyDataTypes, ...)` next to
the existing `isCollapsed` check (same O(members) cost) would make this
structural, or `fromPartitionings` could normalize the field the way it interns
`partitionKeys`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean) extends Expression with Partitioning with
Unevaluable {
override val numPartitions = partitionKeys.length
+ // The keys carry their own types, so the field can be checked against the
thing it describes
+ // rather than argued about. One key is enough: `concat` is the only copy
that mixes rows from
+ // several partitionings, and it checks all of them.
+ require(keyDataTypes.length == expressions.length,
+ "A KeyedPartitioning must have one key data type per partition expression")
+ require(partitionKeys.headOption.forall(_.dataTypes == keyDataTypes),
+ "A KeyedPartitioning's keyDataTypes must be the types its partitionKeys
were built with")
+
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Drops the `keyDataTypes`, so that `explain` shows what it showed before
the field existed. They
+ * are the types of the keys printed beside them, which adds nothing a
reader of a plan wants.
+ */
+ override protected def stringArgs: Iterator[Any] =
Review Comment:
nit: the justification does not hold. `InternalRowComparableWrapper` has no
`toString`, so the keys print as `InternalRowComparableWrapper@<hash>` and no
type is legible beside them. The hide is also asymmetric: `TreeNode.jsonFields`
/ `asCode` use `productIterator`, so the field shows up in `toJSON` but not in
`explain`.
The case this PR exists for (a marked `days(...)` expression of `DateType`
over `LongType` keys, possibly with no key at all) is exactly where `explain`
would show a misleading expression type with no way to see the real one, and
two partitionings that differ only in `keyDataTypes` print identically in
`require`/`assert` messages. No golden file or test asserts this string, so I
would drop the override. If it is kept for output stability, a one-line comment
saying so would be clearer than arguing the information is worthless.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -572,6 +572,14 @@ case class CoalescedNullAwareHashPartitioning(
* comparison and grouping. One per partition. Typically
in sorted order when
* produced by a data source or `GroupPartitionsExec`,
but this is not
* guaranteed after projection. May contain duplicates
when ungrouped.
+ * @param keyDataTypes The types the `partitionKeys` rows were built with, one
per expression.
Review Comment:
nit: this paragraph, the constructor comment and the `concat` scaladoc each
inventory the copy sites (`project`, `concat`, `toGrouped`,
`fromPartitionings`) and argue the design. The lists will silently rot at the
next `copy(partitionKeys = ...)` (`GroupPartitionsExec` already builds one
directly). I'd keep the contract only, e.g. "The types the `partitionKeys` rows
were built with, one per expression; kept even when there is no key row.", and
on `concat`: "Children must agree on `keyDataTypes`; the constructor only
checks the first key."
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean) extends Expression with Partitioning with
Unevaluable {
override val numPartitions = partitionKeys.length
+ // The keys carry their own types, so the field can be checked against the
thing it describes
+ // rather than argued about. One key is enough: `concat` is the only copy
that mixes rows from
+ // several partitionings, and it checks all of them.
+ require(keyDataTypes.length == expressions.length,
+ "A KeyedPartitioning must have one key data type per partition expression")
+ require(partitionKeys.headOption.forall(_.dataTypes == keyDataTypes),
+ "A KeyedPartitioning's keyDataTypes must be the types its partitionKeys
were built with")
+
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Drops the `keyDataTypes`, so that `explain` shows what it showed before
the field existed. They
+ * are the types of the keys printed beside them, which adds nothing a
reader of a plan wants.
+ */
+ override protected def stringArgs: Iterator[Any] =
+ Iterator(expressions, partitionKeys, isGrouped, isCollapsed)
+
override protected def withNewChildrenInternal(
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 differ from the `expressionDataTypes` in two cases. A join that
reduced both sides' keys
- * onto a key space no transform names leaves a marked expression whose type
can be anything, see
- * `expressionsDescribeKeys`. A one-side reduce keeps them equal, because
the expression the
- * partitioning then reports is the target transform and
`EnsureRequirements` refuses a reducer
- * whose result type disagrees with it.
`KeyedShuffleSpec.createPartitioning` is the other case.
- * It puts the other child's expressions over these keys with no reducer in
sight, so a struct
- * field can be named differently on the two sides. With no key at all the
expressions are all
- * there is, and there is no row to read or to place.
+ * The types the partition expressions produce. Not what the `partitionKeys`
rows hold, whenever
+ * the expressions have stopped describing the keys, which happens in two
ways.
Review Comment:
nit: this reads as if `KeyedShuffleSpec.createPartitioning` were a case
where the expressions have stopped describing the keys, but it only does
`partitioning.copy(expressions = newExpressions)` and sets no marker, so
`expressionsDescribeKeys` stays true there and the last paragraph's
"`expressionsDescribeKeys` is what keeps them sound" only covers the reduce
case. Something like: "May differ from `keyDataTypes` in two cases: (a) a
both-sides reduce marks the expressions (`expressionsDescribeKeys`); (b)
`KeyedShuffleSpec.createPartitioning` re-targets the expressions at the other
child's attributes, so struct field names can differ while the expressions
still describe the keys."
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
Review Comment:
Optional, for consideration: the (types, keys) pair now appears as two
constructor args here, two tuple-returning helpers (`projectKeys`,
`reduceKeys`), two `PartitionGrouping` fields and the `fold` seeds in
`EnsureRequirements`, and the pairing is guaranteed only by a head-key
`require` that is vacuous when the key list is empty. A small value object (say
`TypedKeys(dataTypes, keys)`) returned by `projectKeys`/`reduceKeys` and held
here and in `PartitionGrouping` would make the pairing structural and is the
deeper fix for both issues above. Fine as a follow-up if you prefer to keep
this PR small.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -835,9 +847,16 @@ object KeyedPartitioning {
*
* Keys repeating across children is not a collapse. Only a child's own
collapse carries over,
* since such a key still stands for several finer-grained ones in the
concatenation.
+ *
+ * This is the one place that mixes key rows from several partitionings, so
it is the one place
+ * where the children's `keyDataTypes` have to be checked rather than
carried. The caller compares
+ * the children's expressions, and equal expressions do not by themselves
mean equal key types: a
+ * reduce leaves a partitioning whose keys are typed by the reducer.
*/
def concat(kps: Seq[KeyedPartitioning]): KeyedPartitioning = {
val concatenatedKeys = kps.flatMap(_.partitionKeys)
+ require(kps.forall(_.keyDataTypes == kps.head.keyDataTypes),
Review Comment:
This `require` is reachable from `UnionExec.comparePartitioning`, the only
caller, which compares children by `semanticEquals` on `expressions` alone and
otherwise falls back to `super.outputPartitioning`. Semantically equal
expressions do not imply equal `keyDataTypes`; this PR itself documents
`KeyedShuffleSpec.createPartitioning` keeping the keyed side's struct field
names. So a query that used to plan now fails with `IllegalArgumentException`.
Concrete case with shuffle-one-side on: `purchases p LEFT JOIN items i ON
p.item_id = i.id`, `items` identity-partitioned on `id: struct<a:int>`,
`purchases` unkeyed. The `purchases` side is shuffled via `createPartitioning`,
and being `LeftOuter` its `KeyedPartitioning(item_id, <items keys>,
[struct<a:int>])` becomes the join output. `SELECT item_id ... UNION ALL SELECT
c FROM t3` with `t3` identity-partitioned on `c: struct<b:int>`.
`BinaryComparison.sameType` ignores struct field names so no `Cast` is inserted
(the existing test `SPARK-59054: shuffle one side: struct partition keys with
different field names` plans exactly this join). `UnionExec` remaps both
expressions to the union output attribute, `semanticEquals` holds, and `concat`
receives `[struct<a:int>]` vs `[struct<b:int>]`.
Before this PR the mixed-type concat ran (with a latent `isGrouped`
miscount, since wrappers of different types never compare equal). I think the
check belongs in `UnionExec.comparePartitioning` next to the expression
comparison, so that a mismatch takes the existing fallback instead of throwing
here.
--
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]
