cloud-fan commented on code in PR #59535:
URL: https://github.com/apache/spark/pull/59535#discussion_r4333012297
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1182,25 +1214,50 @@ 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
Review Comment:
**Nit (P3):** This paragraph still describes the round-one rule, under which
a grouped concatenation kept `HASH_ROUTING`. Since e2c076e9748 the body (lines
1250-1254) also returns `ROUTING_LOST` for a grouped result with more than one
nonempty leg. Take a routed marked [1, 2, 3] leg and an unmarked [4, 5] leg.
Every leg is routed, so "the weakest routing of its legs" would be
`HASH_ROUTING`, but `concat` returns `ROUTING_LOST`, and the new
ShuffleSpecSuite test asserts exactly that. The doc names only the ungrouped
case as a cause.
This is documentation only. The code and the inline comment at line 1249 are
right, and `UnionExec`, the only caller, never builds such a concatenation. But
the earlier thread asked `concat` to uphold this rule on its own, so a later
caller or a relaxed condition based on this contract could treat a grouped
multi-leg result as a hash-routed shuffle template. That is the lost-match case
from that thread. Restating the inline comment's rule here would close it: hash
routing survives only a grouped concatenation in which at most one leg holds
partitions. Every other marked concatenation records `ROUTING_LOST`, because
each leg routed its undeclared rows by a hash over its own partition count.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59535#discussion_r4324836928).
**Verification:**
- **Inspection:** The routing rule in the concat method documentation agrees
with the downgrade condition in its body and with the inline comment beside it:
a grouped result with more than one nonempty leg, or any ungrouped result, is
documented as losing hash routing.
- **Behavior:** Concat routing coverage still passes unchanged, showing that
a grouped concatenation of a routed marked leg with a nonempty accurate leg
loses routing while empty legs around a routed leg keep it.
##########
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:
Thanks, verified in e2c076e9748: `canMergeMarkedKeys` builds
`declared.toSet` once, only when an unmarked leg exists, and the ordered
comparison between marked legs is unchanged.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4324836901","thread_id":"inline:4324836901","verdict_sha256":"7089052e7047ccaa4bfd0c1c87df8559df7f9efa44aaac13090af6585efe82f7"}
-->
##########
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:
Thanks, verified: `concat` now keeps `HASH_ROUTING` only for a grouped
result with at most one nonempty leg, and the new ShuffleSpecSuite cases cover
it. The method Scaladoc above `concat` still describes the earlier rule; I left
a separate comment on it in partitioning.scala.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4324836928","thread_id":"inline:4324836928","verdict_sha256":"7089052e7047ccaa4bfd0c1c87df8559df7f9efa44aaac13090af6585efe82f7"}
-->
##########
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:
Thanks, verified: the sweep now generates `ROUTING_LOST` and the interleaved
duplicate-key shape under the `ValidateRequirements` oracle, and the
differing-key-set join runs with AQE on and off and falls back to a valid plan.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4324836938","thread_id":"inline:4324836938","verdict_sha256":"7089052e7047ccaa4bfd0c1c87df8559df7f9efa44aaac13090af6585efe82f7"}
-->
##########
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:
Thanks, verified: the comment now matches the code, stating that only
identity regrouping keeps unknown-key routing and that grouping clears the
ungrouping stamp.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4324836951","thread_id":"inline:4324836951","verdict_sha256":"7089052e7047ccaa4bfd0c1c87df8559df7f9efa44aaac13090af6585efe82f7"}
-->
--
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]
