peter-toth commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3915661513
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1100,6 +1153,15 @@ object PartitioningCollection {
case _ => None
}
+ /**
+ * Whether `p` or any nested [[KeyedPartitioning]] may contain unknown
partition keys, read
+ * from its first keyed member, the same one
`checkKeyedPartitioningInvariant` compares
+ * against. Keyless inputs answer false: `EnsureRequirements` normalizes a
shuffled join's
+ * children onto one template, so they never meet a keyed sibling here.
+ */
+ private[sql] def mayContainUnknownPartitionKeys(p: Partitioning): Boolean =
Review Comment:
**Finding 19.** This answers `false` for two different inputs: a
partitioning with no `KeyedPartitioning` in it, and one whose keyed members are
unmarked. `clearUnknownPartitionKeys` folds over exactly this value
(`ShuffledJoin.scala:101`), so a keyless sibling reads as "unmarked",
`markers.forall(_ == markers.head)` is false, and the keyed input's marker is
cleared — a genuine claim dropped because the other side has no keys at all.
Round 4's `members.forall(_.mayContainUnknownPartitionKeys)` over the
flattened keyed members could not fail that way, since a keyless input
contributed no member.
I could not construct the shape at this head, so this is latent, and your
doc sentence is right about why:
- `KeyedShuffleSpec.isCompatibleWith` ends in `case _ => false`
(`partitioning.scala:1629`), so a keyed spec never keeps a keyless partner in
place.
- `SinglePartitionShuffleSpec.canCreatePartitioning` is `false`
(`partitioning.scala:1326`), and `RangeShuffleSpec`'s is too, so a keyless spec
cannot become `bestSpecOpt` and leave the keyed child untouched. With the keyed
spec best the keyless child is re-shuffled by `createPartitioning`; with
neither in `candidateSpecs` both children get `dist.createPartitioning(...)`.
- `checkKeyGroupCompatible` returns `None` unless both children yield a
`KeyedShuffleSpec`.
What makes it worth closing is that the argument is a property of one of
this helper's two call sites. `GroupPartitionsExec.scala:91` passes a single
child's partitioning, where "meets a keyed sibling" is not a question at all,
and the keyless `false` is right there for an unrelated reason: `p.transform`
has no `KeyedPartitioning` to rewrite either way. So the sentence documents the
caller that needs it, on the helper that does not.
Letting the helper say "no keyed member" separates the two. It keeps the
short-circuit and the zero-allocation path, and needs no presence filter:
```scala
private[sql] def keyedMarkerOf(p: Partitioning): Option[Boolean] =
representativeOf(p).map(_.mayContainUnknownPartitionKeys)
```
Then in `ShuffledJoin`:
```scala
val markers = partitionings.flatMap(PartitioningCollection.keyedMarkerOf)
if (markers.isEmpty || markers.forall(_ == markers.head)) {
```
with the per-input pattern guard becoming
`keyedMarkerOf(partitioning).contains(true)`, and `GroupPartitionsExec` reading
`PartitioningCollection.keyedMarkerOf(p).contains(true)` — the same value it
reads today.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -131,9 +131,20 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
if (projectablePositions.isEmpty) return LazyList.empty
- // All input KPs share the same partitionKeys and isCollapsed flag by
invariant, so the first
- // one projects the keys and both flags for every combination below. Only
the expressions
- // differ.
+ // Collection members carry a uniform marker (mixed collections are
cleared at
Review Comment:
**Finding 20.** "mixed collections are cleared at construction by
`ShuffledJoin`" stopped being the reason the head represents them all. Since
6021d83 the reason is `PartitioningCollection`'s `require`
(`partitioning.scala:1105`) plus the OR in `fromPartitionings`
(`partitioning.scala:1178`): a mixed collection is now unrepresentable, whether
or not anything cleared. And without the clearing the OR would not leave a
mixed collection either — it would mark every member.
That matters for whoever reads this line next. It points at the wrong
invariant, so deleting the clearing looks like it would break `kps.head`. It
would not; it would spread a spurious marker instead.
```scala
// `PartitioningCollection` requires its members to agree on the marker,
so the head
// represents them all.
```
Two more comments carry the same attribution:
- `ShuffledJoin.scala:92` — "when marked and unmarked members meet". They
cannot meet inside one collection any more. What meet are the two *inputs*.
- `partitioning.scala:604` — rule (2) reads "the only site that mixes them,
clears it at construction; `PartitioningCollection` *additionally* normalizes
the marker by OR". With the OR in place the normalization is the guarantee and
the clearing is the precision, not the other way round.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/DistributionSuite.scala:
##########
@@ -447,6 +448,59 @@ class DistributionSuite extends SparkFunSuite {
assert(interned.partitionKeys eq kpX.partitionKeys)
}
+ test("SPARK-59050: a marked one-partition layout keeps the global ordering
claim") {
+ // The one-partition exemption inside `keysSatisfy`'s ordered branch: a
single partition
+ // holds every row, so an out-of-set key cannot break the cross-partition
sequence, while
+ // two partitions can (the e2e `ORDER BY` repro measures that). Positive
control: an
+ // always-false gate would shuffle these plans for nothing.
+ val a = AttributeReference("a", IntegerType)()
+ val ordered = OrderedDistribution(
+ Seq(org.apache.spark.sql.catalyst.expressions.SortOrder(a,
Review Comment:
**Finding 21.** Four names are fully qualified inline where this file
imports: `SortOrder` and `Ascending` here, `InternalRow` on 460/463/464, and
`SQLConf` on 467. `InternalRow` needs nothing at all — the suite is in
`org.apache.spark.sql.catalyst`, and the sibling test added in the same commit
writes `InternalRow(1)`. The other three want two import lines: `Ascending,
SortOrder` added to the existing `catalyst.expressions` import, and `import
org.apache.spark.sql.internal.SQLConf`.
The suite also has `checkSatisfied` for exactly this assertion, and it
prints the partitioning and the distribution on failure:
```scala
val a = AttributeReference("a", IntegerType)()
val ordered = OrderedDistribution(Seq(SortOrder(a, Ascending)))
val markedOne = KeyedPartitioning(Seq(a), Seq(InternalRow(1)))
.copy(mayContainUnknownPartitionKeys = true)
val markedTwo = KeyedPartitioning(Seq(a), Seq(InternalRow(1),
InternalRow(2)))
.copy(mayContainUnknownPartitionKeys = true)
withSQLConf(SQLConf.V2_BUCKETING_SORTING_ENABLED.key -> "true") {
checkSatisfied(markedOne, ordered, true)
checkSatisfied(markedTwo, ordered, false)
}
```
That drops your two assertion messages, so keep the plain `assert`s if you
prefer them. The imports are the substance.
--
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]
