This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5cdfefe4118a99b4fbf25cce2df437a7eba819db Author: fhan <[email protected]> AuthorDate: Wed Jun 3 18:44:39 2026 +0800 fix(spark): align CTAS partition fields by table partition order (#18899) * fix(spark): fix CTAS partition field order * fix(spark): add a short comment --------- Co-authored-by: fhan <[email protected]> (cherry picked from commit ba8c4c7b82c7c94405f712e5d1964226412f4e13) --- .../spark/sql/hudi/analysis/HoodieAnalysis.scala | 34 +++++++++++++++++- .../spark/sql/hudi/ddl/TestCreateTable.scala | 40 ++++++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala index d01627ae1fd5..46f9ed36ad78 100644 --- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala +++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala @@ -423,7 +423,8 @@ case class ResolveImplementationsEarly(spark: SparkSession) extends Rule[Logical // Convert to CreateHoodieTableAsSelectCommand case ct @ CreateTable(table, mode, Some(query)) if sparkAdapter.isHoodieTable(table) && ct.query.forall(_.resolved) => - val alignedQuery = stripMetaFieldAttributes(query) + val alignedQuery = alignCtasQueryByPartitionOrder( + stripMetaFieldAttributes(query), table.partitionColumnNames) CreateHoodieTableAsSelectCommand(table, mode, alignedQuery) case ct: CreateTable => @@ -445,6 +446,37 @@ case class ResolveImplementationsEarly(spark: SparkSession) extends Rule[Logical case _ => plan } } + + private def alignCtasQueryByPartitionOrder(query: LogicalPlan, partitionColumns: Seq[String]): LogicalPlan = { + if (partitionColumns.isEmpty) { + query + } else { + val resolver = spark.sessionState.conf.resolver + val (dataAttrs, partitionAttrs) = query.output.partition { attr => + !partitionColumns.exists(partition => resolver(partition, attr.name)) + } + + if (partitionAttrs.size != partitionColumns.size) { + throw new HoodieAnalysisException(s"Partition columns ${partitionColumns.mkString("[", ", ", "]")} " + + s"do not match query output ${query.output.map(_.name).mkString("[", ", ", "]")}") + } + + val alreadyAligned = partitionColumns.zip(partitionAttrs).forall { + case (partition, attr) => resolver(partition, attr.name) + } + // Avoid adding a redundant Project when partition columns are already in the table-defined order. + if (alreadyAligned) { + query + } else { + val orderedPartitionAttrs = partitionColumns.map { partition => + partitionAttrs.find(attr => resolver(partition, attr.name)).getOrElse { + throw new HoodieAnalysisException(s"Cannot resolve partition column $partition in CTAS query output") + } + } + Project(dataAttrs ++ orderedPartitionAttrs, query) + } + } + } } /** diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala index ead994342162..88ec4b7fb4c3 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala +++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/ddl/TestCreateTable.scala @@ -416,6 +416,46 @@ class TestCreateTable extends HoodieSparkSqlTestBase { Seq(1, "a1", 10, "2021-04-01") ) + // Create table with multi-level partition + val tableNameMultiLevelPartition = generateTableName + spark.sql( + s""" + | create table $tableNameMultiLevelPartition using hudi + | partitioned by (year, month, day) + | tblproperties( + | primaryKey = 'id', + | type = '$tableType' + | ) + | location '${tmp.getCanonicalPath}/$tableNameMultiLevelPartition' + | AS + | select 1 as id, 'a1' as name, 10 as price, '2021' as year, '04' as month, '01' as day + """.stripMargin + ) + + checkAnswer(s"select id, name, price, year, month, day from $tableNameMultiLevelPartition")( + Seq(1, "a1", 10, "2021", "04", "01") + ) + + // Create table with multi-level partition and out-of-order partition columns + val tableNameMultiLevelPartitionDisorder = generateTableName + spark.sql( + s""" + | create table $tableNameMultiLevelPartitionDisorder using hudi + | partitioned by (year, month, day) + | tblproperties( + | primaryKey = 'id', + | type = '$tableType' + | ) + | location '${tmp.getCanonicalPath}/$tableNameMultiLevelPartitionDisorder' + | AS + | select 1 as id, 'a1' as name, 10 as price, '04' as month, '01' as day, '2021' as year + """.stripMargin + ) + + checkAnswer(s"select id, name, price, year, month, day from $tableNameMultiLevelPartitionDisorder")( + Seq(1, "a1", 10, "2021", "04", "01") + ) + // Create Partitioned table with timestamp data type val tableName3 = generateTableName // CTAS failed with null primaryKey
