[jira] [Commented] (SPARK-58974) requireAllClusterKeys bypasses the narrowed-partitioning skew guard in KeyedPartitioning.groupedSatisfies

["Peter Toth (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-58974?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18108220#comment-18108220 ] 

Peter Toth commented on SPARK-58974:
------------------------------------

I'm working on this.

> 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org