peter-toth commented on PR #58316:
URL: https://github.com/apache/spark/pull/58316#issuecomment-5437215882
@dongjoon-hyun I filed SPARK-58974 for a related but separate defect in the
same guard (#58338), and I want to be transparent about how the two relate.
The `isNarrowed && !isGrouped` check sat in the `requireAllClusterKeys =
false` arm of `groupedSatisfies`, so with
`spark.sql.requireAllClusterKeysForDistribution = true` it never ran at all.
That is what #58338 fixes. The two defects are independent: when I add
`spark.sql.requireAllClusterKeysForDistribution = true` to your new test, it
fails on your current head (`8c7e66d53bd`) - the aggregate coalesces the
narrowed keys with `allowKeysSubsetOfPartitionKeys` off, one shuffle instead of
two. Unmodified, your test passes. Your change fixes the value the check reads;
mine fixes whether the check runs at all.
I first had both fixes in one PR and then dropped yours from it, so #58338
is only the hoist. The reason is that I do not think this PR is the right fix.
`toGrouped`'s result never becomes a node's `outputPartitioning` - its only
caller reads the keys, types and `numPartitions` to build the physical
`KeyGroupedPartitioner` - so that half is not observable.
`GroupPartitionsExec.outputPartitioning` is the node that actually rebuilds the
partitioning after grouping, and it drops the flag too, which this PR does not
cover.
The deeper point, which I also wrote into #58338's description: the
`isNarrowed` contract itself is the problem. The flag records provenance
("positions were dropped, or my input was already narrowed"), while the guard's
own comment describes collapse ("partitions that held distinct keys in the
original finer-grained partitioning"). `!isGrouped` has causes that have
nothing to do with narrowing - a source that reports several splits per
partition key, or a union whose children have distinct keys individually and
repeat keys across children. So restoring the flag makes the guard refuse in
shapes where grouping would merge only same-key partitions, which needs no
opt-in. In my measurements that costs an extra shuffle, and in one
partially-clustered case also a `GroupPartitionsExec` above the join.
So I am following up with the contract fix: the flag will mean actual
collapse - set only when a projection loses distinct keys, sticky from there on
- which is decidable at the one site that projects the keys, since both key
lists are in hand there. `isNarrowed` arrived in 4.3.0 and 4.3 is unreleased,
so we can still correct what it means rather than preserve it, and I am aiming
to get both the hoist and the contract fix into 4.3. That is the direction I
would take instead of this PR: with collapse semantics the flag is false in
exactly the shapes where restoring it changes a decision today, so the
propagation stops mattering. I will open it under its own JIRA and link it here.
--
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]
