From 0e0c238079de2cdb19fb2c757d216050a9fdc245 Mon Sep 17 00:00:00 2001
From: sablejade <315442805+sablejade@users.noreply.github.com>
Date: Tue, 11 Aug 2026 02:59:19 +0000
Subject: [PATCH] [spark] Support configurable dynamic partition column order
---
.../spark_connector_configuration.html | 6 +
.../paimon/spark/SparkConnectorOptions.java | 16 +
.../catalyst/analysis/PaimonAnalysis.scala | 74 ++++-
.../paimon/spark/util/OptionUtils.scala | 16 +-
.../sql/InsertOverwriteTableTestBase.scala | 32 ++
...aimonDynamicPartitionColumnOrderTest.scala | 303 ++++++++++++++++++
6 files changed, 440 insertions(+), 7 deletions(-)
create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonDynamicPartitionColumnOrderTest.scala
diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html
index cd95fd5fd41a..7bb20d791b03 100644
--- a/docs/generated/spark_connector_configuration.html
+++ b/docs/generated/spark_connector_configuration.html
@@ -86,6 +86,12 @@
| Boolean |
Whether to adjust the target split size based on pruned (projected) columns. If enabled, split size estimation uses only the columns actually being read. |
+
+ sql.dynamic-partition-column-order |
+ AUTO |
+ Enum |
+ Controls how non-BY-NAME Spark SQL dynamic partition writes interpret partition columns. TABLE uses table schema order, HIVE expects dynamic partition columns at the end when they are declared in a PARTITION clause or when a dynamic overwrite query matches that Hive-style output, and AUTO preserves compatible table and Hive order detection.
Possible values: |
+
vector-search.lateral-join.parallelism |
16 |
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
index 2f315b8df0f5..d43a80da72cd 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
@@ -82,6 +82,15 @@ public class SparkConnectorOptions {
.withDescription(
"If true, v2 write will be used. Currently, only HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1 write for other bucket modes. Currently, Spark V2 write does not support TableCapability.STREAMING_WRITE.");
+ public static final ConfigOption DYNAMIC_PARTITION_COLUMN_ORDER =
+ key("sql.dynamic-partition-column-order")
+ .enumType(DynamicPartitionColumnOrder.class)
+ .defaultValue(DynamicPartitionColumnOrder.AUTO)
+ .withDescription(
+ "Controls how non-BY-NAME Spark SQL dynamic partition writes interpret partition columns. "
+ + "TABLE uses table schema order, HIVE expects dynamic partition columns at the end when they are declared in a PARTITION clause or when a dynamic overwrite query matches that Hive-style output, "
+ + "and AUTO preserves compatible table and Hive order detection.");
+
public static final ConfigOption DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_MAX_ATTEMPTS =
key("write.data-evolution.update-conflict-retry.max-attempts")
.intType()
@@ -152,4 +161,11 @@ public class SparkConnectorOptions {
.withDescription(
"Whether to adjust the target split size based on pruned (projected) columns. "
+ "If enabled, split size estimation uses only the columns actually being read.");
+
+ /** Column order policy for non-BY-NAME dynamic partition writes. */
+ public enum DynamicPartitionColumnOrder {
+ AUTO,
+ TABLE,
+ HIVE
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
index d888401c25d0..97a21d0ef238 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
@@ -19,11 +19,13 @@
package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.options.Options
+import org.apache.paimon.spark.SparkConnectorOptions.DynamicPartitionColumnOrder
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable
import org.apache.paimon.spark.catalyst.plans.logical.{PaimonDropPartitions, PaimonHiveDynamicPartitionQuery}
import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper}
+import org.apache.paimon.spark.util.OptionUtils
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.{PaimonUtils, SparkSession}
@@ -111,19 +113,40 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
val query = stripHiveDynamicPartitionMarker(v2WriteCommand.query)
hiveDynamicPartitionColumns(v2WriteCommand.query) match {
case Some(dynamicPartitionColumns) if !v2WriteCommand.isByName =>
+ val configuredColumnOrder = OptionUtils.dynamicPartitionColumnOrder()
+ val columnOrder = configuredColumnOrder match {
+ case DynamicPartitionColumnOrder.AUTO
+ if dynamicPartitionColumnsUseTableOrder(query, table, dynamicPartitionColumns) =>
+ DynamicPartitionColumnOrder.TABLE
+ case order => order
+ }
resolveDynamicPartitionWrite(
query,
table,
- hiveStyleDynamicPartitionOutput(table, dynamicPartitionColumns),
+ columnOrder,
+ hiveStyleDynamicPartitionOutput(query, table, dynamicPartitionColumns),
options,
mergeSchemaEnabled)
case _ =>
v2WriteCommand match {
case o: OverwritePartitionsDynamic if !o.isByName =>
+ val hiveStyleCandidate = hiveStyleDynamicPartitionOutput(query, table)
+ val configuredColumnOrder =
+ if (hiveStyleCandidate.isDefined) {
+ OptionUtils.dynamicPartitionColumnOrder()
+ } else {
+ DynamicPartitionColumnOrder.AUTO
+ }
+ val hiveStyleOutput = configuredColumnOrder match {
+ case DynamicPartitionColumnOrder.TABLE => None
+ case DynamicPartitionColumnOrder.HIVE | DynamicPartitionColumnOrder.AUTO =>
+ hiveStyleCandidate
+ }
resolveDynamicPartitionWrite(
query,
table,
- hiveStyleDynamicPartitionOutput(query, table),
+ configuredColumnOrder,
+ hiveStyleOutput,
options,
mergeSchemaEnabled)
case _ =>
@@ -153,13 +176,17 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
private def resolveDynamicPartitionWrite(
query: LogicalPlan,
table: DataSourceV2Relation,
+ columnOrder: DynamicPartitionColumnOrder,
hiveStyleOutput: Option[Seq[Attribute]],
options: Options,
mergeSchemaEnabled: Boolean): LogicalPlan = {
hiveStyleOutput match {
case Some(hiveStyleOutput)
- if !sameOutputNames(query.output, table.output) &&
- !sameOutputNames(hiveStyleOutput, table.output) =>
+ if (columnOrder == DynamicPartitionColumnOrder.HIVE &&
+ !sameOutputNames(query.output, table.output)) ||
+ (columnOrder == DynamicPartitionColumnOrder.AUTO &&
+ !sameOutputNames(query.output, table.output) &&
+ !sameOutputNames(hiveStyleOutput, table.output)) =>
val hiveStyleQuery =
resolveWriteOutput(query, table.name, hiveStyleOutput, byName = false, mergeSchemaEnabled)
resolveWriteOutput(
@@ -214,12 +241,31 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
table: DataSourceV2Relation): Option[Seq[Attribute]] = {
val dynamicPartitionColumns =
table.table.asInstanceOf[SparkTable].getTable.partitionKeys().asScala.toSeq
- hiveStyleDynamicPartitionOutput(table, dynamicPartitionColumns).filter {
+ hiveStyleDynamicPartitionOutput(query, table, dynamicPartitionColumns).filter {
hiveStyleOutput => sameOutputNames(query.output, hiveStyleOutput)
}
}
+ private def dynamicPartitionColumnsUseTableOrder(
+ query: LogicalPlan,
+ table: DataSourceV2Relation,
+ dynamicPartitionColumns: Seq[String]): Boolean = {
+ if (query.output.size != table.output.size) {
+ false
+ } else {
+ val dynamicPartitionAttrs = table.output.zipWithIndex.filter {
+ case (attr, _) =>
+ dynamicPartitionColumns.exists(partition => conf.resolver(attr.name, partition))
+ }
+ dynamicPartitionAttrs.size == dynamicPartitionColumns.size &&
+ dynamicPartitionAttrs.forall {
+ case (attr, index) => conf.resolver(query.output(index).name, attr.name)
+ }
+ }
+ }
+
private def hiveStyleDynamicPartitionOutput(
+ query: LogicalPlan,
table: DataSourceV2Relation,
dynamicPartitionColumns: Seq[String]): Option[Seq[Attribute]] = {
val partitionKeys = table.table.asInstanceOf[SparkTable].getTable.partitionKeys().asScala.toSeq
@@ -238,7 +284,23 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
}
val hiveStyleOutput = dataAttrs ++ dynamicPartitionAttrs
if (dynamicPartitionAttrs.size == dynamicPartitionColumns.size) {
- Some(hiveStyleOutput)
+ val staticPartitionAttrsByIndex = table.output.zipWithIndex.collect {
+ case (attr, index)
+ if partitionKeys.exists(partition => conf.resolver(attr.name, partition)) &&
+ !dynamicPartitionColumns.exists(partition => conf.resolver(attr.name, partition)) &&
+ query.output
+ .lift(index)
+ .exists(queryAttr => conf.resolver(queryAttr.name, attr.name)) =>
+ index -> attr
+ }.toMap
+ val staticPartitionAttrs = staticPartitionAttrsByIndex.values.toSeq
+ val remainingAttrs = hiveStyleOutput.filterNot {
+ attr =>
+ staticPartitionAttrs.exists(staticAttr => conf.resolver(attr.name, staticAttr.name))
+ }.iterator
+ Some(table.output.indices.map {
+ index => staticPartitionAttrsByIndex.getOrElse(index, remainingAttrs.next())
+ })
} else {
None
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
index 37521ceda653..a0657418eb31 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.catalog.Identifier
import org.apache.paimon.options.ConfigOption
import org.apache.paimon.spark.{SparkCatalogOptions, SparkConnectorOptions}
+import org.apache.paimon.spark.SparkConnectorOptions.DynamicPartitionColumnOrder
import org.apache.paimon.table.Table
import org.apache.spark.internal.Logging
@@ -29,7 +30,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.internal.StaticSQLConf
-import java.util.{HashMap => JHashMap, Map => JMap}
+import java.util.{HashMap => JHashMap, Locale, Map => JMap}
import java.util.regex.Pattern
import scala.collection.JavaConverters._
@@ -106,6 +107,19 @@ object OptionUtils extends SQLConfHelper with Logging {
configuredValue && isVersionSupported
}
+ def dynamicPartitionColumnOrder(): DynamicPartitionColumnOrder = {
+ val configuredValue = getOptionString(SparkConnectorOptions.DYNAMIC_PARTITION_COLUMN_ORDER)
+ try {
+ DynamicPartitionColumnOrder.valueOf(configuredValue.trim.toUpperCase(Locale.ROOT))
+ } catch {
+ case _: IllegalArgumentException =>
+ throw new IllegalArgumentException(
+ s"Invalid value '$configuredValue' for " +
+ s"spark.paimon.${SparkConnectorOptions.DYNAMIC_PARTITION_COLUMN_ORDER.key()}. " +
+ "Supported values are AUTO, TABLE, and HIVE.")
+ }
+ }
+
def writeMergeSchemaEnabled(): Boolean = {
getOptionString(SparkConnectorOptions.MERGE_SCHEMA).toBoolean
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala
index ad6836011411..d8d034d8bfe5 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala
@@ -811,6 +811,38 @@ abstract class InsertOverwriteTableTestBase extends PaimonSparkTestBase {
}
}
+ test("Paimon Insert: table dynamic partition order survives UNION output aliases") {
+ if (gteqSpark3_4) {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "table"
+ ) {
+ withTable("dynamic_union") {
+ sql("""
+ |CREATE TABLE dynamic_union (
+ | ds STRING,
+ | part STRING,
+ | uid STRING,
+ | value STRING
+ |) PARTITIONED BY (ds, part)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT OVERWRITE dynamic_union PARTITION (ds, part)
+ |SELECT '2026-08-10' AS ds, 'p1' AS part, 'u1' AS uid, 'v1' AS detail_ratio
+ |UNION ALL
+ |SELECT '2026-08-10' AS ds, 'p2' AS part, 'u2' AS uid, 'v2' AS value
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT ds, part, uid, value FROM dynamic_union ORDER BY part"),
+ Seq(Row("2026-08-10", "p1", "u1", "v1"), Row("2026-08-10", "p2", "u2", "v2")))
+ }
+ }
+ }
+ }
+
test("Paimon Insert: dynamic insert into table with partition columns contain primary key") {
withSparkSQLConf("spark.sql.shuffle.partitions" -> "10") {
withTable("pk_pt") {
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonDynamicPartitionColumnOrderTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonDynamicPartitionColumnOrderTest.scala
new file mode 100644
index 000000000000..4de0620d24de
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonDynamicPartitionColumnOrderTest.scala
@@ -0,0 +1,303 @@
+/*
+ * 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.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.util.OptionUtils
+
+import org.apache.spark.sql.PaimonUtils.createDataset
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.QueryPlanningTracker
+import org.apache.spark.sql.catalyst.plans.logical.V2WriteCommand
+
+class PaimonDynamicPartitionColumnOrderTest extends PaimonSparkTestBase {
+
+ private val targetTableName = "dynamic_partition_order"
+
+ private def withDynamicPartitionTable(f: => Unit): Unit = {
+ withTable(targetTableName) {
+ sql(s"""
+ |CREATE TABLE $targetTableName (
+ | ds STRING,
+ | part STRING,
+ | uid STRING,
+ | value STRING
+ |) PARTITIONED BY (ds, part)
+ |""".stripMargin)
+ f
+ }
+ }
+
+ private def analyzedWriteQuery(insert: String) = {
+ val parsed = spark.sessionState.sqlParser.parsePlan(insert)
+ spark.sessionState.analyzer
+ .executeAndCheck(parsed, new QueryPlanningTracker)
+ .asInstanceOf[V2WriteCommand]
+ .query
+ }
+
+ private def tableOrderUnion: String =
+ s"""
+ |INSERT OVERWRITE $targetTableName PARTITION (ds, part)
+ |SELECT '2026-08-10' AS ds, 'p1' AS part, 'u1' AS uid, 'v1' AS detail_ratio
+ |UNION ALL
+ |SELECT '2026-08-10' AS ds, 'p2' AS part, 'u2' AS uid, 'v2' AS value
+ |""".stripMargin
+
+ test("table order preserves UNION output aliases") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "table"
+ ) {
+ withDynamicPartitionTable {
+ assert(
+ createDataset(spark, analyzedWriteQuery(tableOrderUnion)).collect().toSeq == Seq(
+ Row("2026-08-10", "p1", "u1", "v1"),
+ Row("2026-08-10", "p2", "u2", "v2")))
+ }
+ }
+ }
+
+ test("community defaults dynamic partition writes to auto order") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true") {
+ withDynamicPartitionTable {
+ assert(
+ createDataset(spark, analyzedWriteQuery(tableOrderUnion)).collect().toSeq == Seq(
+ Row("2026-08-10", "p1", "u1", "v1"),
+ Row("2026-08-10", "p2", "u2", "v2")))
+ }
+ }
+ }
+
+ test("auto detects table order despite UNION output aliases") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "auto"
+ ) {
+ withDynamicPartitionTable {
+ assert(
+ createDataset(spark, analyzedWriteQuery(tableOrderUnion)).collect().toSeq == Seq(
+ Row("2026-08-10", "p1", "u1", "v1"),
+ Row("2026-08-10", "p2", "u2", "v2")))
+ }
+ }
+ }
+
+ test("auto maps Hive order dynamic partition columns to table order") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "auto"
+ ) {
+ withDynamicPartitionTable {
+ val hiveOrderInsert =
+ s"""
+ |INSERT OVERWRITE $targetTableName PARTITION (ds, part)
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds, 'p1' AS part
+ |""".stripMargin
+
+ assert(
+ createDataset(spark, analyzedWriteQuery(hiveOrderInsert)).collect().toSeq == Seq(
+ Row("2026-08-10", "p1", "u1", "v1")))
+ }
+ }
+ }
+
+ test("hive order maps dynamic partition columns at the end to table order") {
+ Seq("true", "false").foreach {
+ useV2Write =>
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> useV2Write,
+ "spark.paimon.sql.dynamic-partition-column-order" -> "hive"
+ ) {
+ withDynamicPartitionTable {
+ sql(s"""
+ |INSERT OVERWRITE $targetTableName PARTITION (ds, part)
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds, 'p1' AS part
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT ds, part, uid, value FROM $targetTableName"),
+ Row("2026-08-10", "p1", "u1", "v1"))
+ }
+ }
+ }
+ }
+
+ test("hive order preserves input already matching table schema") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "hive"
+ ) {
+ withDynamicPartitionTable {
+ val tableOrderInsert =
+ s"""
+ |INSERT OVERWRITE $targetTableName PARTITION (ds, part)
+ |SELECT '2026-08-10' AS ds, 'p1' AS part, 'u1' AS uid, 'v1' AS value
+ |""".stripMargin
+
+ assert(
+ createDataset(spark, analyzedWriteQuery(tableOrderInsert)).collect().toSeq == Seq(
+ Row("2026-08-10", "p1", "u1", "v1")))
+ }
+ }
+ }
+
+ test("hive order keeps VALUES positional without partition clause") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "hive"
+ ) {
+ withDynamicPartitionTable {
+ sql(s"""
+ |INSERT OVERWRITE $targetTableName VALUES
+ | ('2026-08-10', 'p1', 'u1', 'v1')
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT ds, part, uid, value FROM $targetTableName"),
+ Row("2026-08-10", "p1", "u1", "v1"))
+ }
+ }
+ }
+
+ test("auto and hive orders detect named Hive-style output without partition clause") {
+ Seq("auto", "hive").foreach {
+ columnOrder =>
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> columnOrder
+ ) {
+ withDynamicPartitionTable {
+ sql(s"""
+ |INSERT OVERWRITE $targetTableName
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds, 'p1' AS part
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT ds, part, uid, value FROM $targetTableName"),
+ Row("2026-08-10", "p1", "u1", "v1"))
+ }
+ }
+ }
+ }
+
+ test("table order remains positional without partition clause") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "table"
+ ) {
+ withDynamicPartitionTable {
+ sql(s"""
+ |INSERT OVERWRITE $targetTableName
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds, 'p1' AS part
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT ds, part, uid, value FROM $targetTableName"),
+ Row("u1", "v1", "2026-08-10", "p1"))
+ }
+ }
+ }
+
+ test("auto and hive orders preserve mixed static and dynamic partitions") {
+ Seq("auto", "hive").foreach {
+ columnOrder =>
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> columnOrder
+ ) {
+ withTable("mixed_partition_order") {
+ sql("""
+ |CREATE TABLE mixed_partition_order (
+ | uid STRING,
+ | ds STRING,
+ | value STRING,
+ | region STRING
+ |) PARTITIONED BY (region, ds)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT OVERWRITE mixed_partition_order PARTITION (region = 'cn', ds)
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT uid, ds, value, region FROM mixed_partition_order"),
+ Row("u1", "2026-08-10", "v1", "cn"))
+ }
+ }
+ }
+ }
+
+ test("table order remains positional for Hive-style input") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "table"
+ ) {
+ withDynamicPartitionTable {
+ val hiveOrderInsert =
+ s"""
+ |INSERT OVERWRITE $targetTableName PARTITION (ds, part)
+ |SELECT 'u1' AS uid, 'v1' AS value, '2026-08-10' AS ds, 'p1' AS part
+ |""".stripMargin
+
+ assert(
+ createDataset(spark, analyzedWriteQuery(hiveOrderInsert)).collect().toSeq == Seq(
+ Row("u1", "v1", "2026-08-10", "p1")))
+ }
+ }
+ }
+
+ test("invalid dynamic partition column order fails clearly") {
+ withSparkSQLConf("spark.paimon.sql.dynamic-partition-column-order" -> "unknown") {
+ val error = intercept[IllegalArgumentException] {
+ OptionUtils.dynamicPartitionColumnOrder()
+ }
+ assert(error.getMessage.contains("Supported values are AUTO, TABLE, and HIVE"))
+ }
+ }
+
+ test("invalid dynamic partition column order does not affect non-partitioned writes") {
+ withSparkSQLConf(
+ "spark.sql.sources.partitionOverwriteMode" -> "dynamic",
+ "spark.paimon.write.use-v2-write" -> "true",
+ "spark.paimon.sql.dynamic-partition-column-order" -> "unknown"
+ ) {
+ withTable("non_partitioned_order") {
+ sql("CREATE TABLE non_partitioned_order (id INT, value STRING)")
+ sql("INSERT INTO non_partitioned_order VALUES (1, 'v1')")
+ checkAnswer(sql("SELECT id, value FROM non_partitioned_order"), Row(1, "v1"))
+ sql("INSERT OVERWRITE non_partitioned_order VALUES (2, 'v2')")
+ checkAnswer(sql("SELECT id, value FROM non_partitioned_order"), Row(2, "v2"))
+ }
+ }
+ }
+}