pan3793 commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3718381898
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -2246,7 +2242,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
Review Comment:
Restored the override in 25c15cb1a06, so the bucket-count comparison is
exercised again. Audited the remaining removed overrides for the same pattern:
the only other negative test ("Compatible buckets does not support SPJ with
push-down values or partially-clustered") has both join keys covered by its
partition keys and a passing SPJ branch, so no other test lost its purpose.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1105,7 +1105,9 @@ object SQLConf {
.doc("When true, the planner requires all the clustering keys as the
hash partition keys " +
"of the children, to eliminate the shuffles for the operator that
needs its children to " +
"be co-partitioned, such as JOIN node. This is to avoid data skews
which can lead to " +
- "significant performance regression if shuffles are eliminated.")
+ "significant performance regression if shuffles are eliminated. For V2
data source " +
+ "partitioning (storage-partitioned join), the check ignores key order
and duplicated " +
+ "clustering keys: it requires every clustering key to be covered by
the partition keys.")
Review Comment:
Adopted the suggested wording in d3b42c82a80 (keeping a short note that hash
partitioning deliberately keeps the positional match), and rewrote the
migration entry, the tuning-guide row, and the PR description around the two
reachable shapes -- the multi-transform column and the subset direction. The
duplicated-join-key story is demoted to a planner-robustness note: the
hand-built test is retitled, and its comment points out that
`BooleanSimplification` dedups the conjunction but is excludable via
`spark.sql.optimizer.excludedRules`, the one configuration where the shape is
query-reachable.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1016,6 +1001,32 @@ class EnsureRequirementsSuite extends SharedSparkSession
{
}
}
+ test("KeyedPartitioning: duplicated join keys do not block SPJ") {
+ // The coverage check of requireAllClusterKeysForCoPartition ignores key
order and
+ // duplicated cluster keys: join keys [a, b, b] are fully covered by
partition keys
+ // on [a, b], so SPJ is allowed with either config value.
+ val plan1 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprB) :: Nil,
Seq.empty))
+ val plan2 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprC) :: Nil,
Seq.empty))
+ val smjExec = SortMergeJoinExec(
+ exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
Review Comment:
Added both tests in d3b42c82a80: the `[a, b]` / `[a, c]` join against the
`(years(a), bucket(4, b), days(a))` partitionings in `EnsureRequirementsSuite`,
and the end-to-end `PARTITIONED BY (bucket(8, id), identity(id))` join in
`KeyGroupedPartitioningSuite`, both with default configs. Verified both fail
against the base gate and pass with the coverage check.
--
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]
