From 029f1bb5a3706b3a282ee748fe3b6944c271f000 Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Tue, 14 Jul 2026 17:54:42 +0800 Subject: [PATCH 1/3] SPARK-58120: doCanonicalize preserves keyGroupedPartitioning expression order --- .../datasources/v2/BatchScanExec.scala | 3 +- .../KeyGroupedPartitioningSuite.scala | 30 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala index e7a7dc8145632..2e3394ac2a082 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala @@ -106,7 +106,8 @@ case class BatchScanExec( runtimeFilters = QueryPlan.normalizePredicates( runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)), output), - keyGroupedPartitioning = keyGroupedPartitioning.map(QueryPlan.normalizePredicates(_, output))) + keyGroupedPartitioning = keyGroupedPartitioning.map(_.map( + QueryPlan.normalizeExpressions(_, output)))) } override def simpleString(maxFields: Int): String = { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala index e765b86301892..389a931e6fc76 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala @@ -4469,4 +4469,34 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with } } } + + test("SPARK-58120: doCanonicalize preserves keyGroupedPartitioning expression order") { + // Regression test: BatchScanExec.doCanonicalize previously used + // QueryPlan.normalizePredicates which combines expressions with And, canonicalizes, + // then splits back. This reorders expressions, causing a mismatch between + // expression data types and partition key row values, leading to ClassCastException. + val partition = Array(identity("id"), identity("data")) + createTable(table, columns, partition) + sql(s"INSERT INTO testcat.ns.$table VALUES " + + "(1, 'aa', cast('2020-01-01' as timestamp)), " + + "(2, 'bb', cast('2021-01-01' as timestamp)), " + + "(3, 'cc', cast('2022-01-01' as timestamp))") + + val df = sql(s"SELECT id, data FROM testcat.ns.$table") + val scan = df.queryExecution.executedPlan.collect { case b: BatchScanExec => b }.head + val canonicalized = scan.canonicalized.asInstanceOf[BatchScanExec] + + // The canonicalized keyGroupedPartitioning expressions must preserve the same order + // as the original: [id (IntegerType), data (StringType)], not reversed. + val originalTypes = scan.keyGroupedPartitioning.get.map(_.dataType) + val canonicalizedTypes = canonicalized.keyGroupedPartitioning.get.map(_.dataType) + assert(originalTypes == canonicalizedTypes, + s"Expression order changed after canonicalization: " + + s"original types=$originalTypes, canonicalized types=$canonicalizedTypes") + + // Also verify that the canonicalized expressions reference the correct attributes + // (first should be IntegerType, second should be StringType) + assert(canonicalizedTypes == Seq(IntegerType, StringType), + s"Expected [IntegerType, StringType] but got $canonicalizedTypes") + } } From 787e51925d7a6ee9f4f4f91446572a8692e7d572 Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Wed, 15 Jul 2026 09:24:15 +0800 Subject: [PATCH 2/3] fix --- .../sql/catalyst/plans/QueryPlanSuite.scala | 19 +++++++++++++++++-- .../KeyGroupedPartitioningSuite.scala | 12 +++++------- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/QueryPlanSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/QueryPlanSuite.scala index 91f990be7bb23..68c0e354950f3 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/QueryPlanSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/QueryPlanSuite.scala @@ -24,11 +24,11 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.dsl.plans._ -import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Expression, ListQuery, Literal, NamedExpression, Rand} +import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSeq, Expression, ExprId, ListQuery, Literal, NamedExpression, Rand} import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation, LogicalPlan, Project, Union} import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, TreePattern} -import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.types.{IntegerType, StringType} class QueryPlanSuite extends SparkFunSuite { @@ -185,4 +185,19 @@ class QueryPlanSuite extends SparkFunSuite { assert(visited.size == 2) assert(visited.forall(_.containsPattern(TreePattern.FILTER))) } + + test("SPARK-58120: normalizePredicates reorders expressions by hashCode via orderCommutative") { + val id = AttributeReference("id", IntegerType)(ExprId(0)) + val data = AttributeReference("data", StringType)(ExprId(1)) + val output: AttributeSeq = Seq(id, data) + val exprs: Seq[Expression] = Seq(id, data) + + val normalizedPredicates = QueryPlan.normalizePredicates(exprs, output) + assert(normalizedPredicates.map(_.dataType) == Seq(StringType, IntegerType), + "normalizePredicates reorders expressions by hashCode") + + val normalizedExpressions = exprs.map(QueryPlan.normalizeExpressions(_, output)) + assert(normalizedExpressions.map(_.dataType) == Seq(IntegerType, StringType), + "normalizeExpressions preserves original expression order") + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala index 389a931e6fc76..5961e8c7ca43f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala @@ -4471,10 +4471,9 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with } test("SPARK-58120: doCanonicalize preserves keyGroupedPartitioning expression order") { - // Regression test: BatchScanExec.doCanonicalize previously used - // QueryPlan.normalizePredicates which combines expressions with And, canonicalizes, - // then splits back. This reorders expressions, causing a mismatch between - // expression data types and partition key row values, leading to ClassCastException. + // The int/string pair is known to reorder under normalizePredicates hashCode sorting. + // See QueryPlanSuite "SPARK-58120: normalizePredicates reorders expressions by hashCode + // via orderCommutative". val partition = Array(identity("id"), identity("data")) createTable(table, columns, partition) sql(s"INSERT INTO testcat.ns.$table VALUES " + @@ -4488,14 +4487,13 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with // The canonicalized keyGroupedPartitioning expressions must preserve the same order // as the original: [id (IntegerType), data (StringType)], not reversed. + assert(scan.keyGroupedPartitioning.isDefined, + "Expected BatchScanExec to have keyGroupedPartitioning set") val originalTypes = scan.keyGroupedPartitioning.get.map(_.dataType) val canonicalizedTypes = canonicalized.keyGroupedPartitioning.get.map(_.dataType) assert(originalTypes == canonicalizedTypes, s"Expression order changed after canonicalization: " + s"original types=$originalTypes, canonicalized types=$canonicalizedTypes") - - // Also verify that the canonicalized expressions reference the correct attributes - // (first should be IntegerType, second should be StringType) assert(canonicalizedTypes == Seq(IntegerType, StringType), s"Expected [IntegerType, StringType] but got $canonicalizedTypes") } From 063d194d56e91bcd5c5d2669d3f2d8db76673ee3 Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Wed, 15 Jul 2026 09:48:59 +0800 Subject: [PATCH 3/3] fix --- .../KeyGroupedPartitioningSuite.scala | 28 ------------------- .../spark/sql/execution/SparkPlanSuite.scala | 26 ++++++++++++++++- 2 files changed, 25 insertions(+), 29 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala index 5961e8c7ca43f..e765b86301892 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala @@ -4469,32 +4469,4 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with } } } - - test("SPARK-58120: doCanonicalize preserves keyGroupedPartitioning expression order") { - // The int/string pair is known to reorder under normalizePredicates hashCode sorting. - // See QueryPlanSuite "SPARK-58120: normalizePredicates reorders expressions by hashCode - // via orderCommutative". - val partition = Array(identity("id"), identity("data")) - createTable(table, columns, partition) - sql(s"INSERT INTO testcat.ns.$table VALUES " + - "(1, 'aa', cast('2020-01-01' as timestamp)), " + - "(2, 'bb', cast('2021-01-01' as timestamp)), " + - "(3, 'cc', cast('2022-01-01' as timestamp))") - - val df = sql(s"SELECT id, data FROM testcat.ns.$table") - val scan = df.queryExecution.executedPlan.collect { case b: BatchScanExec => b }.head - val canonicalized = scan.canonicalized.asInstanceOf[BatchScanExec] - - // The canonicalized keyGroupedPartitioning expressions must preserve the same order - // as the original: [id (IntegerType), data (StringType)], not reversed. - assert(scan.keyGroupedPartitioning.isDefined, - "Expected BatchScanExec to have keyGroupedPartitioning set") - val originalTypes = scan.keyGroupedPartitioning.get.map(_.dataType) - val canonicalizedTypes = canonicalized.keyGroupedPartitioning.get.map(_.dataType) - assert(originalTypes == canonicalizedTypes, - s"Expression order changed after canonicalization: " + - s"original types=$originalTypes, canonicalized types=$canonicalizedTypes") - assert(canonicalizedTypes == Seq(IntegerType, StringType), - s"Expected [IntegerType, StringType] but got $canonicalizedTypes") - } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala index b167dd13dcbd6..c0f5ad61d72e3 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala @@ -34,7 +34,7 @@ import org.apache.spark.sql.catalyst.trees.LeafLike import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession -import org.apache.spark.sql.types.{DataType, IntegerType} +import org.apache.spark.sql.types.{DataType, IntegerType, StringType} import org.apache.spark.sql.vectorized.ColumnarBatch import org.apache.spark.util.ThreadUtils @@ -232,6 +232,30 @@ class SparkPlanSuite extends SharedSparkSession { executor.shutdown() } } + + test("SPARK-58120: BatchScanExec doCanonicalize preserves keyGroupedPartitioning order") { + // The int/string pair is known to reorder under normalizePredicates hashCode sorting. + // See QueryPlanSuite "SPARK-58120: normalizePredicates reorders expressions by hashCode + // via orderCommutative". + val id = AttributeReference("id", IntegerType)(ExprId(0)) + val data = AttributeReference("data", StringType)(ExprId(1)) + val scan = BatchScanExec( + output = Seq(id, data), + scan = null, + runtimeFilters = Seq.empty, + table = null, + keyGroupedPartitioning = Some(Seq(id, data))) + val canonicalized = scan.canonicalized.asInstanceOf[BatchScanExec] + assert(scan.keyGroupedPartitioning.isDefined, + "Expected BatchScanExec to have keyGroupedPartitioning set") + val originalTypes = scan.keyGroupedPartitioning.get.map(_.dataType) + val canonicalizedTypes = canonicalized.keyGroupedPartitioning.get.map(_.dataType) + assert(originalTypes == canonicalizedTypes, + s"Expression order changed after canonicalization: " + + s"original types=$originalTypes, canonicalized types=$canonicalizedTypes") + assert(canonicalizedTypes == Seq(IntegerType, StringType), + s"Expected [IntegerType, StringType] but got $canonicalizedTypes") + } } case class ColumnarOp(child: SparkPlan) extends UnaryExecNode {