diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala index 2354cf69205a0..2f07bc213e7b5 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala +++ b/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) /** * Projects a sequence of partition keys by selecting only the specified positions. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala index 3566319c39ad9..d33e32453f140 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala +++ b/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) + /** Creates a shared factory method for a given row schema to avoid excessive cache lookups. */ def getInternalRowComparableWrapperFactory( dataTypes: Seq[DataType]): InternalRow => InternalRowComparableWrapper = { diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala new file mode 100644 index 0000000000000..b720174239c09 --- /dev/null +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala @@ -0,0 +1,39 @@ +/* + * 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.catalyst.util + +import org.apache.spark.SparkFunSuite +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning +import org.apache.spark.sql.types.{IntegerType, LongType} + +class InternalRowComparableWrapperSuite extends SparkFunSuite { + + test("SPARK-59249: the grouped key layout and wrapper equality hold one ordering instance") { + // Identity is the property to assert, because behaviour is not what changes here: both sides + // were already built by the same function, so they already compared the same way. What one + // instance buys is that neither side can later be given a definition the other does not have. + // `InternalRowComparableWrapper.equals` compares its rows with the instance below, and + // `KeyedPartitioning` sorts and groups partition keys with it. The two type lists are built + // separately, so this pins the shared cache as well. + val wrapper = InternalRowComparableWrapper + .getInternalRowComparableWrapperFactory(Seq(IntegerType, LongType))(InternalRow(1, 2L)) + + assert(KeyedPartitioning.groupedKeyRowOrdering(Seq(IntegerType, LongType)) eq wrapper.ordering) + } +} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala index f00d8b9b82cb4..56b9219e3039c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2 import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression, RowOrdering, SortOrder} +import org.apache.spark.sql.catalyst.expressions.{Ascending, Expression, SortOrder} import org.apache.spark.sql.catalyst.plans.physical import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning import org.apache.spark.sql.catalyst.util.truncatedString @@ -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)