dongjoon-hyun commented on code in PR #58552:
URL: https://github.com/apache/spark/pull/58552#discussion_r3972192250
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -719,16 +755,18 @@ case class KeyedPartitioning(
}
copy(
expressions = positions.map(expressions),
- partitionKeys = projectedKeys,
- isGrouped = !collapses && sourceOf.size == projectedKeys.length,
- isCollapsed = isCollapsed || collapses)
+ layout = KeyLayout(
Review Comment:
`project` used to go through `copy(...)`, which carried
`mayContainUnknownPartitionKeys` along with the other fields. Building a fresh
`KeyLayout` here leaves it at the default `false`, so a marked partitioning
projected with any non-identity position list comes out unmarked, and
`satisfies(ClusteredDistribution(subset))` would then accept it under
`v2BucketingAllowKeysSubsetOfPartitionKeys`.
No caller reaches this today (`joinKeyPositions` / `projectablePositions`
are always ascending, so a marked input either narrows and is refused, or is
the identity and returns `this`), but the comment in `createShuffleSpec`
("`project` carries the unknown-keys marker across") no longer describes what
`project` does. Could we pass `mayContainUnknownPartitionKeys =
mayContainUnknownPartitionKeys` here so the contract holds regardless of the
caller?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -572,58 +642,37 @@ case class CoalescedNullAwareHashPartitioning(
* }}}
*
* @param expressions Partition transform expressions (e.g., `years(col)`,
`bucket(10, col)`).
- * @param partitionKeys Partition keys wrapped in InternalRowComparableWrapper
for efficient
- * comparison and grouping. One per partition. Typically
in sorted order when
- * produced by a data source or `GroupPartitionsExec`,
but this is not
- * 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 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 key here can
- * stand for several of the original ones. Sticky. See "Key
Collapse" above for
- * what it gates and how it travels.
- * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
- * not among the declared `partitionKeys`.
`KeyGroupedPartitioner`
- * routes such rows by a deterministic hash
when a side is
- * re-shuffled onto this partitioning (see
- * `KeyedShuffleSpec.createPartitioning`), so
co-location holds
- * for whole keys only: two marked
partitionings declaring the
- * same keys in the same order and using the
same partition
- * function per position still pair (equal
undeclared keys hash
- * to the same partition), but a row of an
undeclared key sits in
- * the partition of some other declared key
and need not be
- * co-located with rows sharing only a subset
of its columns.
- * `satisfies` and
`KeyedShuffleSpec.areKeysCompatible` therefore
- * accept a marked partitioning only for
full-key clustering,
- * never for a subset of its partition columns
and never for a
- * global ordering across several partitions.
Two carry rules:
- * (1) a node that changes the declared key
set must drop the
- * keyed partitioning, whether it coarsens it
(a key-dropping
- * projection, a key-changing reduction, a
join-key projection)
- * or expands it over a marked leg (a union,
where another leg
- * may declare exactly the key that leg holds
out-of-set);
- * (2) marker agreement is enforced by
`PartitioningCollection`:
- * the constructor requires it and
`fromPartitionings` normalizes
- * by OR, so members are uniformly marked or
unmarked and
- * consumers may read one member.
`ShuffledJoin`'s `InnerLike`
- * arm, the only site that meets a marked
input with an unmarked
- * one, clears the markers first. That is
precision, not the
- * guarantee: without it the OR would spread
the spurious marker
- * onto the accurate side.
+ * @param layout The partitions this one describes, which is everything about
them except the
+ * expressions naming them. See [[KeyLayout]].
*/
case class KeyedPartitioning(
expressions: Seq[Expression],
- @transient partitionKeys: Seq[InternalRowComparableWrapper],
- isGrouped: Boolean,
- isCollapsed: Boolean,
- mayContainUnknownPartitionKeys: Boolean = false)
- extends Expression with Partitioning with Unevaluable {
- override val numPartitions = partitionKeys.length
+ layout: KeyLayout) extends Expression with Partitioning with Unevaluable {
+ override val numPartitions = layout.partitionKeys.length
+
+ def partitionKeys: Seq[InternalRowComparableWrapper] = layout.partitionKeys
+ def isGrouped: Boolean = layout.isGrouped
+ def isCollapsed: Boolean = layout.isCollapsed
+ def mayContainUnknownPartitionKeys: Boolean =
layout.mayContainUnknownPartitionKeys
+
+ /** This partitioning over a changed layout, e.g.
`withLayout(_.copy(isGrouped = false))`. */
+ def withLayout(f: KeyLayout => KeyLayout): KeyedPartitioning = copy(layout =
f(layout))
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Prints the layout's contents where the value object would print, which
keeps the plan string as
+ * it was before the layout held them. The list is curated rather than the
layout's own fields,
+ * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
+ * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
+ * printing it would put a struct field name into a plan that appears
nowhere in the query. A
+ * field the layout grows is therefore a decision here.
+ */
+ override protected def stringArgs: Iterator[Any] =
Review Comment:
Before this PR there was no `stringArgs` override, so the default
`productIterator` printed all five fields, including
`mayContainUnknownPartitionKeys`. This list drops the marker, so the plan
string did change: a partitioning built by
`KeyedShuffleSpec.createPartitioning` (always marked) now prints the same as an
unmarked one, which loses the one visible signal in `EXPLAIN` for why a
subset-key window/aggregate still shuffles.
Either append `mayContainUnknownPartitionKeys` here
(`TransformExpression.stringArgs` is a precedent for printing a trailing flag
conditionally), or update this comment and the PR description, which says there
are no plan string changes.
--
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]
