peter-toth commented on code in PR #58351:
URL: https://github.com/apache/spark/pull/58351#discussion_r3882332636
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -528,22 +554,25 @@ case class CoalescedNullAwareHashPartitioning(
* guaranteed after projection. May contain duplicates
when ungrouped.
* @param isGrouped Whether partition keys are unique (no duplicates).
Computed on first
* creation, then preserved through copy operations to avoid
recomputation.
- * @param isNarrowed Whether this partitioning was derived from a
finer-grained one by dropping key
- * positions (e.g. via
`PartitioningPreservingUnaryExecNode`). When true and the
- * keys are no longer unique, `GroupPartitionsExec` may
merge partitions that held
- * distinct keys in the original partitioning, carrying the
same skew risk as
- * `allowKeysSubsetOfPartitionKeys`. "May", because the
condition is a proxy: the
- * duplicate keys can also come from a source that reports
several splits per
- * partition key, in which case grouping merges only
same-key partitions. Such a
- * partitioning can only satisfy `ClusteredDistribution` by
being grouped, and
- * `groupedSatisfies` refuses that unless the config is
enabled, regardless of
- * `requireAllClusterKeysForDistribution`.
+ * @param isCollapsed Whether a projection or a reduction mapped keys that
were distinct in the
+ * partitioning this one was derived from onto the same
key, so one partition
+ * here can stand for several of the original ones -- see
"Key Collapse" above.
+ * Dropping key positions does not set it on its own; the
projected keys have to
+ * actually lose distinctness. Sticky, because neither
grouping nor a further
+ * projection can make a partitioning finer again. One case
sets it without a
+ * collapse of its own: the side shuffled onto a collapsed
partitioning's keys
+ * inherits it, because the two are then co-located on that
key set -- see
+ * `KeyedShuffleSpec.createPartitioning`.
+ * Together with `!isGrouped` it decides whether
`groupedSatisfies` may coalesce
+ * the duplicate keys without
`allowKeysSubsetOfPartitionKeys`: `isCollapsed`
+ * says the collapse happened, `!isGrouped` says there is
still something left
+ * to merge, and only both together mean there is an
outstanding risk to gate.
*/
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
isGrouped: Boolean,
- isNarrowed: Boolean = false) extends Expression with Partitioning with
Unevaluable {
+ isCollapsed: Boolean = false) extends Expression with Partitioning with
Unevaluable {
Review Comment:
Done, the default is gone. In main code only the companion `apply` relied on
it, and it now passes `isCollapsed = false` explicitly with a comment saying
why a fresh source partitioning is the layout everything else is compared
against.
It also caught a live instance of exactly the laundering you describe, in
test code: `DistributionAndOrderingSuiteBase.resolvePartitioning` destructured
the flag away with `_` and rebuilt the partitioning without it. That is the
concrete payoff for the compile-time change, so it is in the commit message.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -643,12 +676,17 @@ case class KeyedPartitioning(
val joinKeyPositions =
result.keyPositions.map(_.nonEmpty).zipWithIndex.filter(_._1).map(_._2)
val projectedExpressions = joinKeyPositions.map(expressions)
val projectedKeys = projectKeys(joinKeyPositions)._2
+ // Projecting onto the operation keys can collapse keys in its own
right, which is
+ // what the key count comparison catches. The gate in `groupedSatisfies`
is bypassed while
+ // this config is on, so the flag decides nothing here today, but it
travels with the
+ // partitioning and leaving a producer to launder it is how the
protection went missing.
+ val projectedCollapsed = isCollapsed || projectedKeys.distinct.length <
distinctKeyCount
Review Comment:
Done. The count now comes off the grouped partitioning that `toGrouped`
already builds, so there is one `distinct` where there were two, and the
comparison is skipped entirely when `joinKeyPositions` selects every position
-- which, as you note, cannot collapse anything.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -137,12 +137,24 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
if (projectablePositions.length == numPositions) keySource.partitionKeys
else keySource.projectKeys(projectablePositions)._2
- val isGrouped = sharedKeys.distinct.size == sharedKeys.size
- // A KP is narrowed if this node drops positions, or if the input KPs were
already narrowed
- // (i.e. came from a finer-grained partitioning). The flag must be sticky:
a subsequent
- // PartitioningPreservingUnaryExecNode that passes all positions through
would otherwise
- // recompute isNarrowed=false, silently dropping the protection.
- val isNarrowed = projectablePositions.length < numPositions ||
keySource.isNarrowed
+ val distinctSharedKeys = sharedKeys.distinct
+ val isGrouped = distinctSharedKeys.size == sharedKeys.size
+ // This projection collapses keys when it maps keys that were distinct in
the input
+ // onto the same projected key -- dropping positions is not enough on its
own, since the
+ // projected keys can stay just as distinct as the originals. The flag is
sticky: a subsequent
+ // PartitioningPreservingUnaryExecNode that passes all positions through
must not recompute it
+ // as false and drop the protection, and no projection can make a
partitioning finer again.
+ //
+ // Both cheap terms come first: an inherited flag or a projection that
drops no position
+ // settles the question without counting distinct keys. A pass-through
projection cannot
+ // collapse anything, since it keeps the input's keys as they are.
+ //
+ // The inherited flag is read from all inputs rather than from the key
source alone. A
+ // `PartitioningCollection` normalizes it across its members, so the two
agree today; reading
+ // all of them keeps this producer correct without depending on that.
+ val isCollapsed = kps.exists(_.isCollapsed) ||
Review Comment:
Added, as `KeyedPartitioning.collapsesOnProjection`. One correction to the
shape you sketched: the `isCollapsed ||` disjunct would be dead at both call
sites, because each already carries the inherited flag outside it -- and
`AliasAwareOutputExpression` has to, since it reads the flag from every input
rather than only from the one whose keys it counts. So the helper is the count
comparison alone, and its scaladoc says that the callers own the inherited term.
`GroupPartitionsExec` deliberately does not use it: it can answer the
question exactly from the key groups it keeps, and the scaladoc points at that
as the reference definition.
--
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]
