peter-toth commented on code in PR #58501:
URL: https://github.com/apache/spark/pull/58501#discussion_r3944836881
##########
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:
Confirmed, and reproduced. Thank you.
Chasing it end to end turned up a fourth site that puts a partitioning's
expressions over keys it did not build:
`PartitioningPreservingUnaryExecNode.projectKeyedPartitionings`, which projects
`kps.head` once and then stamps every alias alternative onto it with
`copy(expressions = ...)`. A `bucket(4, id)` member ended up reporting the
`identity(id)` member's `LongType`. So an independent `keyDataTypes` field has
to be decided at four places, and two of them mix members.
That is what killed the design. This PR is reshaped: there is no
`keyDataTypes` field, it stays derived, and the constructor `require` is gone
with it. `GroupPartitionsExec` stamps no types.
Carrying the types on the partitioning moved to SPARK-59285. There they live
on a shared `KeyLayout` that the collection's members hold by reference, so no
member can report another member's types, and the four re-targeting sites have
nothing to decide.
##########
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:
Confirmed. The `require` is gone, since there is no `keyDataTypes` field to
check.
Your parenthesis turned out to be the important part. I measured the latent
`isGrouped` miscount, and it is not latent: **it silently drops rows.** The
union reports unique keys, so nothing regroups it,
`KeyedShuffleSpec.canCreatePartitioning` accepts it, and the other side is
shuffled straight onto those keys. `KeyGroupedPartitioner`'s map holds one
partition per key, so the union partition holding the earlier copy of the
repeated key receives no rows at all.
-- items(id struct<a:int>, name string) partitioned by identity(id),
keys ('a',1), ('a',2)
-- purchases(item_id struct<b:int>, ...) unpartitioned
-- t3(c struct<b:int>) partitioned by identity(c), key
('b',1)
-- s4(k4 struct<b:int>, w string) unpartitioned, rows
(('b',1),'x'), (('b',2),'y')
SELECT u.k, s.w FROM (
SELECT p.item_id AS k FROM purchases p LEFT JOIN items i ON p.item_id
= i.id
UNION ALL SELECT c AS k FROM t3
) u JOIN s4 s ON u.k = s.k4
2 rows on master where 3 are correct, with `v2.bucketing.shuffle.enabled`
and `union.output.partitioning.enabled` on and AQE off. An inner join loses a
row, with no error.
That is now half of what this PR fixes, and it is the test `SPARK-59187: a
union of a key that repeats across children under two namings joins right`.
Thank you for spotting it.
##########
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:
You are right, and the PR no longer claims it. There is no field, so
`keyDataTypes` is derived per member again.
Part of what you describe is closed by the erasure: two members over one key
space that differ only in how their struct fields are named now give one
answer, whichever reader consults whichever member.
Part of it is not. Two members can describe genuinely different key spaces,
`[String]` and `[Int]` in your example, and nothing here refuses that. Your
`require`, or normalizing in `fromPartitionings`, is in SPARK-59285:
`fromPartitionings` interns one shared layout and refuses a member that
describes another space, which also covers `isGrouped` that the field-by-field
check leaves out.
I kept it out of this PR because I measured the mixed collection on master
and it is not a live bug. The plan passes `ValidateRequirements` and the query
returns the right answer, so that part is hardening.
--
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]
