ulysses-you commented on code in PR #59535:
URL: https://github.com/apache/spark/pull/59535#discussion_r4332337107
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1040,6 +1036,36 @@ 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)) &&
+ kps.filterNot(_.mayContainUnknownPartitionKeys)
+ .forall(_.partitionKeys.forall(declared.contains))
Review Comment:
Changed the containment check in e2c076e9748 to build `declared.toSet` once,
only when an unmarked leg exists. The ordered-list comparison between marked
legs is unchanged.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1182,25 +1216,49 @@ object KeyedPartitioning {
KeyLayout(comparablePartitionKeys, factory.dataTypes, isGrouped,
isCollapsed = false))
}
+ /**
+ * The unknown-keys state a layout built from several members carries: the
marked members have to
+ * agree, since one layout cannot carry two descriptions of where an
undeclared row sits, and a
+ * disagreement leaves `ROUTING_LOST`. An unmarked member takes no part.
Asked of the members'
+ * states, so a new origin joins without this changing.
+ */
+ def combine(states: Seq[Option[UnknownKeysOrigin]]):
Option[UnknownKeysOrigin] =
+ states.flatten.distinct match {
+ case Seq() => None
+ case Seq(one) => Some(one)
+ case _ => Some(ROUTING_LOST)
+ }
+
/**
* Concatenates partitionings that agree on their expressions, which is what
a `UnionExec` does to
* its children's partitions. The result reports one key per output
partition, so its keys are the
* children's keys in child order.
*
* 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.
+ *
+ * One marked child marks the result, and the concatenation reports the
weakest routing of its
+ * legs': ungrouped, it routes an undeclared row per contributing leg rather
than by a hash over
+ * its own count.
*/
def concat(kps: Seq[KeyedPartitioning]): KeyedPartitioning = {
val concatenatedKeys = kps.flatMap(_.partitionKeys)
val ungroupingOrigin =
PartitioningCollection.concatUngrouping(kps.map(_.ungroupingOrigin))
+ // A child that has duplicates of its own puts them in the concatenation
too, which answers
+ // this without walking the keys.
+ val grouped = kps.forall(_.isGrouped) &&
+ concatenatedKeys.distinct.length == concatenatedKeys.length
+ // Only a grouped concatenation, where one marked leg holds every
partition, keeps the routing.
+ val routing = combine(kps.map(_.layout.unknownKeysOrigin)) match {
+ case Some(HASH_ROUTING) if !grouped => Some(ROUTING_LOST)
Review Comment:
In e2c076e9748, `concat` preserves hash routing only for a grouped result
with at most one nonempty leg. Added direct tests for disjoint nonempty legs,
empty legs around a routed leg, and an ungrouped leg.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -2809,7 +2809,7 @@ class EnsureRequirementsSuite extends SharedSparkSession {
marked <- Seq(false, true)
} yield {
val kp = KeyedPartitioning(exprs, keys)
- val p = if (marked)
kp.withLayout(_.copy(mayContainUnknownPartitionKeys = true)) else kp
+ val p = if (marked) kp.withLayout(_.copy(unknownKeysOrigin =
Some(HASH_ROUTING))) else kp
Review Comment:
Extended the pairing sweep in e2c076e9748 with `ROUTING_LOST` and
interleaved duplicate-key layouts. Added an AQE-on/off join control with
differing declared key sets; it returns the expected rows and falls back to two
shuffles with AQE off.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -817,21 +827,25 @@ private[sql] object GroupPartitionsExec {
case _ => false
}
// A `copy` of the child's layout rather than a fresh one, so the marker
and anything the
- // layout grows later are carried without this having to name them. The
origin is the one
+ // layout grows later are carried without this having to name them, except
for the routing:
+ // only an identity regrouping leaves an undeclared row where it was. The
origin is the one
Review Comment:
Shortened the comment in e2c076e9748 to describe both origin fields
explicitly: identity regrouping preserves unknown-key routing, and grouping
clears the ungrouping stamp.
--
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]
