peter-toth commented on code in PR #58262:
URL: https://github.com/apache/spark/pull/58262#discussion_r3879911888
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -544,6 +550,21 @@ case class KeyedPartitioning(
@transient lazy val expressionDataTypes: Seq[DataType] =
expressions.map(_.dataType)
+ /**
+ * The data types of the `partitionKeys` rows: the types each key was built
with, and the ones it
+ * is hashed and compared under (`InternalRowComparableWrapper.dataTypes`).
+ *
+ * These are the `expressionDataTypes` unless a reducer has rewritten the
keys. With
+ * `v2BucketingAllowCompatibleTransforms` a storage-partitioned join reduces
one or both sides'
+ * keys onto a common key space, while the partitioning keeps reporting the
expressions it was
+ * built from: joining an `identity(ts)`-partitioned table to a
`years(ts)`-partitioned one leaves
+ * `IntegerType` year values under a `TimestampType`-declared expression.
Reading a key row is
+ * only sound at the types the row was written with, which is what these are
for; use
+ * `expressionDataTypes` only where the question is about the expressions
themselves.
+ */
+ @transient lazy val keyDataTypes: Seq[DataType] =
+ partitionKeys.headOption.map(_.dataTypes).getOrElse(expressionDataTypes)
Review Comment:
Confirmed, and fixed by deriving `keyRowOrdering` from `keyDataTypes`.
Thanks - the trace is right, and it made me walk every reader rather than just
this one.
Four places read key rows at the expressions' types and now read them at
`keyDataTypes`: `keyRowOrdering` (so `toGrouped`, and `PushDownUtils`' key sort
through `keyOrdering`), `reduceKeys`, the base types the reduce path passes on
both sides in `EnsureRequirements`, and `PushDownUtils`' wrapper factory for
the keys a scan reports after runtime filtering - that last one for consistency
only, since it sees a scan's own partitioning, where the two sources coincide.
The `keyRowOrdering` one also closes a latent divergence:
`GroupPartitionsExec.groupAndSortByKeys` already sorted the reduced keys at the
reduced types, so for exactly these partitionings the two sides of the contract
in `groupedKeyRowOrdering`'s scaladoc did not agree.
One place keeps `expressionDataTypes`, with a comment now saying why:
`ShuffleExchangeExec` wraps lookup keys it *evaluates* from the expressions per
row, and the stored keys it matches them against have to be declared the same
way, so the two move together. For a reducer-rewritten partitioning neither
choice works - the stored keys are in the reduced key space and the evaluated
ones are not - so such a partitioning must not be shuffled onto at all. That
gate is `KeyedShuffleSpec.canCreatePartitioning`, and closing it belongs with
the stale-expression follow-up.
The `OrderedDistribution` arm you also point at is not a type swap.
`RowOrdering.create(o.ordering, attrs)` binds the distribution's sort orders to
the partition attributes, so with reduced keys the ordering is over the wrong
space rather than merely at the wrong type; rebuilding it over the key types
would have to carry the distribution's directions and null orderings, and
refusing the partitioning is probably the better answer. A query does reach it,
by the way - `v2BucketingAllowSorting` with a global sort on the partition key
over a reduced join - and it throws the same `ClassCastException` on `master`,
at the same place. On the follow-up list, named explicitly as you suggest.
Test added in `EnsureRequirementsSuite`, `createShuffleSpec sorts the
projected keys at the types they were built with` - your repro inverted. It
throws your `ClassCastException` before the fix and asserts the sorted
projected keys after.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -867,42 +879,259 @@ case class EnsureRequirements(
}
/**
- * Splits a partitioning into three categories:
- * 1. Non-KeyedPartitioning (HashPartitioning, RangePartitioning, etc.)
- * 2. Grouped KeyedPartitioning (isGrouped = true)
- * 3. Non-grouped KeyedPartitioning (isGrouped = false)
+ * The positions of `kp`'s partition expressions that are operation keys of
`distribution`, and so
+ * have to survive a projection. All of them when nothing needs projecting.
+ *
+ * Under `v2BucketingAllowKeysSubsetOfPartitionKeys` a [[KeyedPartitioning]]
may be grouped on
+ * more keys than the operation requires, in which case partitions sharing
an operation key are
+ * still separate. A partition expression is an operation key in two ways:
one of its *references*
+ * is a cluster key - the form `groupedSatisfies` and
`KeyedShuffleSpec.keyPositions` both use,
+ * where a `bucket(4, a)` transform covers the cluster key `a` - or the
expression *itself* is a
+ * cluster key. The second is never decisive in practice, because
`IdentityTransform` resolves to
+ * the attribute itself and then the reference-level test matches the same
position anyway; it is
+ * kept so that a partition expression which is a cluster key can never be
projected away.
+ *
+ * Returns every position for a co-partitioned operator: there the
multi-child block owns the
+ * projection, and doing it here as well would leave that block deriving
positions from an already
+ * projected partitioning and applying them to the unprojected partition
expressions.
+ *
+ * An empty result means no partition expression covers an operation key, so
there is nothing to
+ * project onto. That is the answer for a member that cannot satisfy
`distribution`, which is why
+ * the caller only asks for members that can. It also happens for a member
that can: one whose
+ * expressions have no references at all makes every `groupedSatisfies`
branch vacuously true, and
+ * nothing at `KeyedPartitioning` construction rejects that. The caller
skips such a member rather
+ * than project it to no position, which would collapse every partition into
one.
+ *
+ * Keeping a position is only sound because `groupedSatisfies`' subset
branch also requires
+ * `expressions.forall(_.references.size == 1)`: a kept expression is then a
function of a single
+ * cluster key, so coalescing on the projected keys cannot put rows that
share an operation key on
+ * different partitions.
+ */
+ private def clusterKeyPositions(
+ kp: KeyedPartitioning,
+ distribution: Distribution,
+ isCoPartitioned: Boolean): BitSet = distribution match {
+ case c: ClusteredDistribution if !isCoPartitioned =>
+ val positions = kp.expressions.indices.filter { i =>
+ val e = kp.expressions(i)
+ c.clustering.exists(_.semanticEquals(e)) ||
+ e.references.exists(ref =>
c.clustering.exists(_.semanticEquals(ref)))
+ }.to(BitSet)
+ positions
Review Comment:
Fixed, thanks.
--
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]
