peter-toth commented on code in PR #58523:
URL: https://github.com/apache/spark/pull/58523#discussion_r3936024935
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -115,6 +115,9 @@ object InternalRowComparableWrapper {
new InternalRowComparableWrapper(partitionRow,
partitionExpression.map(_.dataType))
}
+ /** The cached ordering a wrapper of these `dataTypes` compares its rows
with in `equals`. */
+ def orderingFor(dataTypes: Seq[DataType]): BaseOrdering =
orderingCache.get(dataTypes)
Review Comment:
Good catch, and I had not traced that.
`PythonUserDefinedType.equals`/`hashCode` comparing only `pyUDT` collapses two
types the sort sites used to keep apart. The root cause is UDT equality rather
than the cache, and it already governs every wrapper-based comparison of such
keys, so I would rather not paper over it here. Worth a follow-up on
`UserDefinedType` itself.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -880,9 +880,13 @@ object KeyedPartitioning {
* `PartitioningCollection`, whose invariant requires equal partition keys
-- but join types that
* expose only one side's partitioning (e.g. LEFT OUTER) run nothing that
compares the two
* orders, and silently return wrong results.
+ *
+ * It is the keys' own ordering, the one
`InternalRowComparableWrapper.equals` compares with, so
+ * one definition answers both. `EnsureRequirements`' `OrderedDistribution`
arm is the one place
+ * that lays grouped keys out in another order, the distribution's own.
*/
def groupedKeyRowOrdering(dataTypes: Seq[DataType]): BaseOrdering =
- RowOrdering.createNaturalAscendingOrdering(dataTypes)
+ InternalRowComparableWrapper.orderingFor(dataTypes)
Review Comment:
Thank you for checking the suites. I am taking the caveat as written rather
than keying on the mode. The two orderings agree semantically, so what is at
stake is coverage of the interpreted path, not a result.
Your `FALLBACK` point is the sharper half: a transient codegen failure would
pin the interpreted instance for the JVM's life. If that ever bites,
`(codegenFactoryMode, dataTypes)` is the change and it stays cheap.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -92,8 +92,7 @@ trait DataSourceV2ScanExecBase
keyGroupedPartitioning match {
Review Comment:
Agreed, and we arrived at the same place from the other direction while
measuring this PR, so it is already filed as SPARK-59252 and in progress.
Measured by instrumenting the body and running
`KeyGroupedPartitioningSuite`: 33,051 executions as a `def` against 1,262 as a
`lazy val`. The same run shows 19,105 of those reads arriving through
`PartitioningPreservingUnaryExecNode.outputPartitioning`, itself a `def` doing
more work per call, so the ticket covers the three nodes together rather than
the scan alone.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -92,8 +92,7 @@ trait DataSourceV2ScanExecBase
keyGroupedPartitioning match {
case Some(exprs) if conf.v2BucketingEnabled &&
KeyedPartitioning.supportsExpressions(exprs) &&
inputPartitions.nonEmpty &&
inputPartitions.forall(_.isInstanceOf[HasPartitionKey]) =>
- val dataTypes = exprs.map(_.dataType)
- val rowOrdering = RowOrdering.createNaturalAscendingOrdering(dataTypes)
+ val rowOrdering =
KeyedPartitioning.groupedKeyRowOrdering(exprs.map(_.dataType))
val partitionKeys =
inputPartitions.map(_.asInstanceOf[HasPartitionKey].partitionKey()).sorted(rowOrdering)
KeyedPartitioning(exprs, partitionKeys)
Review Comment:
Right. The clean form needs `KeyedPartitioning.apply` to take the types
instead of recomputing them, which is what SPARK-59187 is about, so I would
rather not add an overload here and remove it there. Leaving the line as it is.
--
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]
