From 5894c444a9ae25e685ef1f454effcdc30493d020 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Fri, 14 Aug 2026 18:28:44 +0800 Subject: [PATCH 1/3] [spark] Add option for Hive-style dynamic partition writes --- .../paimon/spark/SparkConnectorOptions.java | 10 +++++ .../catalyst/analysis/PaimonAnalysis.scala | 7 ++- .../paimon/spark/util/OptionUtils.scala | 4 ++ .../sql/InsertOverwriteTableTestBase.scala | 43 ++++++++++++++++++- 4 files changed, 60 insertions(+), 4 deletions(-) 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..108fe12dac79 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,16 @@ 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 HIVE_STYLE_DYNAMIC_PARTITION_ENABLED = + key("write.hive-style-dynamic-partition.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "If true, positional SQL inserts with explicit dynamic partitions " + + "use Hive's column order, with non-dynamic columns followed by " + + "dynamic partition columns. If false, the query output follows " + + "the table schema order."); + public static final ConfigOption DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_MAX_ATTEMPTS = key("write.data-evolution.update-conflict-retry.max-attempts") .intType() 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..1ecb417abe4d 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 @@ -24,6 +24,7 @@ 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} @@ -109,8 +110,10 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { options: Options, mergeSchemaEnabled: Boolean): LogicalPlan = { val query = stripHiveDynamicPartitionMarker(v2WriteCommand.query) + val hiveStyleDynamicPartitionEnabled = OptionUtils.hiveStyleDynamicPartitionEnabled() hiveDynamicPartitionColumns(v2WriteCommand.query) match { - case Some(dynamicPartitionColumns) if !v2WriteCommand.isByName => + case Some(dynamicPartitionColumns) + if hiveStyleDynamicPartitionEnabled && !v2WriteCommand.isByName => resolveDynamicPartitionWrite( query, table, @@ -119,7 +122,7 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { mergeSchemaEnabled) case _ => v2WriteCommand match { - case o: OverwritePartitionsDynamic if !o.isByName => + case o: OverwritePartitionsDynamic if hiveStyleDynamicPartitionEnabled && !o.isByName => resolveDynamicPartitionWrite( query, table, 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..10d403248bd3 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 @@ -110,6 +110,10 @@ object OptionUtils extends SQLConfHelper with Logging { getOptionString(SparkConnectorOptions.MERGE_SCHEMA).toBoolean } + def hiveStyleDynamicPartitionEnabled(): Boolean = { + getOptionString(SparkConnectorOptions.HIVE_STYLE_DYNAMIC_PARTITION_ENABLED).toBoolean + } + def writeMergeSchemaExplicitCastEnabled(): Boolean = { getOptionString(SparkConnectorOptions.EXPLICIT_CAST).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..40f5d1e120db 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 @@ -727,11 +727,50 @@ abstract class InsertOverwriteTableTestBase extends PaimonSparkTestBase { } } - test("Paimon Insert: V2 dynamic overwrite accepts Hive partition column order") { + test("Paimon Insert: [table-order-default] dynamic partition follows table order") { + for (useV2Write <- Seq("true", "false")) { + withSparkSQLConf( + "spark.sql.sources.partitionOverwriteMode" -> "dynamic", + "spark.paimon.write.use-v2-write" -> useV2Write) { + withTable("target_table") { + sql(""" + |CREATE TABLE target_table ( + | ds STRING, + | part STRING, + | uid STRING, + | value STRING + |) PARTITIONED BY (ds, part) + |TBLPROPERTIES ( + | 'primary-key' = 'ds,part,uid', + | 'bucket' = '2', + | 'bucket-key' = 'uid' + |) + |""".stripMargin) + + sql(""" + |INSERT OVERWRITE target_table PARTITION (ds, part) + |SELECT + | '20260808' AS ds, + | 'p1' AS part, + | '1001' AS uid, + | '0.8' AS metric + |""".stripMargin) + + checkAnswer( + sql("SELECT ds, part, uid, value FROM target_table"), + Row("20260808", "p1", "1001", "0.8")) + } + } + } + } + + test("Paimon Insert: [hive-tail-enabled] dynamic overwrite accepts Hive partition order") { if (gteqSpark3_4) { withSparkSQLConf( "spark.sql.sources.partitionOverwriteMode" -> "dynamic", - "spark.paimon.write.use-v2-write" -> "true") { + "spark.paimon.write.use-v2-write" -> "true", + "spark.paimon.write.hive-style-dynamic-partition.enabled" -> "true" + ) { withTable("my_table") { sql(""" |CREATE TABLE my_table ( From 3fc67919368d3a6fbbb31792cb02d30d6ec10a70 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Fri, 14 Aug 2026 19:01:22 +0800 Subject: [PATCH 2/3] [docs] Document Hive-style dynamic partition write option --- docs/docs/spark/sql-write.md | 27 +++++++++++++++++++ .../spark_connector_configuration.html | 6 +++++ 2 files changed, 33 insertions(+) diff --git a/docs/docs/spark/sql-write.md b/docs/docs/spark/sql-write.md index 1cc6c0bc691f..782bbee53e72 100644 --- a/docs/docs/spark/sql-write.md +++ b/docs/docs/spark/sql-write.md @@ -123,6 +123,33 @@ SELECT * FROM my_table; */ ``` +#### Dynamic Partition Column Order + +By default, positional SQL inserts with dynamic partition columns interpret the query output in the table schema order. For example, `dt` remains in its table position below: + +```sql +CREATE TABLE my_table ( + id INT, + dt STRING, + name STRING, + hr STRING +) PARTITIONED BY (dt, hr); + +INSERT OVERWRITE my_table PARTITION (dt, hr) +SELECT 1, '2026-08-14', 'Alice', '10'; +``` + +To migrate Hive-style SQL whose dynamic partition values are placed after all non-dynamic columns, enable the following Spark session configuration: + +```sql +SET spark.paimon.write.hive-style-dynamic-partition.enabled=true; + +INSERT OVERWRITE my_table PARTITION (dt, hr) +SELECT 1, 'Alice', '2026-08-14', '10'; +``` + +Enable this option only for positional writes that follow the Hive-style order. It does not affect `INSERT ... BY NAME`. + ## Truncate Table The `TRUNCATE TABLE` statement removes all the rows from a table or partition(s). diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html index cd95fd5fd41a..875b3d563994 100644 --- a/docs/generated/spark_connector_configuration.html +++ b/docs/generated/spark_connector_configuration.html @@ -104,6 +104,12 @@ Long Wait time in milliseconds between retry attempts for Spark V1 UPDATE on data-evolution tables after row-id range update conflicts. + +
write.hive-style-dynamic-partition.enabled
+ false + Boolean + If true, positional SQL inserts with explicit dynamic partitions use Hive's column order, with non-dynamic columns followed by dynamic partition columns. If false, the query output follows the table schema order. +
write.merge-schema
false From 41eb4031f1bf1cf8a068b702cf9c8de81376a840 Mon Sep 17 00:00:00 2001 From: Kerwin Zhang Date: Fri, 14 Aug 2026 21:48:45 +0800 Subject: [PATCH 3/3] [docs] Remove Hive-style dynamic partition SQL example --- docs/docs/spark/sql-write.md | 27 --------------------------- 1 file changed, 27 deletions(-) diff --git a/docs/docs/spark/sql-write.md b/docs/docs/spark/sql-write.md index 782bbee53e72..1cc6c0bc691f 100644 --- a/docs/docs/spark/sql-write.md +++ b/docs/docs/spark/sql-write.md @@ -123,33 +123,6 @@ SELECT * FROM my_table; */ ``` -#### Dynamic Partition Column Order - -By default, positional SQL inserts with dynamic partition columns interpret the query output in the table schema order. For example, `dt` remains in its table position below: - -```sql -CREATE TABLE my_table ( - id INT, - dt STRING, - name STRING, - hr STRING -) PARTITIONED BY (dt, hr); - -INSERT OVERWRITE my_table PARTITION (dt, hr) -SELECT 1, '2026-08-14', 'Alice', '10'; -``` - -To migrate Hive-style SQL whose dynamic partition values are placed after all non-dynamic columns, enable the following Spark session configuration: - -```sql -SET spark.paimon.write.hive-style-dynamic-partition.enabled=true; - -INSERT OVERWRITE my_table PARTITION (dt, hr) -SELECT 1, 'Alice', '2026-08-14', '10'; -``` - -Enable this option only for positional writes that follow the Hive-style order. It does not affect `INSERT ... BY NAME`. - ## Truncate Table The `TRUNCATE TABLE` statement removes all the rows from a table or partition(s).