peter-toth commented on code in PR #58338:
URL: https://github.com/apache/spark/pull/58338#discussion_r3872842072
##########
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:
Kept both values: "the guard answers the same either way" is the claim this
test makes, and the `false` iteration is its control. I added a comment saying
so, so the overlap with the SPARK-46367 scenario does not read as accidental.
##########
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:
Done. The test now builds on `items`/`purchases` and
`selectWithMergeJoinHint`, with the same data as the neighbouring `SPARK-46367:
narrowing projection requires allowKeysSubsetOfPartitionKeys` test (id=1
mapping to two partitions), so the two are directly comparable.
##########
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:
Done - one `allowKeysSubsetOfPartitionKeys` val per `ClusteredDistribution`
arm, so both branches visibly read the same switch.
--
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]
