ulysses-you commented on code in PR #58552:
URL: https://github.com/apache/spark/pull/58552#discussion_r3974961050
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -2598,6 +2598,89 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59285: two legs whose struct field names differ are still
co-partitioned") {
+ // Both legs prune to nothing, and their key spaces differ only in a
struct field name, which
+ // `identity` carries into the key type. A reduce cannot bridge that,
since there is no reducer
+ // between two attributes, so calling the two sides incompatible leaves
nowhere to go: the join
+ // must keep taking them as one layout.
+ withTable("p1", "p2", "p3", "p4") {
+ createTable("p1", Array(Column.create("id", structA)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.p1 VALUES (named_struct('a', 1))")
+ createTable("p2", Array(Column.create("id", structA)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.p2 VALUES (named_struct('a', 2))")
+ createTable("p3", Array(Column.create("k", structB)),
Array(identity("k")))
+ sql("INSERT INTO testcat.ns.p3 VALUES (named_struct('b', 1))")
+ createTable("p4", Array(Column.create("k", structB)),
Array(identity("k")))
+ sql("INSERT INTO testcat.ns.p4 VALUES (named_struct('b', 2))")
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ val df = sql(
+ """SELECT leg1.id, leg2.k FROM
+ | (SELECT p1.id AS id FROM testcat.ns.p1 JOIN testcat.ns.p2 ON
p1.id = p2.id) leg1
+ | JOIN
+ | (SELECT p3.k AS k FROM testcat.ns.p3 JOIN testcat.ns.p4 ON p3.k
= p4.k) leg2
+ | ON leg1.id = leg2.k
+ |""".stripMargin)
+ assert(collectShuffles(df.queryExecution.executedPlan).isEmpty,
+ "the two legs are taken as one layout, so neither is shuffled")
+ checkAnswer(df, Nil)
+ }
+ }
+ }
+
+ test("SPARK-59285: two sides whose partitions were all pruned are not one
layout") {
Review Comment:
Non-blocking (test-coverage): nothing pins the claimed mechanism ("It
reduces the identity side onto the bucket key space instead"). The main assert
is an assert-on-absence: if a future regression made this shape fall back to
plain hash shuffles, the legs join would report `HashPartitioning`,
`keyedPartitioningsOf` would return empty lists, `distinct.size` would be 0,
and the assert passes vacuously - as do `ValidateRequirements.validate` and
`checkAnswer`. It discriminates against the bug (on master the members report
`[LongType]` and `[IntegerType]` via the expression fallback, so `distinct ==
2`) but not against the degenerate fix.
The sibling struct-names test pins plan shape with
`collectShuffles(...).isEmpty`; the same assert holds here (the legs join
aligns via `IdentityReducer` GPEs, the FULL OUTER via expected-keys GPEs - zero
shuffles), or positively pin that the legs join reports exactly one non-empty
key space (`distinct == Seq(Seq(IntegerType))`).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1687,6 +1744,13 @@ case class KeyedShuffleSpec(
case otherSpec @ KeyedShuffleSpec(otherPartitioning, otherDistribution, _)
=>
distribution.clustering.length == otherDistribution.clustering.length &&
numPartitions == other.numPartitions && areKeysCompatible(otherSpec) &&
+ // The key rows are compared at their types, since
`InternalRowComparableWrapper.equals`
+ // compares those first. Two empty key lists compare equal whatever
they describe, so the
+ // key space is asked separately: without that, a join between two
sides whose partitions
+ // were all pruned would call two different spaces one layout, and
+ // `ShuffledJoin.outputPartitioning` would then report both as
alternative descriptions of
+ // it. Where a key row exists this clause is implied.
+ partitioning.keyDataTypes == otherPartitioning.keyDataTypes &&
Review Comment:
Non-blocking (pre-existing, untouched by this PR): the gate identifies a key
space by its erased type list, so two sides pruned to nothing whose spaces
differ but share a type still merge: `identity(id INT)` (raw values) vs
`bucket(4, id INT)` (bucket ids) gives `[IntegerType] == [IntegerType]` and
`Nil == Nil`, and the join reports one collection with an identity member and a
bucket member over one empty layout (same for the `fromPartitionings` require).
I confirmed with a probe - the twin of the new test with `IntegerType` on both
legs: a mixed collection forms, every member has `numPartitions == 0`, and the
answer stays correct.
It is inert: an empty layout holds no rows, so every claim over it is
vacuous, and any merge that later brings real rows under the layout rewrites
every member's expressions via `reducersBothWays`/`GroupPartitionsExec`. The
comment above the `fromPartitionings` require already concedes two empty lists
compare equal whatever they describe; consider adding the vacuity argument
there (or here), so a future reader knows the residual is a decision rather
than an oversight - and optionally a follow-up to require
same-function-per-position for empty-vs-empty pairs when no reducer will run.
--
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]
