This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bd9274da9d [spark] Make saveAsTable+overwrite behave as INSERT
OVERWRITE (#8225)
bd9274da9d is described below
commit bd9274da9d52c6a8549bedf867fa462451dbd5f7
Author: Zouxxyy <[email protected]>
AuthorDate: Sun Jun 14 10:14:06 2026 +0800
[spark] Make saveAsTable+overwrite behave as INSERT OVERWRITE (#8225)
Previously, `df.write.mode("overwrite").saveAsTable("t")` produced a
`ReplaceTableAsSelect` plan, which could drop + recreate the table when
the user did not re-specify `partitionBy()` and primary-key options —
silently losing the partition spec, primary keys, and table properties.
This PR makes `saveAsTable` + `overwrite` on an existing table (Spark
3.4+) be rewritten to `OverwriteByExpression` (or
`OverwritePartitionsDynamic` when `partitionOverwriteMode=dynamic`),
preserving the existing table definition. This aligns with the behavior
of `INSERT OVERWRITE` and is consistent with Delta Lake.
SQL `CREATE OR REPLACE TABLE AS SELECT` and V2 `writeTo().replace()` are
not affected.
---
docs/docs/spark/dataframe.md | 27 ++--
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 50 ++++++-
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 52 ++++++-
.../paimon/spark/sql/DataFrameWriteTestBase.scala | 156 +++++++++++++++++++--
4 files changed, 255 insertions(+), 30 deletions(-)
diff --git a/docs/docs/spark/dataframe.md b/docs/docs/spark/dataframe.md
index 14cc60a028..d69b629497 100644
--- a/docs/docs/spark/dataframe.md
+++ b/docs/docs/spark/dataframe.md
@@ -56,7 +56,7 @@ Note: `insertInto` ignores the column names and just uses
position-based write,
if you need to write by column name, use `saveAsTable` or `save` instead.
### Insert Overwrite
-You can achieve INSERT OVERWRITE semantics by setting the mode to `overwrite`
with `insertInto`.
+You can achieve INSERT OVERWRITE semantics by setting the mode to `overwrite`.
It supports dynamic partition overwritten for partitioned table.
To enable dynamic overwritten you need to set the Spark session configuration
`spark.sql.sources.partitionOverwriteMode` to `dynamic`.
@@ -66,25 +66,18 @@ val data: DataFrame = ...
data.write.format("paimon")
.mode("overwrite")
- .insertInto("test_tbl")
+ .insertInto("test_tbl") // or .saveAsTable("test_tbl")
```
-## Replace Table
-You can achieve REPLACE TABLE semantics by setting the mode to `overwrite`
with `saveAsTable` or `save`.
+{{< hint info >}}
+Since Spark 3.4, `saveAsTable` with `overwrite` mode only overwrites data and
preserves
+the existing table definition (partitions, primary keys, and properties).
+If you need to replace the table definition, use SQL `CREATE OR REPLACE TABLE
... AS SELECT`.
-It first drops the existing table and then create a new one,
-so you need to specify the table's properties or partition columns if needed.
-
-```scala
-val data: DataFrame = ...
-
-data.write.format("paimon")
- .option("primary-key", "a,pt")
- .option("k1", "v1")
- .partitionBy("pt")
- .mode("overwrite")
- .saveAsTable("test_tbl") // or .save("/path/to/default.db/test_tbl")
-```
+Before Spark 3.4, `saveAsTable` with `overwrite` mode drops and recreates the
table, so the
+table definition is reset to the DataFrame's schema and only the partitions
and options
+explicitly re-specified via `partitionBy()` / write options are kept.
+{{< /hint >}}
## Query
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index b741627cdb..12ec5ee1d0 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
@@ -25,11 +25,13 @@ import org.apache.paimon.spark.catalog.SparkBaseCatalog
import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException,
ResolvedIdentifier}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, ReplaceTable,
ReplaceTableAsSelect, TableSpec}
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
OverwriteByExpression, OverwritePartitionsDynamic, ReplaceTable,
ReplaceTableAsSelect, TableSpec}
import org.apache.spark.sql.connector.catalog.{Identifier,
StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.{PaimonStrategyHelper, SparkPlan}
-import
org.apache.spark.sql.execution.datasources.v2.{AtomicReplaceTableAsSelectExec,
ReplaceTableAsSelectExec}
+import
org.apache.spark.sql.execution.datasources.v2.{AtomicReplaceTableAsSelectExec,
DataSourceV2Relation, ReplaceTableAsSelectExec}
+import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -51,6 +53,17 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
orCreate,
analyzedQuery) if
PaimonReplaceTableStrategyHelper.supportsCatalog(catalog, tableSpec) =>
assert(analyzedQuery.isDefined)
+ // For V1 saveAsTable + overwrite on an existing table, rewrite to
+ // OverwriteByExpression to preserve table definition.
+ if (PaimonReplaceTableStrategyHelper.isV1SaveAsTableOverwrite) {
+ val overwrite = PaimonReplaceTableStrategyHelper
+ .rewriteToOverwrite(spark, catalog, ident, analyzedQuery.get,
options)
+ if (overwrite.isDefined) {
+ val qe = spark.sessionState.executePlan(overwrite.get)
+ return qe.sparkPlan :: Nil
+ }
+ }
+
val (tableOptions, writeOptions) =
PaimonStrategyHelper.splitTableAndWriteOptions(options)
val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions)
val writeOpts = new CaseInsensitiveStringMap(writeOptions.asJava)
@@ -137,6 +150,39 @@ private[shim] object PaimonReplaceTableStrategyHelper {
case _ => false
}
+ /** @see PaimonReplaceTableStrategyHelper in paimon-spark-common for full
documentation. */
+ def isV1SaveAsTableOverwrite: Boolean = {
+ Thread.currentThread().getStackTrace.exists {
+ e =>
+ val cls = e.getClassName
+ cls.contains("DataFrameWriter") && !cls.contains("DataFrameWriterV2")
+ }
+ }
+
+ /** @see PaimonReplaceTableStrategyHelper in paimon-spark-common for full
documentation. */
+ def rewriteToOverwrite(
+ spark: SparkSession,
+ catalog: SparkBaseCatalog,
+ ident: Identifier,
+ query: LogicalPlan,
+ writeOptions: Map[String, String]): Option[LogicalPlan] = {
+ try {
+ val existing = catalog.loadTable(ident)
+ if (!existing.isInstanceOf[SparkTable]) return None
+ val relation =
+ DataSourceV2Relation.create(existing,
Some(catalog.asInstanceOf[TableCatalog]), Some(ident))
+ val dynamicOverwrite = existing.partitioning().nonEmpty &&
+ spark.sessionState.conf.partitionOverwriteMode ==
PartitionOverwriteMode.DYNAMIC
+ if (dynamicOverwrite) {
+ Some(OverwritePartitionsDynamic.byName(relation, query, writeOptions))
+ } else {
+ Some(OverwriteByExpression.byName(relation, query, Literal(true),
writeOptions))
+ }
+ } catch {
+ case _: NoSuchTableException => None
+ }
+ }
+
/**
* Whether replace can use Spark's staged replace path. Paimon's
replaceTable is not a
* rollbackable atomic replace; it swaps the current schema and truncates
current data while
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index 60410b631e..9e156c3e26 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
@@ -25,10 +25,13 @@ import org.apache.paimon.spark.catalog.SparkBaseCatalog
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException,
ResolvedIdentifier}
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, ReplaceTable,
ReplaceTableAsSelect, TableSpec}
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
OverwriteByExpression, OverwritePartitionsDynamic, ReplaceTable,
ReplaceTableAsSelect, TableSpec}
import org.apache.spark.sql.connector.catalog.{Identifier,
StagingTableCatalog, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.{PaimonStrategyHelper, SparkPlan,
SparkStrategy}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import scala.collection.JavaConverters._
@@ -46,6 +49,17 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
options,
orCreate,
true) if PaimonReplaceTableStrategyHelper.supportsCatalog(catalog,
tableSpec) =>
+ // For V1 saveAsTable + overwrite on an existing table, rewrite to
+ // OverwriteByExpression to preserve table definition.
+ if (PaimonReplaceTableStrategyHelper.isV1SaveAsTableOverwrite) {
+ val overwrite = PaimonReplaceTableStrategyHelper
+ .rewriteToOverwrite(spark, catalog, ident, query, options)
+ if (overwrite.isDefined) {
+ val qe = spark.sessionState.executePlan(overwrite.get)
+ return qe.sparkPlan :: Nil
+ }
+ }
+
val (tableOptions, writeOptions) =
PaimonStrategyHelper.splitTableAndWriteOptions(options)
val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions)
if (PaimonReplaceTableStrategyHelper.canAtomicReplace(catalog, ident,
qualifiedSpec, parts)) {
@@ -119,6 +133,42 @@ private[shim] object PaimonReplaceTableStrategyHelper {
case _ => false
}
+ /** Whether the current call originates from V1
DataFrameWriter.saveAsTable(). */
+ def isV1SaveAsTableOverwrite: Boolean = {
+ Thread.currentThread().getStackTrace.exists {
+ e =>
+ val cls = e.getClassName
+ cls.contains("DataFrameWriter") && !cls.contains("DataFrameWriterV2")
+ }
+ }
+
+ /**
+ * Rewrite to OverwriteByExpression or OverwritePartitionsDynamic for an
existing table,
+ * preserving table definition. Returns None if the table does not exist.
+ */
+ def rewriteToOverwrite(
+ spark: SparkSession,
+ catalog: SparkBaseCatalog,
+ ident: Identifier,
+ query: LogicalPlan,
+ writeOptions: Map[String, String]): Option[LogicalPlan] = {
+ try {
+ val existing = catalog.loadTable(ident)
+ if (!existing.isInstanceOf[SparkTable]) return None
+ val relation =
+ DataSourceV2Relation.create(existing,
Some(catalog.asInstanceOf[TableCatalog]), Some(ident))
+ val dynamicOverwrite = existing.partitioning().nonEmpty &&
+ spark.sessionState.conf.partitionOverwriteMode ==
PartitionOverwriteMode.DYNAMIC
+ if (dynamicOverwrite) {
+ Some(OverwritePartitionsDynamic.byName(relation, query, writeOptions))
+ } else {
+ Some(OverwriteByExpression.byName(relation, query, Literal(true),
writeOptions))
+ }
+ } catch {
+ case _: NoSuchTableException => None
+ }
+ }
+
/**
* Whether replace can use Spark's staged replace path. Paimon's
replaceTable is not a
* rollbackable atomic replace; it swaps the current schema and truncates
current data while
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
index de31d8ccc4..e4b91730ed 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
@@ -27,6 +27,8 @@ import org.junit.jupiter.api.Assertions
import java.sql.{Date, Timestamp}
+import scala.collection.JavaConverters._
+
abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
override protected def sparkConf: SparkConf = {
@@ -132,16 +134,33 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
Seq(Row("pt=p1"), Row("pt=p2"), Row("pt=p3"))
)
- // saveAsTable with overwrite mode will call replace table internal,
- // so here we set the props and partitions again.
- Seq((5, "x5", "p1"))
- .toDF("a", "b", "pt")
- .write
- .format("paimon")
- .option("primary-key", "a,pt")
- .partitionBy("pt")
- .mode("overwrite")
- .saveAsTable("t")
+ if (gteqSpark3_4) {
+ // On Spark 3.4+, saveAsTable with overwrite mode behaves as
INSERT OVERWRITE,
+ // preserving the table definition (partitions, primary keys,
properties).
+ // No need to re-specify partitionBy or primary-key.
+ Seq((5, "x5", "p1"))
+ .toDF("a", "b", "pt")
+ .write
+ .format("paimon")
+ .mode("overwrite")
+ .saveAsTable("t")
+
+ // Verify table definition is preserved
+ val table = loadTable("t")
+ Assertions.assertEquals(Seq("pt"),
table.partitionKeys().asScala.toSeq)
+ Assertions.assertEquals(Seq("a", "pt"),
table.primaryKeys().asScala.toSeq)
+ } else {
+ // On Spark 3.2/3.3, saveAsTable with overwrite still requires
re-specifying
+ // partitionBy and primary-key as the insert-overwrite
optimization is not available.
+ Seq((5, "x5", "p1"))
+ .toDF("a", "b", "pt")
+ .write
+ .format("paimon")
+ .option("primary-key", "a,pt")
+ .partitionBy("pt")
+ .mode("overwrite")
+ .saveAsTable("t")
+ }
checkAnswer(
spark.read.format("paimon").table("t").orderBy("a"),
Seq(Row(5, "x5", "p1"))
@@ -155,6 +174,123 @@ abstract class DataFrameWriteTestBase extends
PaimonSparkTestBase {
}
}
+ test("Paimon dataframe: saveAsTable overwrite preserves table definition and
snapshots") {
+ assume(gteqSpark3_4)
+ withTable("t") {
+ // Create a partitioned table with primary key and custom properties
+ spark.sql("""CREATE TABLE t (a INT, b STRING, pt STRING)
+ |USING paimon
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES ('primary-key' = 'a,pt', 'bucket' = '2')
+ |""".stripMargin)
+ spark.sql("INSERT INTO t VALUES (1, 'old1', 'p1'), (2, 'old2', 'p2')")
+
+ val oldLocation = loadTable("t").location().toString
+ val oldSnapshotId = loadTable("t").snapshotManager().latestSnapshotId()
+
+ // Overwrite without re-specifying partitionBy or primary-key
+ Seq((3, "new3", "p3"))
+ .toDF("a", "b", "pt")
+ .write
+ .format("paimon")
+ .mode("overwrite")
+ .saveAsTable("t")
+
+ // Verify table definition is fully preserved
+ val table = loadTable("t")
+ Assertions.assertEquals(Seq("pt"), table.partitionKeys().asScala.toSeq)
+ Assertions.assertEquals(Seq("a", "pt"),
table.primaryKeys().asScala.toSeq)
+ Assertions.assertEquals("2", table.options().get("bucket"))
+ Assertions.assertEquals(oldLocation, table.location().toString)
+
+ // Verify data is replaced
+ checkAnswer(sql("SELECT * FROM t ORDER BY a"), Row(3, "new3", "p3") ::
Nil)
+
+ // Verify old snapshots are still accessible via time travel
+ checkAnswer(
+ sql(s"SELECT * FROM t VERSION AS OF $oldSnapshotId ORDER BY a"),
+ Row(1, "old1", "p1") :: Row(2, "old2", "p2") :: Nil)
+ }
+ }
+
+ test("Paimon dataframe: saveAsTable overwrite on non-partitioned table") {
+ assume(gteqSpark3_4)
+ withTable("t") {
+ Seq((1, "x1"), (2, "x2"))
+ .toDF("a", "b")
+ .write
+ .format("paimon")
+ .mode("append")
+ .saveAsTable("t")
+
+ // Overwrite
+ Seq((3, "x3"))
+ .toDF("a", "b")
+ .write
+ .format("paimon")
+ .mode("overwrite")
+ .saveAsTable("t")
+
+ checkAnswer(sql("SELECT * FROM t ORDER BY a"), Row(3, "x3") :: Nil)
+ // Verify still non-partitioned
+ Assertions.assertTrue(loadTable("t").partitionKeys().isEmpty)
+ }
+ }
+
+ test("Paimon dataframe: saveAsTable overwrite creates table when not
exists") {
+ assume(gteqSpark3_4)
+ withTable("new_tbl") {
+ Seq((1, "x1"))
+ .toDF("a", "b")
+ .write
+ .format("paimon")
+ .option("primary-key", "a")
+ .mode("overwrite")
+ .saveAsTable("new_tbl")
+
+ checkAnswer(sql("SELECT * FROM new_tbl"), Row(1, "x1") :: Nil)
+ Assertions.assertEquals(Seq("a"),
loadTable("new_tbl").primaryKeys().asScala.toSeq)
+ }
+ }
+
+ test("Paimon dataframe: saveAsTable overwrite respects dynamic partition
overwrite mode") {
+ assume(gteqSpark3_4)
+ withTable("t") {
+ spark.sql("""CREATE TABLE t (a INT, b STRING, pt STRING)
+ |USING paimon
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES ('primary-key' = 'a,pt')
+ |""".stripMargin)
+ spark.sql("INSERT INTO t VALUES (1, 'x1', 'p1'), (2, 'x2', 'p2'), (3,
'x3', 'p1')")
+
+ // Dynamic partition overwrite: only p1 should be replaced, p2 untouched
+ withSparkSQLConf("spark.sql.sources.partitionOverwriteMode" ->
"dynamic") {
+ Seq((4, "x4", "p1"))
+ .toDF("a", "b", "pt")
+ .write
+ .format("paimon")
+ .mode("overwrite")
+ .saveAsTable("t")
+ }
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY a"),
+ Row(2, "x2", "p2") :: Row(4, "x4", "p1") :: Nil)
+
+ // Static overwrite (default): all partitions replaced
+ Seq((5, "x5", "p3"))
+ .toDF("a", "b", "pt")
+ .write
+ .format("paimon")
+ .mode("overwrite")
+ .saveAsTable("t")
+ checkAnswer(sql("SELECT * FROM t ORDER BY a"), Row(5, "x5", "p3") :: Nil)
+
+ // Verify table definition is preserved in both cases
+ Assertions.assertEquals(Seq("pt"),
loadTable("t").partitionKeys().asScala.toSeq)
+ Assertions.assertEquals(Seq("a", "pt"),
loadTable("t").primaryKeys().asScala.toSeq)
+ }
+ }
+
test("Paimon: DataFrameWrite.saveAsTable") {
withTable("test_ctas") {
Seq((1L, "x1"), (2L, "x2"))
