naveenp2708 commented on code in PR #59317:
URL: https://github.com/apache/spark/pull/59317#discussion_r4327758303
##########
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:
Added ALLOW_COMPATIBLE_TRANSFORMS back in 0bd6bd6, so partial clustering is
the only reason canReduceKeys is false. Ran your mutation (dropping the
partial-clustering term) and the test now fails on the overflow, so it’s
actually pinning partial clustering now. This re-adds the conf from your
round-1 finding 3, which makes sense now that the fix reads it.
##########
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:
Good catch @peter-toth . Gave key 0 a second split in 0bd6bd6 so it actually
spreads over two slots on t1 and repeats on t2 (numPartitions == Seq(5, 5)).
Before, the layout flag was set but nothing really spread since every key had
one split.
##########
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:
Updated the user-facing section to say the pair stays on raw keys whenever
canReduceKeys is false, not just under partial clustering. Agree raw keys is
the right call here.
##########
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:
Done in 0bd6bd6. Reworded the doc to describe the no-reduce path the fix
keeps.
--
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]
