dongjoon-hyun commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4315065547
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5645,6 +5664,63 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59971: a side is laid out on another only if its keys cover
every join key") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+ // The same ids as `items`, so `keyed_purchases` holds the same partition
keys and can pair
+ // with `items` as it stands.
+ val purchaseRows = "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp)), " +
+ "(4, 26.0, cast('2020-03-01' as timestamp))"
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES $purchaseRows")
+ val keyedPurchases = "keyed_purchases"
+ createTable(keyedPurchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$keyedPurchases VALUES $purchaseRows")
+
+ val shuffle = SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key
+ val requireAll = SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key
+ val v2RequireAll =
SQLConf.V2_BUCKETING_REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key
+ // `items` is partitioned by `id` alone, and the join is on `id` and the
time.
+ Seq(
+ // With the requirement on, the join must not run on `id` alone, so both
sides are shuffled.
+ (keyedPurchases, Seq(shuffle -> "true", requireAll -> "true"), 2),
+ (purchases, Seq(shuffle -> "true", requireAll -> "true"), 2),
+ // With it off, the storage-partitioned join pairs the keyed sides as
they stand, and the
+ // unpartitioned side is shuffled onto `items`.
+ (keyedPurchases, Seq(shuffle -> "true", requireAll -> "false"), 0),
+ (purchases, Seq(shuffle -> "true", requireAll -> "false"), 1),
+ // The V2 config lifts it for storage-partitioned layouts alone, both in
the one-side
+ // shuffle and in the storage-partitioned join. Without the one-side
shuffle, only the
+ // storage-partitioned join can keep the keyed sides as they stand.
+ (purchases, Seq(shuffle -> "true", requireAll -> "true", v2RequireAll ->
"false"), 1),
+ (keyedPurchases, Seq(shuffle -> "false", requireAll -> "true",
v2RequireAll -> "false"), 0)
Review Comment:
These rows test the V2 conf in one direction only: set to `false` while the
general conf is on. Nothing runs the opposite setting, the general conf off and
the V2 conf on, through the storage-partitioned join. The docs promise that
setting when they say the V2 conf, once set, decides for storage-partitioned
layouts while hash partitionings follow the general one.
For example, suppose the gate in `createKeyedShuffleSpecs`
(EnsureRequirements.scala:1149) lifted the requirement whenever either conf is
`false`. All six rows would still pass. The first four leave the V2 conf unset,
so it equals the general one, and the last two set it to `false`.
`ShuffleSpecSuite` covers that setting for `canCreatePartitioning` only, and a
catalyst test cannot reach `createKeyedShuffleSpecs`.
Could we add a row like this one?
```scala
(keyedPurchases, Seq(shuffle -> "false", requireAll -> "false", v2RequireAll
-> "true"), 2)
```
By my reading it takes two shuffles on all the join keys, so the
`expectedShuffles == 2` check below applies as well. The last row already
catches a plain revert to the general conf.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1614,8 +1614,11 @@ class KeyGroupedPartitioningSuite
// layout next to `bucket(4, id)`. The second join must not shuffle
`xy` onto the first
// one. That builds `bucket(4, y)` for `xy`, not the bucket of the
same function of `y`. A
// row with `b = y` then lands in a different bucket on each side. The
join order decides
- // which of the two layouts comes first, so both orders run.
- withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // which of the two layouts comes first, so both orders run. Each
layout covers one of the
+ // two join keys, so the co-partition key requirement is turned off.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false") {
Review Comment:
With the conf on, neither member of the first join's output can be the
layout any more, because `[bucket(4, k.id)]` and `[bucket(4, p.b + 1)]` each
cover only one of the two join keys. So by my reading, under the default conf
`xy` is no longer shuffled onto `bucket(4, x)`. Instead, the first join's
output is re-shuffled onto `xy`'s hash layout. No test asserts that plan now.
This is also the only test where a collection covers the join keys only
between its members. The SPARK-59080 and SPARK-59256 tests in
`EnsureRequirementsSuite` also run with the conf on, but each of their
collections has a member that covers both keys and wins either way. So a change
that judged coverage over a whole collection, e.g. over `spec.flatten` in
`pickCoPartitionTarget`, would put `xy` back on `bucket(4, x)` under the
default conf, and no test would fail.
Could we also run this test with the conf on, and assert that both sides of
the second join are shuffled on both of its keys? I found this by reading the
code and have not run it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2101,10 +2118,12 @@ case class IdentityReducer(transform:
TransformExpression) extends Reducer[Any,
* all, and it reaches `None` two ways. An identity
projection over already
* grouped and sorted keys rebuilds the same
partitioning. A marked claim is
* refused outright, since neither narrowing nor
sorting its keys leaves the
- * routing it promises for its undeclared rows. The
first always leaves a
- * spec a consumer can be laid out on, and so does the
second where every
- * partition expression covers a clustering key. A
marked claim that narrows
- * does not: one `keyPositions` entry is empty there,
which
+ * routing it promises for its undeclared rows. The
first leaves a spec a
+ * consumer can be laid out on, and so does the second
where every
+ * partition expression covers a clustering key. With
the co-partition
+ * key requirement on (see
`KeyedPartitioning.referencesAllClusterKeys`),
+ * both also need the keys to cover every clustering
key. A marked claim
Review Comment:
Nit: the inserted sentence now sits between "leaves a spec a consumer can be
laid out on" and the elliptical "A marked claim that narrows does not". So on a
first read, "does not" picks up "need the keys to cover every clustering key"
and reads as an exemption. Only the clause after the colon sets it right.
Could we spell the verb out, e.g. "A marked claim that narrows leaves no
such spec: ...", or move the requirement sentence after "... both turn away."?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2489,11 @@ case class KeyedShuffleSpec(
// cannot rewrite it. This also keeps the unprojected spec returned by
`createShuffleSpec`
// for a marked narrowing projection from being chosen as the best spec.
keyPositions.forall(_.nonEmpty) &&
+ // A child lined up on these keys is co-partitioned only on the cluster
keys they cover. So
+ // the co-partition key requirement applies here too. It uses the
coverage rule of
Review Comment:
Nit: "the co-partition key requirement" here was written when this clause
read `spark.sql.requireAllClusterKeysForCoPartition`, the conf
`HashShuffleSpec` reads. Since b886e54 the two specs read different confs.
Contrasting only the rule with `HashShuffleSpec`'s positional match reads as if
they shared one requirement. Could we say "the V2 co-partition key requirement"
here?
Two more places still name the general conf, or just say "the config":
- `createKeyedShuffleSpecs` starts with "The config requires all the cluster
keys ..." (EnsureRequirements.scala:1142).
- The SPARK-59901 test comment says `b + c` does not cover `b` and `c` "for
`spark.sql.requireAllClusterKeysForCoPartition`"
(KeyGroupedPartitioningSuite.scala:1904-1907), but that admission now reads the
V2 conf.
This PR already renamed the main-code references (L1584,
ValidateRequirements.scala:98, EnsureRequirements.scala:985).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2489,11 @@ case class KeyedShuffleSpec(
// cannot rewrite it. This also keeps the unprojected spec returned by
`createShuffleSpec`
// for a marked narrowing projection from being chosen as the best spec.
keyPositions.forall(_.nonEmpty) &&
+ // A child lined up on these keys is co-partitioned only on the cluster
keys they cover. So
+ // the co-partition key requirement applies here too. It uses the
coverage rule of
+ // `referencesAllClusterKeys`, not the positional match of
`HashShuffleSpec`.
+ (!SQLConf.get.v2BucketingRequireAllClusterKeysForCoPartition ||
+ partitioning.referencesAllClusterKeys(distribution)) &&
Review Comment:
Nit: with this clause, the comment in `pickCoPartitionTarget`
(EnsureRequirements.scala:372-374) no longer holds for keyed members: "Any
other member would build a partitioning its own child is not laid out on." Take
a grouped `KeyedPartitioning(Seq(id))` under the clustering `(id, t1)`. It
passes every other clause here, and its own child is laid out on exactly the
keys `createPartitioning` would build from it. It is turned away as a skew
heuristic, as the new test puts it (ShuffleSpecSuite.scala:727): "The same
specs are accepted without the requirement, so it is what turns them away."
Could we reword that comment, e.g. "... would build a partitioning its own
child is not laid out on, or is turned away by a co-partition key requirement
to avoid skew"? The sentence was already loose for `HashShuffleSpec`, which
refuses for the same reason. The comment itself is outside the diff, so I left
this here.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2489,11 @@ case class KeyedShuffleSpec(
// cannot rewrite it. This also keeps the unprojected spec returned by
`createShuffleSpec`
// for a marked narrowing projection from being chosen as the best spec.
keyPositions.forall(_.nonEmpty) &&
+ // A child lined up on these keys is co-partitioned only on the cluster
keys they cover. So
+ // the co-partition key requirement applies here too. It uses the
coverage rule of
+ // `referencesAllClusterKeys`, not the positional match of
`HashShuffleSpec`.
+ (!SQLConf.get.v2BucketingRequireAllClusterKeysForCoPartition ||
Review Comment:
Minor: the whole gate is spelled out here and again in
`createKeyedShuffleSpecs` (EnsureRequirements.scala:1149-1150). So the shared
helper carries only the coverage predicate, not the conf that decides whether
it applies. b886e54 had to switch the conf at both places in lockstep. A third
caller that copied `HashShuffleSpec`'s
`getConf(REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION)` would let the
storage-partitioned join and the one-side shuffle disagree again.
Since you'd keep the shared helper so that both places stay on one rule,
could the conf read move into it as well? `mayProjectToClusterKeys` does the
same for `v2BucketingAllowKeysSubsetOfPartitionKeys`, since "the permission has
to be one rule rather than two spellings". For example:
```scala
def meetsCoPartitionKeyRequirement(required: ClusteredDistribution): Boolean
=
!SQLConf.get.v2BucketingRequireAllClusterKeysForCoPartition ||
referencesAllClusterKeys(required)
```
`createKeyedShuffleSpecs` can still ask it before `keysMaySatisfy`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2596,6 +2600,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val V2_BUCKETING_REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION =
+
buildConf("spark.sql.sources.v2.bucketing.requireAllClusterKeysForCoPartition.enabled")
+ .doc("When true, a storage-partitioned join, or the one-side shuffle of
" +
+ s"${V2_BUCKETING_SHUFFLE_ENABLED.key}, runs on the partition keys only
if they cover " +
+ "every join or MERGE key. Set it to false to also allow partition keys
that cover only " +
Review Comment:
This says the conf is about "every join or MERGE key", but through
`KeyedShuffleSpec.canCreatePartitioning` it also decides whether a cogroup runs
on a keyed side. `checkKeyGroupCompatible` handles only `SortMergeJoinExec` and
`ShuffledHashJoinExec`, so a `FlatMapCoGroupsInPandasExec` or a `CoGroupExec`
reaches only `pickCoPartitionTarget`. The new cogroup test in
`EnsureRequirementsSuite` pins exactly that, and the migration guide says "A
cogroup changes the same way".
Since `spark.sql.requireAllClusterKeysForCoPartition` is internal, this is
the only description that `SET -v` and the generated configuration page show.
Could we say "every join, cogroup or MERGE key" here and in the new row at
sql-performance-tuning.md:663?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2596,6 +2600,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val V2_BUCKETING_REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION =
+
buildConf("spark.sql.sources.v2.bucketing.requireAllClusterKeysForCoPartition.enabled")
+ .doc("When true, a storage-partitioned join, or the one-side shuffle of
" +
+ s"${V2_BUCKETING_SHUFFLE_ENABLED.key}, runs on the partition keys only
if they cover " +
+ "every join or MERGE key. Set it to false to also allow partition keys
that cover only " +
+ "some of those keys, at the risk of data skew and reduced parallelism
from the coarser " +
+ "storage partitioning. Hash partitionings still follow " +
+ s"${REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key}. Defaults to that
config.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
Review Comment:
Nit: this conf is read only during physical planning
(`KeyedShuffleSpec.canCreatePartitioning` and
`EnsureRequirements.createKeyedShuffleSpecs`), so it cannot change what a view,
UDF or procedure body resolves to. Step 1 of the `ConfigBindingPolicy` scaladoc
says to use `NOT_APPLICABLE` in that case. SPARK-57377 relabeled the
`spark.sql.window.segmentTree.*` configs from `SESSION` to `NOT_APPLICABLE` for
the same reason. The neighboring `v2.bucketing` confs that use `SESSION`
predate that guidance.
It makes no difference at runtime, since the two are handled alike. Could we
use `ConfigBindingPolicy.NOT_APPLICABLE` here?
--
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]
