ulysses-you commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3910710070
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1678,7 +1751,18 @@ case class KeyedShuffleSpec(
// The shuffled side is laid out on this side's partition keys, so it
inherits the flag. That
// is conservative rather than strictly true, and it can only ever add a
shuffle: a later
// grouping of the shared key set carries the collapsed side's risk.
- partitioning.copy(expressions = newExpressions)
+ //
+ // The child re-shuffled onto this layout may hold keys outside the
declared set, so every
+ // partitioning produced here carries the unknown-keys marker too (see
+ // `KeyedPartitioning.mayContainUnknownPartitionKeys`).
`createPartitioning` is only reached
+ // from `EnsureRequirements`' shuffle loop to re-shuffle a join child onto
the best spec's
+ // layout, and the re-shuffled child's keys are never provably a subset of
the declared keys:
+ // a non-keyed (v1) child's keys are unknown to the planner, and a keyed
child with an
+ // incompatible partitioning is re-evaluated in this spec's key space,
which the planner
+ // cannot bound. Marking every such partitioning is therefore
+ // sound (conservative only when the two sides happen to share a transform
and the re-shuffled
+ // keys are a known subset).
+ partitioning.copy(expressions = newExpressions,
mayContainUnknownPartitionKeys = true)
Review Comment:
Thanks @cloud-fan -- fixed in 7464c25 exactly as scoped: the
`OrderedDistribution` branch keeps the ordering claim of a marked layout only
with `numPartitions == 1` (one partition holds everything, so local order is
global order). The repro test `SPARK-59050: SPJ: a global ORDER BY over an
unknown-keyed layout must range-partition` measures your [1, 4, 2] on the
pre-gate tree, compares `collect()` directly rather than through `checkAnswer`
(which sorts both sides), and pins the range exchange.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala:
##########
@@ -192,6 +192,35 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
assert(gpe.outputOrdering === Nil)
}
+ test("SPARK-59050: unknown-keyed child with reducers gives up the keyed
claim at the real " +
+ "count") {
+ // No planner path reaches the give-up branch with the built-in transforms
--
+ // `createShuffleSpec` refuses a narrowing projection of an unknown-keyed
partitioning one
+ // hop earlier, and `areKeysCompatible` pairs it only with same-function
partners that have
+ // no reducer (see `GroupPartitionsExec.outputPartitioning`) -- so pin the
contract
+ // directly: the node must report `UnknownPartitioning` with its physical
grouped count.
+ // Reporting zero partitions is what threw once a parent join built a
+ // `PartitioningCollection` over both sides.
+ val partitionKeys = Seq(row(1), row(2), row(1))
+ val child = DummySparkPlan(
+ outputPartitioning = KeyedPartitioning(Seq(exprA), partitionKeys)
+ .copy(mayContainUnknownPartitionKeys = true))
+ val gpe = GroupPartitionsExec(child, reducers = Some(Seq(None)))
Review Comment:
Good catch, thanks -- done in 7464c25. The pin now builds a real mod-2
`KeyReducer` ([1, 2, 3] reduced to [1, 0, 1]), asserts 2 grouped partitions
where the child had 3, and ends with `PartitioningCollection.fromPartitionings`
over a 2-partition partner: reverting the give-up count to
`UnknownPartitioning(0)` fails both the count and the collection.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1545,6 +1574,45 @@ case class KeyedShuffleSpec(
}
} && expressions.zip(otherExpressions).forall {
case (l, r) => isExpressionCompatible(l, r)
+ } && {
+ // An unknown-keyed side (see
`KeyedPartitioning.mayContainUnknownPartitionKeys`) guarantees
+ // co-location only for its declared keys, and `KeyGroupedPartitioner`'s
out-of-set routing
+ // is a deterministic hash -- so it can pair only with a side whose keys
are a subset of the
+ // declared keys.
+ //
+ // The key comparison below must also happen in a single domain:
`isExpressionCompatible`
+ // admits an `AttributeReference` against a `TransformExpression` (and
two different-but-
+ // compatible transforms) when `v2BucketingAllowCompatibleTransforms` is
on, and in those
+ // cases the two sides' `partitionKeys` hold raw values on one side and
transform outputs on
+ // the other, so the subset test would compare unrelated values. Require
the partition
+ // expressions to be the same function per position before comparing
keys.
+ //
+ // Two unknown-keyed sides are compatible only when they agree on the
declared keys *and*
+ // their order: the out-of-set keys hash to the same-index partition on
both sides, and a
+ // `GroupPartitionsExec` regrouping re-labels each partition by that
side's declared key, so
+ // a differing declared order would push the out-of-set keys into
different output
+ // partitions and lose their matches.
+ if (partitioning.mayContainUnknownPartitionKeys ||
+ other.partitioning.mayContainUnknownPartitionKeys) {
+ expressions.zip(otherExpressions).forall {
+ case (_: AttributeReference, _: AttributeReference) => true
+ case (l: TransformExpression, r: TransformExpression) =>
l.isSameFunction(r)
Review Comment:
Added in 7464c25, thanks: a local reducible bucket-like function (catalyst
has no `BucketFunction`) pairs bucket(4) against bucket(8) under
`allowCompatibleTransforms`, marked in both argument orders (refused) and
unmarked (still admissible, guarding the pre-existing relaxation), with both
specs declaring the same key set {0, 1} so only the same-function gate can
cause the refusal.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -573,12 +577,29 @@ case class CoalescedNullAwareHashPartitioning(
* 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 to arbitrary partitions
when a side is
+ * re-shuffled onto this partitioning (see
+ * `KeyedShuffleSpec.createPartitioning`), so
only the declared
Review Comment:
Documented in 7464c25, thanks -- the `@param` now states the exception the
`areKeysCompatible` branch relies on: two marked partitionings declaring the
same keys in the same order still pair, since equal undeclared keys hash to the
same partition.
--
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]
