ulysses-you commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3892006428
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -71,10 +71,23 @@ case class GroupPartitionsExec(
// can only differ in `expressions`; their `partitionKeys` reference
is shared (enforced by
// `PartitioningCollection`), so `groupedPartitions` is computed only
once.
val partitionKeys = groupedPartitions.map(_._1)
+ // When `reducers` is defined, `partitionKeys` above are the *reduced*
keys: an
+ // out-of-set key of a partitioning that may contain unknown partition
keys can reduce
+ // into the declared set (e.g. identity keys {0, 1, 2, 3} holding an
out-of-set id=4,
+ // reduced by `bucket(4, id)`, declares {0, 1, 2, 3} again), so the
reduced keyed claim
+ // cannot be trusted and must be dropped entirely -- not just the
marker, because the
+ // regrouped data layout no longer matches the child's declared keys
either.
+ if (reducers.isDefined && p.exists {
+ case k: KeyedPartitioning => k.mayContainUnknownPartitionKeys
+ case _ => false
+ }) {
+ return UnknownPartitioning(0)
Review Comment:
Fixed both halves in 9a450ca: `UnknownPartitioning(groupedPartitions.size)`
as you suggested, and the predicate now drops the claim only when every KP
member carries the marker, so the mixed-collection case you traced -- where the
marker is spurious -- keeps the claim instead of dropping. Your end-to-end
experiment is now a test: `SPARK-59050: SPJ: spurious marker of an inner join
keeps the reduced SPJ` throws at planning on 948df55 and plans the reduced SPJ
with a single shuffle on 9a450ca. I kept the `reducers` branch as the guard for
the third-party self-reducer hole and documented its (un)reachability for the
built-in transforms in the comment.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -131,6 +131,16 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
if (projectablePositions.isEmpty) return LazyList.empty
+ // If any input KP may contain unknown partition keys, its keyed claim
only holds while the
+ // declared key set survives verbatim: the claim asserts that rows outside
the declared keys
+ // cannot exist, and dropping a key position coarsens the declared set so
that an out-of-set
+ // key can land inside it (see
`KeyedPartitioning.mayContainUnknownPartitionKeys`). Drop the
+ // keyed claim entirely in that case.
+ if (projectablePositions.length < numPositions &&
+ kps.exists(_.mayContainUnknownPartitionKeys)) {
Review Comment:
Fixed in 9a450ca: `projectKeyedPartitionings` now computes the propagated
marker as `kps.forall(_.mayContainUnknownPartitionKeys)` and the narrowing-drop
guard uses that value, so a mixed collection's claim passes through as accurate
while a genuinely all-marked collection still drops (with both sides
re-shuffled, matched rows can carry an out-of-set key). `GroupPartitionsExec`
got the same scoping. Pinned by `SPARK-59050: SPJ: inner join with in-set rows
keeps the sound SPJ downstream` (2 shuffles vs 1 under the `exists` scoping)
and a data-level variant carrying a genuine out-of-set row through the inner
join (`...inner join drops out-of-set rows before the cleared marker is
trusted`). Noted on #58351: the scoping is deliberately per-member marker
semantics, neither the OR-normalization nor the require-agreement; happy to
rebase whichever lands second.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -538,12 +538,23 @@ case class CoalescedNullAwareHashPartitioning(
* partitioning can only satisfy `ClusteredDistribution` by
being grouped, and
* `groupedSatisfies` refuses that unless the config is
enabled, regardless of
* `requireAllClusterKeysForDistribution`.
+ * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
+ * not among the declared `partitionKeys`.
This happens when a side
+ * is re-shuffled onto this partitioning (see
+ * `KeyedShuffleSpec.createPartitioning`):
`KeyGroupedPartitioner`
+ * silently routes keys outside the declared
set to arbitrary
+ * partitions, so only the declared keys are
guaranteed to be
+ * co-located. Such a partitioning is unsound
to storage-partition
+ * join against a side whose partition keys
are not a subset of the
+ * declared keys -- see
`KeyedShuffleSpec.areKeysCompatible`.
Review Comment:
Done in 9a450ca: the `== Partition Keys ==` layout paragraph gained the
exception sentence -- when `mayContainUnknownPartitionKeys` is set a partition
may hold rows whose key is not declared -- pointing at the `@param`. When
#58351 lands the sentence moves with wherever the layout paragraph ends up.
--
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]
