peter-toth commented on PR #58262: URL: https://github.com/apache/spark/pull/58262#issuecomment-5438398111
@ulysses-you thanks for closing #58245 in favour of this one, and for the three window tests, which I brought over verbatim. Since the two PRs look quite different, here is why I ended up somewhere else rather than iterating on yours. **The root cause is one level up from the branch you patched.** `splitKeyedPartitionings` split the child's `KeyedPartitioning`s by `isGrouped`, and `isGrouped` is not the question that matters here. It only says whether the *full* partition keys are unique. It says nothing about whether the *projected* keys are, and under `allowKeysSubsetOfPartitionKeys` those are different things. A `GroupPartitionsExec` is needed either to coalesce duplicate partition keys, or to project down to the operation keys, or both, and splitting by `isGrouped` puts those two reasons in different branches. So patching one branch leaves the other one wrong. **And the sibling branch was wrong in the same way.** For a `ClusteredDistribution` a non-grouped `KeyedPartitioning` always lands in `case _ => GroupPartitionsExec(child)`, with no positions, because `nonGroupedSatisfies` is `Partitioning.satisfies0` and answers `false` there. I measured it on your head: an `(id, name)`-partitioned table with two splits for `(1,'aa')` and `SUM(price) OVER (PARTITION BY id)` returned 25.0 and 20.0 instead of 45.0. Worth knowing why your tests could not catch it: `WindowGroupLimit Final` also requires a `ClusteredDistribution`, so the projecting node lands *above* it and the wrong grouping underneath is harmless, because a group limit can only keep too many rows. It takes a plain window with no group limit above it. **`isJoin` also turned out to be the wrong predicate, and it regressed a cogroup.** The multi-child block owns the projection for *any* operator with more than one `ClusteredDistribution` child, `FlatMapCoGroupsInPandasExec` included, not only for a `ShuffledJoin`. With `KP([n, i])` and a cogroup on `i`, master grouped on `i` (2 groups) and the PR grouped on `n` (3 groups). The predicate that works is `requiredChildDistributions.count(_.isInstanceOf[ClusteredDistribution]) > 1`, which is what this PR uses. **On the positions themselves.** Taking them from `createShuffleSpec` answers the reference-level question, which is exactly right for a storage-partitioned join: a `bucket(4, a)` transform covers the cluster key `a`. But a cluster key can also be the partition expression *itself*, and then the reference-level lookup drops that position and coalesces for nothing. That is why this PR derives the positions from the required clustering instead, honouring both forms. **One more difference worth naming, and it is plan quality rather than correctness.** Your version decided from a single member, the one `groupedSatisfies` picked out of the child's partitioning. This one asks every member, because a `PartitioningCollection`'s members can disagree about which of their positions are operation keys, and an inner join is where they do: `ShuffledJoin.outputPartitioning` unions the two sides without enumerating the mixed combinations, so a window keyed on one side's first key column and the other side's remaining ones sees one member covering position 0 and one covering the rest. Taking the wrong one is still *correct* - any admitted member's projection satisfies the distribution - but it can insert a node where another member needs none, which costs a `CoalescedRDD` layer and some parallelism. That case is measured in the suite. The extra look is cheaper than it sounds. A position set contained in another one is dropped without projecting it, since projecting to fewer positions can merge partitions but never split them. In the ordinary case one set contains the rest and it is a single projection, same as yours; it is only more when the sets genuinely disagree. -- 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]
