dongjoon-hyun commented on PR #58351:
URL: https://github.com/apache/spark/pull/58351#issuecomment-5442589807
Thanks for the detailed writeup. I read through the main changes and the
tests, and I think the core direction is right: the gate's comment described
*collapse* while the flag recorded *provenance*, and `distinct-after <
distinct-before` is exactly the right predicate -- it is true iff two keys that
were distinct in the input map onto the same projected key. Filling in the
producers that were laundering the flag (`toGrouped`, `createShuffleSpec`,
`KeyedShuffleSpec.createPartitioning`, `GroupPartitionsExec`) is a good pickup
from #58316 too.
One property worth noting, because it makes the new class doc read
consistently: a projection preserves the partition count, so `distinct-after <
distinct-before <= numPartitions` means a freshly computed `isCollapsed` always
implies `!isGrouped`. The `!isGrouped` term in the gate therefore only does
work for a partitioning that inherited a sticky `isCollapsed` and was then
re-grouped -- by `GroupPartitionsExec` or by reducing keys onto a coarser
transform -- which is precisely what the new doc says. Good.
A few things below.
### 1. One of this PR's own producers breaks the doc rule the PR adds
The class doc gains:
> A producer that projects or merges the members has to take `isCollapsed`
from all of them, not from one
but `GroupPartitionsExec.outputPartitioning` projects the members and reads
the flag per member:
```scala
val isCollapsed = k.isCollapsed || projectedDistinctKeyCount <
k.distinctKeyCount
```
The comment right above it says "There can be multiple `KeyedPartitioning`s
in an output partitioning of a join", so a collection here is an anticipated
input. Meanwhile `PartitioningPreservingUnaryExecNode` does
`kps.exists(_.isCollapsed)` in the same situation. The two producers disagree,
and the new unit test ("isCollapsed is taken from every member of a
PartitioningCollection") pins only the `ProjectExec` side.
### 2. `isCollapsed` looks like a property of the physical layout, not of a
member
Members of a `PartitioningCollection` describe the same physical
partitioning and share the `partitionKeys` reference. If the left leg was
coarsened, an output partition really does cover several partitions of the
finer left layout, whichever side's expressions you name it by. So members
disagreeing on the flag seems like the anomaly rather than something to
preserve.
It also leaves the gate bypassable. `EnsureRequirements` uses:
```scala
val nonGroupedSatisfiesWhenGrouped =
nonGrouped.find(_.groupedSatisfies(distribution))
```
`find`, so an ungrouped collection holding one coarsened and one plain
member is accepted via the plain one and gets a `GroupPartitionsExec`. The
reachable path is narrow (a join over a join, made ungrouped by a padding
`GroupPartitionsExec`), so I do not think it blocks this PR -- but the doc
presents it as intended ("the satisfaction path is separate and accepts when
any single member does"), and to me it reads more like a remaining hole than a
design choice.
Suggestion: normalize `isCollapsed` by OR across members in
`PartitioningCollection.fromPartitionings` / `checkKeyedPartitioningInvariant`,
the same way `partitionKeys` references are interned. That would remove the
producer disagreement in (1) and this bypass at once, and let the doc drop both
caveats.
### 3. `KeyedShuffleSpec.createPartitioning` -- question
The shuffled side inherits `partitioning.isCollapsed`. Under the new
definition ("coarser than the layout it was derived from") the shuffled side
has no finer layout it was derived from: its key-1 partition holding all of its
key-1 rows is just what a `HashPartitioning` would give. The direction is safe
(it can only add shuffles), but the flag is sticky, so it can cost a refusal
further up with nothing behind it. Is this deliberate conservatism, or is the
intent to widen the definition? Either way it would help to say so at the call
site, since the comment there argues from the coarsened side's keys rather than
from this side's history.
### 4. Comment volume (nit)
The 11-line comment above the single `false` in `groupedSatisfies` overlaps
heavily with the new "Coarsened Partitionings" section in the class doc.
Trimming the inline one to a few lines and leaving the rest in the doc would be
easier to read. The doc section itself is worth having -- especially the
rationale for keeping a coarsened partitioning instead of dropping it to
`UnknownPartitioning`, which was nowhere in the code before.
### Checked, no issues
- No `isNarrowed` references left; every producer goes through the 4-arg
constructor.
- The `isGrouped` short-circuit in `distinctKeyCount`, and the evaluation
order in `PartitioningPreservingUnaryExecNode` (inherited flag -> position
dropped -> distinct count), are both correct.
- Comparing against this side's own key count rather than the aligned key
list is well argued (filtering is pruning, padding is repetition), and the new
end-to-end test pins it.
- The doc claim that `OrderedDistribution` is not gated checks out: that
path goes through `distributePartitions = true`, where `padTo` gives one
partition per split and nothing is coalesced.
--
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]
