ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4081502311
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1437,6 +1445,54 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * Whether a finished plan may report an ungrouped keyed member, which only
partially clustered
+ * distribution builds. Read by `specsForPairing` and by
`ValidateRequirements`' per-side
+ * admission, so the two cannot drift.
+ */
+ private[sql] def mayUngroupedMember: Boolean =
+ SQLConf.get.v2BucketingPartiallyClusteredDistributionEnabled
+
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve
it, each of them the
+ * layout that member reports. A keyed member is admitted on `satisfies`,
the as-it-stands
+ * question, count included; a member that is not keyed is asked on
`satisfies` as well and has no
+ * projection to make.
+ *
+ * The one member whose layout a finished plan may report without satisfying
the distribution is
+ * an ungrouped keyed one: that is the shape partially clustered
distribution spreads, and
+ * `mayUngroupedMember` confines it to the configuration that builds one.
`keysSatisfy` is the
+ * as-it-stands question for it, since `satisfies` adds `isGrouped` on top.
+ *
+ * This is the planner's admission of a member
(`EnsureRequirements.createKeyedShuffleSpecs`) less
+ * the coverage of every operation key it requires there
+ * (`spark.sql.requireAllClusterKeysForCoPartition`), which is a skew
heuristic: a member whose
Review Comment:
Thanks @cloud-fan! Corrected in f95e44a9f42, in the helper's text and in the
PR description: disabling the coverage check admits a member whose partitioning
keys cover only a subset of the operation's keys, and the partition keys being
a superset is the separate projection case under
`allowKeysSubsetOfPartitionKeys`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1457,6 +1457,30 @@ object PartitioningCollection {
case other => other.satisfies(required)
}
+ /**
+ * The specs `p` offers for `distribution`, one per member that may serve
it, built the way the
+ * planner builds them (`createShuffleSpec`): a keyed member that covers
part of the operation's
+ * keys contributes the projection onto the keys it covers, which is the
layout the alignment
+ * about to be planned will emit for it. A member that is not keyed has no
projection to make and
+ * is asked for its own. A keyed member is admitted on `keysMaySatisfy`, any
other member on
+ * `satisfies`, and a count the operation pinned is asked as it stands, the
way
+ * `maySatisfyAfterProjection` asks it. That is the planner's admission of a
member
+ * (`EnsureRequirements.createKeyedShuffleSpecs`) less the coverage of every
operation key it
+ * also requires there: that requirement is the skew heuristic of
+ * `spark.sql.requireAllClusterKeysForCoPartition`, while a member covering
a subset of the
+ * operation keys is a sound pairing.
+ */
+ private[sql] def specsForPairing(
+ p: Partitioning,
+ distribution: ClusteredDistribution): Seq[ShuffleSpec] =
+ flatten(p).flatMap {
+ case k: KeyedPartitioning =>
+ Option.when(distribution.requiredNumPartitions.forall(_ ==
k.numPartitions) &&
+ k.keysMaySatisfy(distribution))(k.createShuffleSpec(distribution))
Review Comment:
One correction to my reply above, from this round's producer-boundary
review: the spec is not the member's layout verbatim. It is that layout viewed
on the key the operation clusters on, so a partition expression carrying no
cluster key is left out, with the member's own count and its own key order, no
key deduped and none re-sorted. `PartitioningCollection.reportedSpecOf` states
that and why, and the point of the reply stands: `createShuffleSpec`'s grouped
layout is not what the validator reads. Head is f95e44a9f42.
--
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]
