peter-toth commented on code in PR #59535:
URL: https://github.com/apache/spark/pull/59535#discussion_r4334983483
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1040,6 +1036,42 @@ case class UnionExec(children: Seq[SparkPlan]) extends
SparkPlan with CodegenSup
}
}
+ /**
+ * Whether the legs' markers let their keys be merged. A marked leg may hold
rows of a key it
+ * does not declare, and `KeyGroupedPartitioner` routes those into the
partition of some declared
+ * key. Two conditions keep every key's rows in one group, which is what a
clustering reads:
+ *
+ * - a marked leg declares every key the other legs declare, so its rows
of such a key do not
+ * land in a partition the merge names for another key;
+ * - marked legs declare the same keys in the same order, and when more
than one is marked,
+ * each still routes by the hash. A routed leg is grouped, so its keys
are unique and the
+ * list length is the modulus the hash is taken over: the position in
that list is the name a
+ * key no leg declares reaches, and equal lists agree on both. A leg
whose routing is lost
+ * names such a key by an earlier layout's order, which its list does
not show. One marked
+ * leg needs no agreement, since its undeclared rows reach one group
either way.
+ *
+ * The merged layout cannot claim that a partition holds only the key it
names, which
+ * [[KeyedPartitioning.concat]] records.
+ */
+ private def canMergeMarkedKeys(kps: Seq[KeyedPartitioning]): Boolean = {
+ val marked = kps.filter(_.mayContainUnknownPartitionKeys)
+ if (marked.isEmpty) {
+ true
+ } else {
+ val declared = marked.head.partitionKeys
+ marked.tail.forall(_.partitionKeys == declared) &&
+ (marked.size == 1 || marked.forall(_.unknownKeysRouted)) && {
Review Comment:
**Finding 3.** Same gap as finding 1, for a third guard. Only the hand-built
`lostUnion` and `mixedUnion` plans
(`KeyGroupedPartitioningSuite.scala:9755-9766`) pin this check, but a SQL query
reaches it too. An aggregate over a merged marked union gives one leg
`ROUTING_LOST` over `[1, 2, 3, 4]`. The other leg is still `HASH_ROUTING` over
the same list:
```scala
// The lostRouting test's tables: a1 {3, 4}, a2 {1, 2}, a3 {1, 2, 3, 4}, s1
{1, 2},
// parquet t1 and t2 with ids 1 to 40.
val query =
"""
|WITH leg1 AS (
| SELECT t1.id FROM (
| SELECT id FROM testcat.ns.a1
| UNION ALL
| SELECT id FROM testcat.ns.a2) a
| RIGHT OUTER JOIN t1 ON a.id = t1.id),
|legA AS (
| SELECT id, count(*) AS c FROM (
| SELECT id FROM leg1
| UNION ALL
| SELECT id FROM testcat.ns.s1) GROUP BY id),
|legB AS (
| SELECT t2.id FROM testcat.ns.a3 a
| RIGHT OUTER JOIN t2 ON a.id = t2.id)
|SELECT id, count(*) AS cnt FROM (
| SELECT id FROM legA
| UNION ALL
| SELECT id FROM legB) u
|GROUP BY id
|""".stripMargin
```
The undeclared rows in `legA` keep the labels of the `[3, 4, 1, 2]` routing,
while those in `legB` follow `[1, 2, 3, 4]`. At the head the outer union
reports `UnknownPartitioning`, and every id counts 2, with AQE off and on. With
this check replaced by `true` (AQE off), the union merges the two legs and the
query returns 76 rows. Each id from 5 to 40 comes back twice with count 1. Only
the hand-built test fails. This could be a third case of the parameterized
`lostRouting` test.
--
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]
