peter-toth commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4311577088
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2387,6 +2387,12 @@ case class KeyedShuffleSpec(
*/
def reducersBothWays(other: KeyedShuffleSpec)
: (Option[Seq[Option[KeyReducer]]], Option[Seq[Option[KeyReducer]]]) = {
+ // `canReduceKeys` is the same condition `areKeysCompatible` consults:
when it is false, for
+ // example under partial clustering, only same-function pairs are
admitted, and those already
+ // share a key space, so neither side needs a reducer. Deriving one anyway
lets a function that
+ // reduces itself merge keys that were never meant to merge, so the spread
side's expected
+ // counts no longer match the groups it builds (SPARK-59234).
+ if (!canReduceKeys) return (None, None)
Review Comment:
**Finding 9.** This also changes plans with partial clustering off.
`V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS` defaults to `false`, so
`canReduceKeys` is false on the default config. I joined two `self_reduce`
tables with keys `{0, 1, 2, 3}` and `{0, 1, 2}` on the default config, which
takes the push branch:
- master: both sides reduce, 2 partitions
- this PR: no reducers, 3 partitions
The same happens with `pushPartValues` off and
`allowKeysSubsetOfPartitionKeys` on. Rows are correct either way. Keeping the
raw keys looks right to me. This follows from the shape I suggested in finding
6, where I only looked at partial clustering.
Could the description's user-facing section say that a same-function pair no
longer reduces whenever `canReduceKeys` is false, not only under partial
clustering?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1328,6 +1328,43 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59234: a self-reducing function is not reduced under partial
clustering") {
+ // self_reduce(id) is a transform that reduces itself. Under partial
clustering
+ // `canReduceKeys` is false, so a same-function pair is admitted without a
reduce and
+ // `reducersBothWays` returns none (SPARK-59234): the join stays on the
raw keys, with no
+ // merging to miscount. Assert that: correct rows, no shuffle, the
partially-clustered layout
+ // (one side spreads with `SPLIT_FOR_JOIN`, the other repeats with
`REPLICATED_FOR_JOIN`), and
+ // that neither side reduced. If a change brought the reducer back, the
merged keys would no
+ // longer match the spread side's groups and the fail-fast guard would
trip.
+ withFunction(SelfReduceFunction) {
+ val cols = Array(Column.create("id", LongType), Column.create("data",
StringType))
+ createTable("t1", cols, Array(Expressions.apply("self_reduce",
Expressions.column("id"))))
+ sql("INSERT INTO testcat.ns.t1 VALUES (0, 'a'), (1, 'b'), (2, 'c'), (3,
'd')")
Review Comment:
**Finding 8.** Each key has one split here. `createTable` puts one row in
each split, and the ids are distinct. So every expected count is 1, and both
sides come out grouped (`isGrouped` true, no origin on the layout). The
`ungroupingOrigin` assertion still passes. It reads the nodes' constructor
field, which `EnsureRequirements` sets whenever partial clustering applies. So
nothing spreads or repeats, although the comment and the description say it
does.
A second split for key 0 makes it real. Key 0 then spreads over two slots on
`t1` and repeats twice on `t2`:
```scala
sql("INSERT INTO testcat.ns.t1 VALUES (0, 'a'), (0, 'e'), (1, 'b'),
(2, 'c'), (3, 'd')")
...
checkAnswer(df, Seq(Row(0L), Row(0L), Row(1L), Row(2L), Row(3L)))
...
assert(gpes.map(_.outputPartitioning.numPartitions) == Seq(5, 5),
s"key 0 gets two slots on both sides:\n$plan")
```
I ran it together with finding 7's conf. It passes at this head. It fails on
master, because both sides shuffle. It also fails under finding 7's mutation,
on the overflow assert.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1328,6 +1328,43 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59234: a self-reducing function is not reduced under partial
clustering") {
+ // self_reduce(id) is a transform that reduces itself. Under partial
clustering
+ // `canReduceKeys` is false, so a same-function pair is admitted without a
reduce and
+ // `reducersBothWays` returns none (SPARK-59234): the join stays on the
raw keys, with no
+ // merging to miscount. Assert that: correct rows, no shuffle, the
partially-clustered layout
+ // (one side spreads with `SPLIT_FOR_JOIN`, the other repeats with
`REPLICATED_FOR_JOIN`), and
+ // that neither side reduced. If a change brought the reducer back, the
merged keys would no
+ // longer match the spread side's groups and the fail-fast guard would
trip.
+ withFunction(SelfReduceFunction) {
+ val cols = Array(Column.create("id", LongType), Column.create("data",
StringType))
+ createTable("t1", cols, Array(Expressions.apply("self_reduce",
Expressions.column("id"))))
+ sql("INSERT INTO testcat.ns.t1 VALUES (0, 'a'), (1, 'b'), (2, 'c'), (3,
'd')")
+ createTable("t2", cols, Array(Expressions.apply("self_reduce",
Expressions.column("id"))))
+ sql("INSERT INTO testcat.ns.t2 VALUES (0, 'w'), (1, 'x'), (2, 'y'), (3,
'z')")
+ val df = sql(
+ "SELECT t1.id FROM testcat.ns.t1 JOIN testcat.ns.t2 ON t1.id = t2.id")
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
"true") {
Review Comment:
**Finding 7.** `canReduceKeys` is false here for two reasons. Partial
clustering is on, and `V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS` is off by
default. So the test doesn't pin that partial clustering alone stops the
reduce. I removed `!conf.v2BucketingPartiallyClusteredDistributionEnabled` from
`canReduceKeys` (`partitioning.scala:2366`):
- this test still passes
- with the conf below turned on, it fails, because planning trips the
overflow assert
That mutation is what a change bringing compatible transforms to partial
clustering would do. With the conf on, the test passes at this head. On master
it still fails because both sides shuffle.
This reverses my finding 3 from round 1. Back then nothing in the fix read
this conf. Now the fix reads it through `canReduceKeys`.
```suggestion
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
-> "true") {
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/functions/SelfReduceFunction.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog.functions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.types.{DataType, IntegerType, LongType}
+
+/**
+ * A partition transform on a long key that reduces itself: `reducer` returns
a merging reducer
+ * (x -> x % 2) when the other side is this same function. The both-sides
reduce contract
+ * r1(f(x)) = r2(f(x)) is vacuous when the two sides share the function, so
any merging reducer is
+ * admitted. It exercises the partial-clustering path where the spread side's
keys are reduced,
+ * which no in-tree reducible function reaches (SPARK-59234).
Review Comment:
**Finding 10.** After the fix, no spread side is reduced under partial
clustering, so this describes the path the fix removed.
```suggestion
* admitted. No in-tree reducible function reduces itself. This one lets a
test check that a
* same-function pair is not reduced under partial clustering (SPARK-59234).
```
--
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]
