peter-toth commented on code in PR #58501:
URL: https://github.com/apache/spark/pull/58501#discussion_r3947778110
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -105,21 +105,59 @@ object InternalRowComparableWrapper {
def apply(
partition: InputPartition with HasPartitionKey,
partitionExpression: Seq[Expression]): InternalRowComparableWrapper = {
- new InternalRowComparableWrapper(
- partition.asInstanceOf[HasPartitionKey].partitionKey(),
partitionExpression.map(_.dataType))
+ apply(partition.partitionKey(), partitionExpression)
}
def apply(
partitionRow: InternalRow,
partitionExpression: Seq[Expression]): InternalRowComparableWrapper = {
- new InternalRowComparableWrapper(partitionRow,
partitionExpression.map(_.dataType))
+
getInternalRowComparableWrapperFactory(partitionExpression.map(_.dataType))(partitionRow)
+ }
+
+ /**
+ * The types a row is compared at, which is the given types with their
naming erased: struct field
+ * names and every nullability go, and nothing else does.
+ *
+ * Two rows of the same value belong together whatever the columns they came
from were called. A
+ * storage-partitioned join relies on that: an equi-join across two structs
whose fields are named
+ * differently is legal, `identity` carries that name into the key type, and
+ * `KeyedShuffleSpec.createPartitioning` puts one side's expressions over
the other side's keys.
+ * So the naming is erased once, here, and every wrapper compares at these
types.
+ *
+ * Everything else is kept exactly, since it decides where a value belongs:
a collation and a
+ * decimal precision still tell two rows apart.
+ */
+ def comparableTypes(dataTypes: Seq[DataType]): Seq[DataType] = {
Review Comment:
Good point, and I took it. Fixed in
[`487da27`](https://github.com/apache/spark/commit/487da27608cc10e37177b72c2da2e7f84604dd39).
Not quite where you suggested, though. `hashCode` and `equals` do not call
`comparableTypes`, they read the `dataTypes` field, and that field is already
erased: the primary constructor is private and both entry points erase, so no
wrapper can exist over a raw list. Normalising inside `equals` would also put
the erasure on a per-comparison path, and the merge's `HashSet` and the
`distinct` in `KeyedPartitioning.apply` compare once per key.
What I did instead answers the same worry a level up. The factory is now a
`Factory` class that answers for the schema it settled on, so a caller
reporting a type list beside the rows it built reads `factory.dataTypes` rather
than erasing on its own:
- `KeyedPartitioning.projectKeys` and `reduceKeys` do that now, which also
drops a second erasure each was paying;
- `comparableTypes` is `private[catalyst]`, which is the tight scope, since
its only callers are `partitioning.scala` and its own suite.
One caller is left, `keyDataTypes`' fallback for a partitioning with no key
row. There is no factory there to read the types off, and building one just to
ask would be two cache lookups for no row. SPARK-59285 carries the types on the
partitioning, and that caller goes with 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]
