address review findings

Author: peter-toth

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

@@ -517,15 +517,18 @@ case class EnsureRequirements(
         val (rightReducedDataTypes, rightReducedKeys) = rightReducers.fold(
           (rightPartitioning.expressionDataTypes, rightPartitioning.partitionKeys)
         )(rightPartitioning.reduceKeys)
-        if (leftReducedDataTypes != rightReducedDataTypes) {
+        val commonReducedDataTypes = if (leftReducedDataTypes == rightReducedDataTypes) {
+          leftReducedDataTypes
+        } else {
           throw QueryExecutionErrors.storagePartitionJoinIncompatibleReducedTypesError(
             leftReducers = leftReducers,
             leftReducedDataTypes = leftReducedDataTypes,
             rightReducers = rightReducers,
             rightReducedDataTypes = rightReducedDataTypes)
         }
 
-        val reducedKeyRowOrdering = RowOrdering.createNaturalAscendingOrdering(leftReducedDataTypes)
+        val reducedKeyRowOrdering =
+          RowOrdering.createNaturalAscendingOrdering(commonReducedDataTypes)
         val reducedKeyOrdering =
           reducedKeyRowOrdering.on((t: InternalRowComparableWrapper) => t.row)
 

sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala

@@ -88,7 +88,7 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
       (id, oldFn)
     }
     try f finally {
-      oldFns.foreach { case (id, oldFn ) =>
+      oldFns.foreach { case (id, oldFn) =>
         catalog.dropFunction(id)
         oldFn.foreach(catalog.createFunction(id, _))
       }