peter-toth commented on code in PR #58273:
URL: https://github.com/apache/spark/pull/58273#discussion_r3855208278
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -4207,6 +4207,40 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-56877: v2 bucketed table with subset join keys joining v1
table") {
+ // The v2 table is partitioned by an extra identity key `dt` plus
`bucket(16, c1)`, while the
+ // join is only on `c1`. allowKeysSubsetOfPartitionKeys lets the operation
key `c1` be a subset
+ // of the partition keys `[dt, bucket(16, c1)]`, so EnsureRequirements
projects the keyed side
+ // to `[bucket(16, c1)]`. v2BucketingShuffleEnabled then re-shuffles only
the v1 side using that
+ // projected KeyedPartitioning. ShuffledJoin wraps the two output
partitionings into a
+ // PartitioningCollection, which requires all KeyedPartitionings to share
equal partitionKeys.
+ // The v2 side's keys are sorted by GroupPartitionsExec, while the keys
re-used for the v1 side
+ // keep their first-occurrence order from createShuffleSpec, so the two
sequences disagree and
+ // the collection construction used to fail.
+ val cols = Array(
+ Column.create("c1", LongType),
+ Column.create("c2", StringType),
+ Column.create("dt", StringType))
+ val partitions = Array(identity("dt"), bucket(16, "c1"))
+
+ createTable("iceberg_t2", cols, partitions)
+ sql("INSERT INTO testcat.ns.iceberg_t2 VALUES (2, 'cc', '2020'), (1, 'aa',
'2021')")
+
+ withTable("t1") {
+ sql("CREATE TABLE t1 (c1 BIGINT, c2 STRING) USING parquet")
+ sql("INSERT INTO t1 VALUES (1, 'aa'), (2, 'cc')")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ val df = sql("SELECT * FROM testcat.ns.iceberg_t2 t0 JOIN t1 ON t0.c1
= t1.c1")
+ checkAnswer(df, Seq(
Review Comment:
**Finding 2.** This is a regression test for a *planning* failure, so it
should pin the plan shape. `checkAnswer` alone stays green if planning later
stops taking the storage-partitioned-join path, and at that point the test
silently no longer reaches `createShuffleSpec` at all. That is not hypothetical
here - the path needs `dt` to survive column pruning into the scan, which is
the exact fragility the two `SPARK-46367` tests above call out in their own
comments.
I measured the plan with AQE off: one shuffle (the v1 side) and one
`GroupPartitionsExec` (the v2 side). The suite's helpers already express that:
```scala
withSQLConf(
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val df = sql("SELECT * FROM testcat.ns.iceberg_t2 t0 JOIN t1 ON
t0.c1 = t1.c1")
val plan = df.queryExecution.executedPlan
// Only the v1 side is re-shuffled; the v2 side is regrouped onto
the join key instead.
assert(collectShuffles(plan).length == 1)
assert(collectGroupPartitions(plan).length == 1)
checkAnswer(df, Seq(
Row(1L, "aa", "2021", 1L, "aa"),
Row(2L, "cc", "2020", 2L, "cc")))
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -613,9 +613,12 @@ case class KeyedPartitioning(
val joinKeyPositions =
result.keyPositions.map(_.nonEmpty).zipWithIndex.filter(_._1).map(_._2)
val projectedExpressions = joinKeyPositions.map(expressions)
val projectedKeys = projectKeys(joinKeyPositions)._2
- val distinctProjectedKeys = projectedKeys.distinct
+ // Sort the distinct projected keys the same way `GroupPartitionsExec`
does. Otherwise, when
+ // only the keyed side is grouped and the other side is re-shuffled
using this spec, the two
+ // `KeyedPartitioning`s carry the same keys in a different order and
+ // `PartitioningCollection.fromPartitionings` rejects them.
val projectedPartitioning =
- KeyedPartitioning(projectedExpressions, distinctProjectedKeys,
isGrouped = true)
+ new KeyedPartitioning(projectedExpressions, projectedKeys, isGrouped =
false).toGrouped
Review Comment:
**Finding 1.** The projected branch is consistent now, but the `else` branch
below still hands the child's `KeyedPartitioning` back verbatim, and
`KeyedShuffleSpec.createPartitioning` (`partitioning.scala:1393`) then reuses
its keys as-is:
```scala
KeyedPartitioning(newExpressions, partitioning.partitionKeys,
partitioning.isGrouped)
```
The physical partitioner does not: `ShuffleExchangeExec.getPartitioner`
builds `KeyGroupedPartitioner` from `k.toGrouped`, i.e. always from the
*sorted* keys (`ShuffleExchangeExec.scala:402`). So whenever the keyed side's
own keys are not sorted, the two sides disagree exactly as they did here -
except nothing throws, because `createPartitioning` passes the same
`partitionKeys` *reference* through, so
`PartitioningCollection.fromPartitionings` interns it on `eq` and the `require`
never runs.
Unsorted keys are reachable without `allowKeysSubsetOfPartitionKeys`: a
narrowing `PartitioningPreservingUnaryExecNode` projects sorted `[dt, id]` keys
down to `id` in first-occurrence order
(`AliasAwareOutputExpression.scala:138`), and `UnionExec` concatenates its
children's keys (`basicPhysicalOperators.scala:989`). The class doc at
`partitioning.scala:449-452` states this outright.
Measured on this branch. Same table, same query, same configs - only the
inserted row values differ, which flips the narrowed key sequence:
| narrowed keys | `v2BucketingShuffleEnabled` | shuffles | rows |
|---|---|---|---|
| `[2, 1]` | `false` | 2 | `[1,2021,x], [2,2020,y]` |
| `[2, 1]` | `true` | 1 | *(empty)* |
| `[1, 2]` | `false` | 2 | `[1,2020,x], [2,2021,y]` |
| `[1, 2]` | `true` | 1 | `[1,2020,x], [2,2021,y]` |
```sql
-- testcat.ns.nt partitioned by [identity(dt), identity(id)], one row per
split,
-- rows (1,'2021'),(2,'2020') for the unsorted case and
(1,'2020'),(2,'2021') for the sorted one
SELECT a.id, a.m, t1.data
FROM (SELECT id, MAX(dt) AS m FROM testcat.ns.nt GROUP BY id) a
JOIN t1 ON a.id = t1.id -- t1 is a plain parquet table
```
Both `true` rows have no `GroupPartitionsExec`, so nothing re-sorts the
keyed side. This behaves identically on `master`, so it is pre-existing and not
something this PR introduced - a separate ticket is fine. I raise it here
because it is the same root cause, and because both halves of a fix live next
to this diff.
The sort cannot move into the `else` branch: there `partitioning` is the
child's own declared layout, and sorting it would misreport the child, whereas
the projected branch is honest only because `EnsureRequirements` inserts the
`GroupPartitionsExec` that realises it. So the minimum is `.toGrouped` in
`createPartitioning`, which makes the reported layout match what the
partitioner builds - that alone re-arms the SPARK-56877 `require` and turns the
silent wrong answer into a loud failure. To keep the query working you then
need either `canCreatePartitioning` to refuse a spec whose `partitionKeys` are
not sorted (falls back to hash shuffles on both sides, loses the
storage-partitioned join), or a `GroupPartitionsExec` on the keyed side with
`expectedPartitionKeys` set to the sorted keys, the way the
`OrderedDistribution` branch already does at `EnsureRequirements.scala:99-107`.
Do you want to take this as a follow-up, or shall I file the ticket and put
up the fix? Either works for me.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -4207,6 +4207,40 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-56877: v2 bucketed table with subset join keys joining v1
table") {
Review Comment:
**Finding 3.** `SPARK-56877` is the ticket that added the invariant; this
change is `SPARK-58988`, so anyone grepping for the new ticket's coverage will
not find it.
```suggestion
test("SPARK-58988: v2 bucketed table with subset join keys joining v1
table") {
```
The `SPARK-56877` reference is worth keeping in the comment body, which
already has it.
--
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]
