You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/08/28 07:43:01 UTC
(spark) branch branch-4.3 updated: [SPARK-58974][SQL] Apply the narrowed-partitioning skew guard regardless of requireAllClusterKeysForDistribution
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new ed82c618223e [SPARK-58974][SQL] Apply the narrowed-partitioning skew guard regardless of requireAllClusterKeysForDistribution
ed82c618223e is described below
commit ed82c618223e21a7a8a1ab02e6c6e14f13034a69
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Fri Aug 28 08:54:47 2026 +0200
[SPARK-58974][SQL] Apply the narrowed-partitioning skew guard regardless of requireAllClusterKeysForDistribution
`KeyedPartitioning.groupedSatisfies` refuses a narrowed, non-grouped partitioning unless `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` is set, because grouping it would merge partitions that held distinct keys in the original finer-grained partitioning. That guard lived inside the `requireAllClusterKeys = false` arm, so it never ran when `spark.sql.requireAllClusterKeysForDistribution` was enabled.
This moves the guard above the `requireAllClusterKeys` check. The skew risk does not depend on which key sets count as matching, so the decision should not either.
The config doc for `allowKeysSubsetOfPartitionKeys.enabled` is updated as well: it did not mention this second thing the config gates, which is not tied to `requireAllClusterKeysForDistribution` being false.
With `spark.sql.requireAllClusterKeysForDistribution = true` a narrowed partitioning was accepted, `EnsureRequirements` inserted a `GroupPartitionsExec`, and the skew exposure was taken with the opt-in config still off. Note this is the *stricter* of the two settings, which makes it the more surprising direction.
Reproduced with a table partitioned by `(id, dept)` projected down to `id` -- keys collapsing to `[1, 1, 2]` -- and joined on `id`: with `requireAllClusterKeys = true` the plan gets a `GroupPartitionsExec` and no shuffle while `allowKeysSubsetOfPartitionKeys` is off, whereas with `requireAllClusterKeys = false` the same query correctly shuffles both sides.
Please note that the guard's condition is imprecise, and the hoist makes that imprecision reachable for one more config value. `isNarrowed && !isGrouped` is a proxy for "the narrowing collapsed distinct keys", but `!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. Grouping such a partitioning merges only same-key partitions, wh [...]
Yes, a plan-level change. With `requireAllClusterKeysForDistribution` enabled, Spark previously coalesced partitions derived from a narrowed partitioning without `allowKeysSubsetOfPartitionKeys.enabled`, risking skewed partitions; it now inserts a shuffle unless that config is enabled. Query results are unchanged.
No migration guide entry: the guard, and with it the bypass, arrived in 4.3.0, which is unreleased, so no released version behaves the old way. This goes to `branch-4.3` as well.
* New unit test in `ProjectedOrderingAndPartitioningSuite` calling `groupedSatisfies` directly on a narrowed, ungrouped partitioning, for both values of `requireAllClusterKeys` and both values of the opt-in. It fails on master with `kp.groupedSatisfies(required) was true` for `requireAllClusterKeys = true`.
* New end-to-end test in `KeyGroupedPartitioningSuite` asserting the guard behaves identically for both values of `requireAllClusterKeys`: no `GroupPartitionsExec`, and a shuffle instead. It fails on master for `requireAllClusterKeys = true`. It also varies `v2BucketingShuffleEnabled`, because that decides whether the refused side is laid out on the other side's declared partition keys (one shuffle) or both sides are shuffled (two).
* That test also covers `allowKeysSubsetOfPartitionKeys = true` for both values of `requireAllClusterKeys`, since this is the first time the opt-in affects `groupedSatisfies` when `requireAllClusterKeysForDistribution` is enabled -- it must restore the coalescing and avoid the shuffles. Neither suite had any coverage of `requireAllClusterKeysForDistribution`, which is how the bug survived.
* Also ran `DistributionSuite`, `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`, `PlannerSuite`, `ProjectedOrderingAndPartitioningSuite`, `DataFrameSetOperationsSuite`, `AdaptiveQueryExecSuite`, `CoalesceShufflePartitionsSuite`, and the TPC-DS plan stability suites -- no golden file changed.
Generated-by: Claude Code (Opus 5)
Closes #58338 from peter-toth/SPARK-58974-narrowing-guard-requireallclusterkeys.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
(cherry picked from commit 9d33e01d8472f207e35b93b7cb2225de5968a9e8)
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../sql/catalyst/plans/physical/partitioning.scala | 65 ++++++++++++------
.../org/apache/spark/sql/internal/SQLConf.scala | 10 +--
.../connector/KeyGroupedPartitioningSuite.scala | 79 ++++++++++++++++++++++
.../ProjectedOrderingAndPartitioningSuite.scala | 32 +++++++++
4 files changed, 161 insertions(+), 25 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
index 12511c1c4224..688968bda16b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
@@ -478,15 +478,23 @@ case class CoalescedNullAwareHashPartitioning(
* unique partition keys, or (2) `GroupPartitionsExec` coalesces partitions with duplicate keys.
*
* == Distribution Satisfaction and Grouping ==
- * Besides the default `satisfies()`, `KeyedPartitioning` exposes two additional methods used by
- * `EnsureRequirements` to handle grouped and non-grouped KPs separately:
+ * Besides the default `satisfies()`, `KeyedPartitioning` exposes two additional methods:
*
- * - `nonGroupedSatisfies()`: called on non-grouped KPs to check as-is satisfaction (without
- * inserting `GroupPartitionsExec`).
- * - `groupedSatisfies()`: called on non-grouped KPs to check whether inserting
- * `GroupPartitionsExec` would satisfy the distribution. When it returns true, the distribution
- * is NOT yet satisfied -- `EnsureRequirements` will insert `GroupPartitionsExec` to coalesce
- * duplicate partition keys.
+ * - `nonGroupedSatisfies()`: as-is satisfaction (without inserting `GroupPartitionsExec`). It is
+ * the default `Partitioning` implementation, so for a `ClusteredDistribution` it is always false.
+ * - `groupedSatisfies()`: whether the distribution would be satisfied once the duplicate partition
+ * keys are coalesced. It has two callers, and they ask different questions:
+ * - `EnsureRequirements` calls it on non-grouped KPs to ask whether inserting a
+ * `GroupPartitionsExec` would help. When it returns true, the distribution is NOT yet
+ * satisfied -- `EnsureRequirements` will insert one to coalesce the duplicate keys.
+ * - `satisfies0()` calls it on grouped KPs. Since `nonGroupedSatisfies()` is false for a
+ * `ClusteredDistribution`, this is the only route by which a grouped KP satisfies one, and no
+ * grouping is involved: the keys are already unique.
+ *
+ * That second caller is why the narrowing guard in `groupedSatisfies()` is a conjunction with
+ * `!isGrouped`. A narrowed KP whose projected keys stayed distinct is grouped, and dropping the
+ * `!isGrouped` term would stop it from satisfying a `ClusteredDistribution` and cost it a shuffle,
+ * even though grouping it would merge nothing.
*
* For `OrderedDistribution`, `GroupPartitionsExec` must also sort the partition keys to meet the
* ordering requirement.
@@ -521,11 +529,15 @@ case class CoalescedNullAwareHashPartitioning(
* @param isGrouped Whether partition keys are unique (no duplicates). Computed on first
* creation, then preserved through copy operations to avoid recomputation.
* @param isNarrowed Whether this partitioning was derived from a finer-grained one by dropping key
- * positions (e.g. via `PartitioningPreservingUnaryExecNode`). When true,
- * `GroupPartitionsExec` will merge partitions that shared distinct keys in the
- * original partitioning, carrying the same skew risk as
- * `allowKeysSubsetOfPartitionKeys`. Such a partitioning will not satisfy
- * `ClusteredDistribution` unless that config is enabled.
+ * positions (e.g. via `PartitioningPreservingUnaryExecNode`). When true and the
+ * keys are no longer unique, `GroupPartitionsExec` may merge partitions that held
+ * distinct keys in the original partitioning, carrying the same skew risk as
+ * `allowKeysSubsetOfPartitionKeys`. "May", because the condition is a proxy: the
+ * duplicate keys can also come from a source that reports several splits per
+ * partition key, in which case grouping merges only same-key partitions. Such a
+ * partitioning can only satisfy `ClusteredDistribution` by being grouped, and
+ * `groupedSatisfies` refuses that unless the config is enabled, regardless of
+ * `requireAllClusterKeysForDistribution`.
*/
case class KeyedPartitioning(
expressions: Seq[Expression],
@@ -579,7 +591,24 @@ case class KeyedPartitioning(
def groupedSatisfies(required: Distribution): Boolean = {
required match {
case c @ ClusteredDistribution(requiredClustering, requireAllClusterKeys, _, _) =>
- if (requireAllClusterKeys) {
+ // Both branches below are gated by the same switch, so read it once.
+ val allowKeysSubsetOfPartitionKeys =
+ SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys
+
+ if (isNarrowed && !isGrouped && !allowKeysSubsetOfPartitionKeys) {
+ // A narrowed, non-grouped partitioning carries the same skew risk as using a subset of
+ // partition keys for a join: GroupPartitionsExec may merge partitions that held distinct
+ // keys in the original finer-grained partitioning. Require the same config to opt in.
+ //
+ // Checked before the `requireAllClusterKeys` branch, because the risk is independent of
+ // which key sets count as matching (SPARK-58974).
+ //
+ // The `!isGrouped` term is required, not redundant: a narrowing projection whose
+ // projected keys stayed distinct is grouped, and `satisfies0` routes such a partitioning
+ // through here (`nonGroupedSatisfies` is false for a `ClusteredDistribution`). Grouping
+ // it would merge nothing, so refusing it would only cost a shuffle.
+ false
+ } else if (requireAllClusterKeys) {
// Checks whether this partitioning is partitioned on exactly same clustering keys of
// `ClusteredDistribution`.
c.areAllClusterKeysMatched(expressions)
@@ -587,17 +616,11 @@ case class KeyedPartitioning(
// We'll need to find leaf attributes from the partition expressions first.
lazy val attributes = AttributeSet.fromAttributeSets(expressions.map(_.references))
- if (SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys) {
+ if (allowKeysSubsetOfPartitionKeys) {
// check that operation keys (required clustering keys)
// overlap with partition keys (KeyedPartitioning attributes)
requiredClustering.exists(x => attributes.exists(_.semanticEquals(x))) &&
expressions.forall(_.references.size == 1)
- } else if (isNarrowed && !isGrouped) {
- // A narrowed, non-grouped partitioning carries the same skew risk as using a subset of
- // partition keys for a join: GroupPartitionsExec will merge partitions that held
- // distinct keys in the original finer-grained partitioning. Require the same config to
- // opt in.
- false
} else {
attributes.forall(x => requiredClustering.exists(_.semanticEquals(x)))
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 1a67e8cc0591..dca3113edf0d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -2499,10 +2499,12 @@ object SQLConf {
.withAlternative("spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled")
.doc("Whether to allow storage-partitioned operations (joins, aggregates, and windows) in " +
"the case where the operation's keys are a subset of the partition keys of the source " +
- "tables. At planning time, Spark will group the partitions by only those keys that are " +
- "in the operation's keys. " +
- s"This is currently enabled only if ${REQUIRE_ALL_CLUSTER_KEYS_FOR_DISTRIBUTION.key} " +
- "is false."
+ "tables. At planning time, Spark will group the partitions by only those keys that are " +
+ "in the operation's keys. That is currently enabled only if " +
+ s"${REQUIRE_ALL_CLUSTER_KEYS_FOR_DISTRIBUTION.key} is false. This config also gates " +
+ "grouping a partitioning that was narrowed to a subset of its keys and whose keys are no " +
+ "longer distinct, which carries the same risk of skew; that applies regardless of " +
+ s"${REQUIRE_ALL_CLUSTER_KEYS_FOR_DISTRIBUTION.key}."
)
.version("4.0.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index c406f6e0ad02..c8cc20b5c42d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -33,6 +33,7 @@ import org.apache.spark.sql.connector.expressions.Expressions._
import org.apache.spark.sql.execution.{
ExtendedMode,
FormattedMode,
+ ProjectExec,
RDDScanExec,
SimpleMode,
SortExec,
@@ -4843,4 +4844,82 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-58974: the narrowing skew guard applies regardless of requireAllClusterKeys") {
+ // Same shape as "SPARK-46367: narrowing projection requires allowKeysSubsetOfPartitionKeys":
+ // items is partitioned by (id, name) and id=1 maps to two of its partitions, so projecting
+ // `name` away narrows [id, name] to [id] and collapses the keys to [1, 1, 2].
+ //
+ // The narrowing guard describes a skew risk that does not depend on which key sets count as
+ // matching, so it must apply for either value of `requireAllClusterKeys`. It used to sit inside
+ // the `requireAllClusterKeys = false` arm of `groupedSatisfies`, so with that setting enabled
+ // such a partitioning was grouped anyway -- taking exactly the exposure
+ // `allowKeysSubsetOfPartitionKeys` exists to gate, with nobody opting in.
+ createTable(items, itemsColumns, Array(identity("id"), identity("name")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'bb', 41.0, cast('2020-01-01' as timestamp)), " +
+ s"(2, 'cc', 10.0, cast('2020-01-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(2, 11.0, cast('2020-01-01' as timestamp))")
+
+ val query =
+ s"""
+ |${selectWithMergeJoinHint("sub", "p")}
+ |sub.id, p.price AS purchase_price
+ |FROM (SELECT id FROM testcat.ns.$items WHERE name >= 'aa') sub
+ |JOIN testcat.ns.$purchases p
+ |ON sub.id = p.item_id
+ |""".stripMargin
+ val expected = Seq(Row(1, 42.0), Row(1, 42.0), Row(2, 11.0))
+
+ for {
+ requireAll <- Seq(true, false)
+ allowSubset <- Seq(true, false)
+ keyedShuffle <- Seq(true, false)
+ } {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_DISTRIBUTION.key -> requireAll.toString,
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> keyedShuffle.toString,
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
+ allowSubset.toString) {
+ val settings = s"requireAllClusterKeys=$requireAll, v2BucketingShuffle=$keyedShuffle"
+ val df = sql(query)
+ val plan = df.queryExecution.executedPlan
+
+ val narrowed = collect(plan) { case p: ProjectExec => p }
+ .map(_.outputPartitioning)
+ .collect { case kp: physical.KeyedPartitioning if kp.isNarrowed => kp }
+ assert(narrowed.exists(!_.isGrouped),
+ "this test needs a narrowed, ungrouped partitioning to be meaningful")
+
+ // Count over the whole plan, so a shuffle or a grouping anywhere in it is caught, not
+ // only the ones inside the join subtree.
+ val groupPartitions = collectAllGroupPartitions(plan)
+ val shuffles = collectAllShuffles(plan)
+ if (allowSubset) {
+ // The opt-in is the only way to keep the coalescing, and it must work for either value
+ // of `requireAllClusterKeys` -- before the fix it was reachable only for `false`.
+ assert(groupPartitions.nonEmpty, s"$settings: the opt-in must restore the coalescing")
+ assert(shuffles.isEmpty, s"$settings: the opt-in must avoid the shuffles")
+ } else {
+ // `requireAllClusterKeys` says which key sets count as matching; it does not authorise
+ // coalescing a narrowed partitioning, so the guard behaves the same either way.
+ assert(groupPartitions.isEmpty,
+ s"$settings must not coalesce a narrowed partitioning without " +
+ "allowKeysSubsetOfPartitionKeys")
+ // Both sides are shuffled, unless `v2BucketingShuffleEnabled` lets the refused side be
+ // laid out on the other side's declared partition keys.
+ val expectedShuffles = if (keyedShuffle) 1 else 2
+ assert(shuffles.size == expectedShuffles,
+ s"$settings must shuffle instead, on $expectedShuffles side(s)")
+ }
+ checkAnswer(df, expected)
+ }
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
index f1f2bb7e0833..408f116540bb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
@@ -491,6 +491,38 @@ class ProjectedOrderingAndPartitioningSuite
}
}
+ test("SPARK-58974: the narrowing guard applies for either value of requireAllClusterKeys") {
+ val x = AttributeReference("x", IntegerType)()
+ val y = AttributeReference("y", IntegerType)()
+
+ // The projected keys have duplicates (x-values: 1, 1, 2), so grouping this partitioning merges
+ // two partitions that held distinct keys. `requireAllClusterKeys` decides which key sets count
+ // as matching; it does not authorise that merge, so the guard must answer the same either way.
+ val keys = Seq(InternalRow(1, 1), InternalRow(1, 2), InternalRow(2, 1))
+ val project = ProjectExec(Seq(x),
+ DummyLeafExecWithPartitioning(output = Seq(x, y),
+ partitioning = KeyedPartitioning(Seq(x, y), keys)))
+ val kp = project.outputPartitioning.asInstanceOf[KeyedPartitioning]
+ assert(kp.isNarrowed && !kp.isGrouped)
+
+ Seq(true, false).foreach { requireAll =>
+ // Both values on purpose: the whole claim of the fix is that the guard answers the same
+ // either way, so the `false` iteration is the control that must keep behaving as before.
+ // The required clustering is exactly this partitioning's single key, so the
+ // `requireAllClusterKeys` branch on its own would accept it.
+ val required = ClusteredDistribution(Seq(x), requireAllClusterKeys = requireAll)
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "false") {
+ assert(!kp.groupedSatisfies(required),
+ s"requireAllClusterKeys=$requireAll must not group a narrowed partitioning whose keys " +
+ "are no longer distinct")
+ }
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") {
+ assert(kp.groupedSatisfies(required),
+ s"requireAllClusterKeys=$requireAll: the opt-in must allow the grouping")
+ }
+ }
+ }
+
test("SPARK-46367: isNarrowed is sticky across chained PartitioningPreservingUnaryExecNodes") {
val x = AttributeReference("x", IntegerType)()
val y = AttributeReference("y", IntegerType)()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org