dongjoon-hyun commented on code in PR #58338:
URL: https://github.com/apache/spark/pull/58338#discussion_r3872577862
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -579,7 +581,17 @@ case class KeyedPartitioning(
def groupedSatisfies(required: Distribution): Boolean = {
required match {
case c @ ClusteredDistribution(requiredClustering,
requireAllClusterKeys, _, _) =>
- if (requireAllClusterKeys) {
+ if (isNarrowed && !isGrouped &&
+ !SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys) {
+ // 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
Review Comment:
[Nit] "GroupPartitionsExec will merge partitions that held distinct keys in
the original finer-grained partitioning" states the risky case as fact, but
when the source was already ungrouped (multiple splits per identical key) and
the narrowing collapsed nothing, the merged partitions held the *same* original
key. Since `isNarrowed && !isGrouped` is a proxy (as the PR description itself
notes), "may merge" would be accurate here and in the `isNarrowed` scaladoc. As
written, the docs make the guard look exact, which could make the promised
follow-up tightening look unnecessary to a future reader.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -579,7 +581,17 @@ case class KeyedPartitioning(
def groupedSatisfies(required: Distribution): Boolean = {
required match {
case c @ ClusteredDistribution(requiredClustering,
requireAllClusterKeys, _, _) =>
- if (requireAllClusterKeys) {
+ if (isNarrowed && !isGrouped &&
Review Comment:
[Non-blocking] The `!isGrouped` conjunct here is load-bearing in a way the
class-level doc contradicts. The doc above ("`groupedSatisfies()`: called on
non-grouped KPs ...") predates this change, but `satisfies0` also calls
`groupedSatisfies` on *grouped* KPs (`isGrouped &&
groupedSatisfies(required)`), and grouped-but-narrowed KPs are constructible:
`PartitioningPreservingUnaryExecNode` recomputes `isGrouped` from the projected
keys while `isNarrowed` stays sticky, which is exactly the state exercised by
the existing test "SPARK-46367: narrowing projection with distinct projected
keys does not require allowKeysSubsetOfPartitionKeys".
If a later cleanup trusts the class doc and drops the "redundant"
`!isGrouped`, grouped narrowed KPs would stop satisfying
`ClusteredDistribution` with the config off and pick up unnecessary shuffles
(that test would fail). Could you touch up the class-level doc (or add a short
note here) so the conjunct's purpose is recorded?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -4727,4 +4728,76 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
assert(shuffles.isEmpty, "should not contain any shuffle")
checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0), Row(2, "bb", 10.0, 19.5)))
}
+
+ test("SPARK-58974: the narrowing skew guard applies regardless of
requireAllClusterKeys") {
+ // 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 a
+ // partitioning whose narrowing collapsed distinct keys was grouped anyway
-- taking exactly the
+ // exposure `allowKeysSubsetOfPartitionKeys` exists to gate, with nobody
opting in.
+ val cols = Array(
+ Column.create("id", LongType),
+ Column.create("dept", StringType),
+ Column.create("data", StringType))
+ val t2cols = Array(Column.create("id", LongType), Column.create("data",
StringType))
+ withTable("t1", "t2") {
+ createTable("t1", cols, Array(identity("id"), identity("dept")))
+ sql("INSERT INTO testcat.ns.t1 VALUES (1, 'x', 'a1'), (1, 'y', 'a2'),
(2, 'z', 'a3')")
Review Comment:
[Nit] The directly analogous test "SPARK-46367: narrowing projection
requires allowKeysSubsetOfPartitionKeys" builds this same shape from the
suite's standing `items`/`itemsColumns` + `purchases`/`purchasesColumns`
fixtures and the `selectWithMergeJoinHint` helper (it only generates the hint
prefix from alias strings, so the subquery alias `u` works with it too -- the
SPARK-46367 test itself passes `"sub"`). Reusing those here would keep the
pre-fix and post-fix narrowing tests directly comparable instead of introducing
a parallel `t1`/`t2` schema.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -579,7 +581,17 @@ case class KeyedPartitioning(
def groupedSatisfies(required: Distribution): Boolean = {
required match {
case c @ ClusteredDistribution(requiredClustering,
requireAllClusterKeys, _, _) =>
- if (requireAllClusterKeys) {
+ if (isNarrowed && !isGrouped &&
+ !SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys) {
Review Comment:
[Nit] `SQLConf.get.v2BucketingAllowKeysSubsetOfPartitionKeys` is now read
twice in this match arm (here and in the `else` branch below). Hoisting it into
a local val at the top of the `case c @ ClusteredDistribution(...)` arm would
make it explicit that both branches are gated by the same switch, which is the
core of this fix, and would keep a future edit from decoupling the two reads.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala:
##########
@@ -491,6 +491,36 @@ class ProjectedOrderingAndPartitioningSuite
}
}
+ test("SPARK-58974: the narrowing guard applies for either value of
requireAllClusterKeys") {
Review Comment:
[Nit] The `requireAll = false` iteration exactly duplicates Scenario 1 of
the existing test "SPARK-46367: narrowing projection with duplicate keys
requires allowKeysSubsetOfPartitionKeys to satisfy ClusteredDistribution" (same
keys, same `ProjectExec` fixture, same `groupedSatisfies` asserts;
`ClusteredDistribution`'s `requireAllClusterKeys` defaults to false). If the
explicit both-values contrast pair is intentional, fine -- otherwise this test
could cover only `requireAll = true`, or the loop could be folded into the
existing test so the two don't have to be kept in lockstep.
--
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]
