peter-toth commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3713962559
##########
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:
**Finding 5.** Both tables here are partitioned by `bucket(N, store_id)`
only, and the join is `ON t1.store_id = t2.store_id AND t1.dept_id =
t2.dept_id`, so `dept_id` is not covered by either side's partition keys:
`allClusterKeysCovered` is false and `createKeyedShuffleSpec` returns `None`
(`EnsureRequirements.scala:795`) before `areKeysCompatible` compares `bucket(2,
store_id)` with `bucket(3, store_id)`. `V2_BUCKETING_SHUFFLE_ENABLED` is off
here, so `KeyedShuffleSpec.canCreatePartitioning` is false and the
`bestSpecOpt` branch is skipped too — the bucket-count comparison the test is
named for now runs nowhere. `assert(shuffles.nonEmpty)` still holds, so CI
stays green, but the test would keep passing even if `(2, 3)` bucket counts
became compatible.
Note this is specific to the coverage check, not to dropping the override:
in the previous revision `satisfies` alone passed here and the bucket check
*was* reached. This is the same partition-keys-cover-only-part-of-the-join-keys
shape for which you restored the override in the 7 other tests, so it belongs
here as well:
```suggestion
SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
```
##########
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:
**Finding 7.** This now documents two different meanings for one config:
coverage for `KeyedPartitioning`, and the positional exact match that
`HashShuffleSpec.canCreatePartitioning` /
`NullAwareHashShuffleSpec.canCreatePartitioning` still use via
`ClusteredDistribution.areAllClusterKeysMatched` (`partitioning.scala:1142`,
`:1208`). The visible consequence is that the case you are fixing stays broken
for V1 bucketing: join keys `[a, b, b]` against `HashPartitioning(a, b)` fails
`areAllClusterKeysMatched` on the length check, exactly as `KeyedPartitioning`
did before this PR.
Nothing in the skew argument distinguishes the two. In both, `satisfies`
already requires the partition keys to be a subset of the cluster keys
(`partitioning.scala:300`, `:595`), so adding coverage on top means the two
sets are equal — order and duplicates stop mattering and nothing coarser than
the join keys is admitted.
If you want to extend it, a shared helper on the distribution keeps the two
call sites honest and puts this next to the check it parallels rather than in a
local `def` inside the planner rule:
```scala
// ClusteredDistribution, next to areAllClusterKeysMatched
def areAllClusterKeysCovered(expressions: Seq[Expression]): Boolean = {
clustering.forall(c =>
expressions.exists(_.references.exists(_.semanticEquals(c))))
}
```
Otherwise, a sentence saying the V2-only scope is deliberate would settle it
— as written a reader can't tell whether V1 was considered.
--
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]
