[
https://issues.apache.org/jira/browse/SPARK-58974?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-58974:
-------------------------------
Description:
`KeyedPartitioning.groupedSatisfies` gates a narrowed, non-grouped partitioning
behind `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
because grouping it would merge partitions that held distinct keys in the
original finer-grained partitioning -- the same data-skew exposure that config
exists to opt into.
That guard sits only in the `requireAllClusterKeys = false` arm:
{code}
if (requireAllClusterKeys) {
c.areAllClusterKeysMatched(expressions) // no guard
} else {
if (SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys) {
...
} else if (isNarrowed && !isGrouped) {
false
} else {
...
}
}
{code}
So with `spark.sql.requireAllClusterKeysForDistribution = true`, a narrowed
partitioning whose projection collapsed distinct partition keys is accepted,
`EnsureRequirements` inserts a `GroupPartitionsExec`, and the skew exposure is
taken while `allowKeysSubsetOfPartitionKeys` is still `false`. Nobody opted in.
It is also the more surprising direction, since
`requireAllClusterKeysForDistribution` is the stricter of the two settings.
`GroupPartitionsExec` then rebuilds the partitioning through the 3-argument
`KeyedPartitioning` constructor, so `isNarrowed` defaults back to `false` and
nothing downstream can observe what happened.
Reproduced on master and confirmed present unchanged in branch-4.3 and
branch-4.x. A table partitioned by `(id, dept)`; `SELECT id FROM t WHERE dept
RLIKE '...'` (RLIKE has no V2 translation, so the scan must keep `dept` and the
`Project` above the `Filter` is what narrows the partitioning to `[id]`,
collapsing its keys to `[1, 1, 2]`); joined on `id`. Measured, same query both
ways:
* `requireAllClusterKeysForDistribution = false`: no `GroupPartitionsExec`,
both sides shuffle -- the guard fires, which is correct.
* `requireAllClusterKeysForDistribution = true`: one `GroupPartitionsExec`, no
shuffle, and the resulting partitioning reports `isNarrowed = false`.
The guard should be reachable regardless of `requireAllClusterKeys`, and
`GroupPartitionsExec` should not silently drop the flag.
> requireAllClusterKeys bypasses the narrowed-partitioning skew guard in
> KeyedPartitioning.groupedSatisfies
> ---------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58974
> URL: https://issues.apache.org/jira/browse/SPARK-58974
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Peter Toth
> Priority: Major
>
> `KeyedPartitioning.groupedSatisfies` gates a narrowed, non-grouped
> partitioning behind
> `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`,
> because grouping it would merge partitions that held distinct keys in the
> original finer-grained partitioning -- the same data-skew exposure that
> config exists to opt into.
> That guard sits only in the `requireAllClusterKeys = false` arm:
> {code}
> if (requireAllClusterKeys) {
> c.areAllClusterKeysMatched(expressions) // no guard
> } else {
> if (SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys) {
> ...
> } else if (isNarrowed && !isGrouped) {
> false
> } else {
> ...
> }
> }
> {code}
> So with `spark.sql.requireAllClusterKeysForDistribution = true`, a narrowed
> partitioning whose projection collapsed distinct partition keys is accepted,
> `EnsureRequirements` inserts a `GroupPartitionsExec`, and the skew exposure
> is taken while `allowKeysSubsetOfPartitionKeys` is still `false`. Nobody
> opted in. It is also the more surprising direction, since
> `requireAllClusterKeysForDistribution` is the stricter of the two settings.
> `GroupPartitionsExec` then rebuilds the partitioning through the 3-argument
> `KeyedPartitioning` constructor, so `isNarrowed` defaults back to `false` and
> nothing downstream can observe what happened.
> Reproduced on master and confirmed present unchanged in branch-4.3 and
> branch-4.x. A table partitioned by `(id, dept)`; `SELECT id FROM t WHERE dept
> RLIKE '...'` (RLIKE has no V2 translation, so the scan must keep `dept` and
> the `Project` above the `Filter` is what narrows the partitioning to `[id]`,
> collapsing its keys to `[1, 1, 2]`); joined on `id`. Measured, same query
> both ways:
> * `requireAllClusterKeysForDistribution = false`: no `GroupPartitionsExec`,
> both sides shuffle -- the guard fires, which is correct.
> * `requireAllClusterKeysForDistribution = true`: one `GroupPartitionsExec`,
> no shuffle, and the resulting partitioning reports `isNarrowed = false`.
> The guard should be reachable regardless of `requireAllClusterKeys`, and
> `GroupPartitionsExec` should not silently drop the flag.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
