dongjoon-hyun commented on code in PR #58345:
URL: https://github.com/apache/spark/pull/58345#discussion_r3873777903
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala:
##########
@@ -407,10 +406,15 @@ object ShuffleExchangeExec {
assert(k.isGrouped,
s"Expected a grouped KeyedPartitioning on ${k.expressions}, but got
${k.numPartitions} " +
"partition keys with duplicates among them")
+ // Project the partition keys to UnsafeRows so that map lookups
compare them by value
+ // (e.g. binary keys by content, NaNs as equal), consistently with
both the
+ // InternalRowComparableWrapper semantics used to group
`partitionKeys` and the lookup
+ // keys produced by `getPartitionKeyExtractor` below.
+ val projection = UnsafeProjection.create(k.expressionDataTypes.toArray)
Review Comment:
Thank you for the thorough analysis and the measured reproduction,
@peter-toth. You are right that `UnsafeRow` byte equality is strictly finer
than the `RowOrdering` equivalence that grouped `partitionKeys`, and the `-0.0`
vs `0.0` case was a regression against the base `Seq[Any]` comparison.
Adopted your suggestion in 5e89b10: the map keys are now the partitioning's
own `InternalRowComparableWrapper`s and `getPartitionKeyExtractor` wraps the
evaluated key via `getInternalRowComparableWrapperFactory`, so the lookup and
the grouping share one equivalence by construction.
`InternalRowComparableWrapper` is now `Serializable` with
`structType`/`ordering` `@transient` and re-derived from the shared caches on
first use after deserialization.
Also added the regression test you asked for: a `signed_zeros` connector
function with `inputTypes() = Array(LongType)` and `resultType() = DoubleType`
mapping 1 to `-0.0` and 2 to `0.0`, plus the `InMemoryBaseTable` whitelist
entry and `getKey` case. I verified it fails against the previous `UnsafeRow`
revision (one match lost) and passes with this one. The suite is 106/106.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1691,6 +1691,43 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-59054: shuffle one side: partition keys with binary type") {
+ val items_partitions = Array(identity("id"))
+ createTable(items, Array(
+ Column.create("id", BinaryType),
+ Column.create("name", StringType),
+ Column.create("price", DoubleType)), items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(X'0101', 'aa', 40.0), " +
+ "(X'0202', 'bb', 10.0), " +
+ "(X'0303', 'cc', 15.5), " +
+ "(X'0404', 'dd', 20.0)")
+
+ createTable(purchases, Array(
+ Column.create("item_id", BinaryType),
+ Column.create("price", DoubleType)), Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(X'0101', 42.0), (X'0101', 44.0), (X'0202', 11.0), (X'0202', 19.5), " +
+ "(X'0303', 26.0), (X'0303', 30.0), (X'0404', 50.0), (X'0404', 60.0)")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
Review Comment:
Done in 5e89b10 — both new tests now loop `Seq(true, false)` and assert 2
shuffles with identical results in the conf-off arm.
--
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]
