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 5dacde516b [core][spark] Fix format table INSERT OVERWRITE ignoring
dynamic-partition-overwrite (#9353)
5dacde516b is described below
commit 5dacde516bf6e29b583aff8b32bab321bfc6e64f
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Sat Aug 22 17:32:37 2026 +0800
[core][spark] Fix format table INSERT OVERWRITE ignoring
dynamic-partition-overwrite (#9353)
---
docs/docs/spark/sql-write.md | 8 ++
.../table/format/FormatBatchWriteBuilder.java | 3 +-
.../paimon/table/format/FormatTableCommit.java | 113 +++++++++++++-----
.../format/FormatTableCommitStatisticsTest.java | 87 +++++++++++++-
.../paimon/table/format/FormatTableCommitTest.java | 131 +++++++++++++++++++--
.../spark/format/FormatTableBatchWrite.scala | 3 +-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 3 +-
.../spark/format/FormatTableBatchWriteBase.scala | 11 +-
.../paimon/spark/format/PaimonFormatTable.scala | 30 +++--
.../apache/spark/sql/paimon/shims/SparkShim.scala | 1 -
.../paimon/spark/sql/FormatTableTestBase.scala | 48 ++++++++
.../paimon/spark/table/PaimonFormatTableTest.scala | 12 +-
.../spark/format/FormatTableBatchWrite.scala | 3 +-
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 3 +-
.../spark/format/FormatTableBatchWrite.scala | 3 +-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 3 +-
16 files changed, 394 insertions(+), 68 deletions(-)
diff --git a/docs/docs/spark/sql-write.md b/docs/docs/spark/sql-write.md
index 5ac34f0834..8c0f8f8531 100644
--- a/docs/docs/spark/sql-write.md
+++ b/docs/docs/spark/sql-write.md
@@ -123,6 +123,14 @@ SELECT * FROM my_table;
*/
```
+A Format Table read through Paimon (`format-table.implementation = paimon`,
the default) follows
+the same rule. An `INSERT OVERWRITE` that names no partition replaces the
whole table, so a
+partition the query does not write is replaced too, and a query that returns
no rows leaves the
+table empty; `dynamic` mode replaces only the partitions written, and writing
nothing then replaces
+nothing. With `metastore.partitioned-table = true` the catalog is the answer
to which partitions
+the table has, so overwriting the whole table empties those and leaves a
directory still waiting
+for `MSCK REPAIR TABLE` alone.
+
## Truncate Table
The `TRUNCATE TABLE` statement removes all the rows from a table or
partition(s).
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index b14ab00890..73c11d7741 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -89,7 +89,8 @@ public class FormatBatchWriteBuilder implements
BatchWriteBuilder {
staticPartition,
syncHiveUri,
table.catalogContext(),
- table.partitionManager());
+ table.partitionManager(),
+ options.dynamicPartitionOverwrite());
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index aec5f73e15..be2105930a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -74,6 +74,7 @@ public class FormatTableCommit implements BatchTableCommit {
private Catalog hiveCatalog;
private Identifier tableIdentifier;
@Nullable private final FormatTablePartitionManager partitionManager;
+ private final boolean dynamicPartitionOverwrite;
public FormatTableCommit(
String location,
@@ -86,7 +87,8 @@ public class FormatTableCommit implements BatchTableCommit {
@Nullable Map<String, String> staticPartitions,
@Nullable String syncHiveUri,
CatalogContext catalogContext,
- @Nullable FormatTablePartitionManager partitionManager) {
+ @Nullable FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite) {
this.location = location;
this.fileIO = fileIO;
this.formatTablePartitionOnlyValueInPath =
formatTablePartitionOnlyValueInPath;
@@ -97,6 +99,7 @@ public class FormatTableCommit implements BatchTableCommit {
this.partitionKeys = partitionKeys;
this.tableIdentifier = tableIdentifier;
this.partitionManager = partitionManager;
+ this.dynamicPartitionOverwrite = dynamicPartitionOverwrite;
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -155,15 +158,25 @@ public class FormatTableCommit implements
BatchTableCommit {
fileIO.mkdirs(partitionPath);
}
} else if (overwrite) {
- Set<Path> partitionPaths = new HashSet<>();
- for (TwoPhaseCommitMessage message : messages) {
-
partitionPaths.add(message.getCommitter().targetPath().getParent());
- }
- for (Path p : partitionPaths) {
- // The parent of a written file is a complete partition
directory - the table
- // directory itself when the table is unpartitioned - so
there is no partition
- // level below it to descend, and it is a partition this
commit writes anyway.
- deletePreviousDataFile(p, 0);
+ if (replacesOnlyWrittenPartitions()) {
+ Set<Path> partitionPaths = new HashSet<>();
+ for (TwoPhaseCommitMessage message : messages) {
+
partitionPaths.add(message.getCommitter().targetPath().getParent());
+ }
+ for (Path p : partitionPaths) {
+ // The parent of a written file is a complete
partition directory - the
+ // table directory itself when the table is
unpartitioned - so there is no
+ // partition level below it to descend, and it is a
partition this commit
+ // writes anyway.
+ deletePreviousDataFile(p, 0);
+ }
+ } else {
+ // Overwriting without naming a partition replaces the
table, so what has to go
+ // is everything the table holds rather than the files
this commit happens to
+ // write: a statement whose query returns nothing still
empties the table.
+ for (Path dataDirectory : tableDataDirectories()) {
+
clearedPartitionPaths.addAll(deletePreviousDataFile(dataDirectory, 0));
+ }
}
}
@@ -238,11 +251,12 @@ public class FormatTableCommit implements
BatchTableCommit {
}
/**
- * Registers the partitions this commit touched, carrying the statistics
of what it wrote. A
- * static prefix overwrite also empties partitions it writes nothing to;
those report an exact
- * zero and are registered with the rest, since a statistic can only be
reported for a partition
- * its own request registers. A truncation writes nothing and reports
every partition it
- * emptied.
+ * Registers the partitions this commit touched, carrying the statistics
of what it wrote. An
+ * overwrite also empties partitions it writes nothing to - those below a
static prefix, and
+ * every partition the table has when the statement names none and dynamic
partition overwrite
+ * is off; those report an exact zero and are registered with the rest,
since a statistic can
+ * only be reported for a partition its own request registers. A
truncation writes nothing and
+ * reports every partition it emptied.
*/
private void reportPartitions(
Set<Map<String, String>> writtenPartitionSpecs,
@@ -433,8 +447,62 @@ public class FormatTableCommit implements BatchTableCommit
{
public void close() throws Exception {}
/**
- * Deletes the data files below a path and returns the directories they
sat in, which for a
- * static prefix overwrite can be partitions this commit never writes.
+ * Whether an overwrite that names no partition replaces only the
partitions this commit wrote
+ * rather than everything the table holds. Same condition a data table
commit applies: the
+ * option is about which partitions to replace, so an unpartitioned table
has nothing for it to
+ * select.
+ */
+ private boolean replacesOnlyWrittenPartitions() {
+ return partitionKeys != null && !partitionKeys.isEmpty() &&
dynamicPartitionOverwrite;
+ }
+
+ /**
+ * The directories this table's data sits in: the table directory itself
when the table is
+ * unpartitioned, and one per partition otherwise, taken from wherever the
table reads its
+ * partitions. A directory no scan of the table reads holds none of its
data - one the catalog
+ * has not registered, or one whose name does not parse into the partition
keys - and replacing
+ * what the table holds leaves it alone, the way {@link #truncateTable()}
does.
+ */
+ private List<Path> tableDataDirectories() {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.singletonList(new Path(location));
+ }
+ List<Path> directories = new ArrayList<>();
+ if (partitionManager != null) {
+ for (Map<String, String> spec :
registeredPartitions(Collections.emptyMap())) {
+ directories.add(
+ buildPartitionPath(
+ location,
+ spec,
+ formatTablePartitionOnlyValueInPath,
+ partitionKeys));
+ }
+ return directories;
+ }
+ for (Pair<LinkedHashMap<String, String>, Path> partition :
partitionsInTheFileSystem()) {
+ directories.add(partition.getRight());
+ }
+ return directories;
+ }
+
+ /** The partition directories a table that discovers its partitions from
the files has. */
+ private List<Pair<LinkedHashMap<String, String>, Path>>
partitionsInTheFileSystem() {
+ return PartitionPathUtils.searchPartSpecAndPaths(
+ fileIO,
+ new Path(location),
+ partitionKeys.size(),
+ partitionKeys,
+ formatTablePartitionOnlyValueInPath,
+ null,
+ null,
+ defaultPartName);
+ }
+
+ /**
+ * Deletes the data files below a path and returns the directories they
sat in. Those can be
+ * partitions this commit never writes: a static prefix overwrite clears
the partitions sitting
+ * below the prefix, and an overwrite that names no partition clears every
partition the table
+ * has.
*/
private Set<Path> deletePreviousDataFile(Path partitionPath, int
partitionLevels)
throws IOException {
@@ -498,16 +566,7 @@ public class FormatTableCommit implements BatchTableCommit
{
// Filesystem partition discovery: the partition directories the scan
reads are the table.
// A directory that does not parse into the partition keys is not one
of them, so
// truncating leaves it alone.
- for (Pair<LinkedHashMap<String, String>, Path> partition :
- PartitionPathUtils.searchPartSpecAndPaths(
- fileIO,
- new Path(location),
- partitionKeys.size(),
- partitionKeys,
- formatTablePartitionOnlyValueInPath,
- null,
- null,
- defaultPartName)) {
+ for (Pair<LinkedHashMap<String, String>, Path> partition :
partitionsInTheFileSystem()) {
try {
deletePreviousDataFile(partition.getRight(), 0);
} catch (IOException e) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
index 199b7b446b..f5ac146070 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
@@ -223,6 +223,66 @@ class FormatTableCommitStatisticsTest {
});
}
+ @Test
+ void testOverwritingTheWholeTableZeroesAPartitionItDidNotRewrite() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "10"), spec("2025", "11"));
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv",
2048);
+ // The statement names no partition, and its query wrote only one of
the two.
+ CommitMessage message = writtenFile(fileIO, tablePath,
"year=2025/month=10", 3, 128);
+
+ overwritingTheWholeTable(tablePath, fileIO, partitionManager)
+ .commit(Collections.singletonList(message));
+
+ Reported reported = capture(partitionManager);
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.specs)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ assertThat(reported.statistics)
+ .anySatisfy(
+ statistics -> {
+
assertThat(statistics.spec()).isEqualTo(spec("2025", "10"));
+ assertThat(statistics.recordCount()).isEqualTo(3);
+
assertThat(statistics.fileSizeInBytes()).isEqualTo(128);
+ assertThat(statistics.fileCount()).isEqualTo(1);
+ })
+ // Emptied and not written to: the whole table was replaced,
so an exact zero
+ // rather than the numbers of the files this commit deleted.
+ .anySatisfy(
+ statistics -> {
+
assertThat(statistics.spec()).isEqualTo(spec("2025", "11"));
+ assertThat(statistics.recordCount()).isZero();
+ assertThat(statistics.fileSizeInBytes()).isZero();
+ assertThat(statistics.fileCount()).isZero();
+ });
+ }
+
+ @Test
+ void
testOverwritingTheWholeTableLeavesADirectoryTheCatalogHasNotRegistered() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "10"));
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv",
4096);
+ // A directory MSCK REPAIR TABLE has not registered yet: no scan of
this table reads it,
+ // so replacing what the table holds is not this directory's business
either.
+ writeDataFile(fileIO, tablePath, "year=2025/month=11",
"unregistered.csv", 2048);
+
+ overwritingTheWholeTable(tablePath, fileIO, partitionManager)
+ .commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=10/data.csv"))).isFalse();
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=11/unregistered.csv")))
+ .isTrue();
+ // Nor does the overwrite register it: reporting a zero for it would
make a partition the
+ // catalog never had, out of a directory that still holds rows.
+ Reported reported = capture(partitionManager);
+ assertThat(reported.specs).containsExactly(spec("2025", "10"));
+ }
+
@Test
void testTruncatingPartitionsReportsAnExactZeroAsTheTotal() throws
Exception {
LocalFileIO fileIO = LocalFileIO.create();
@@ -584,6 +644,24 @@ class FormatTableCommitStatisticsTest {
boolean overwrite,
Map<String, String> staticPartitions,
boolean onlyValueInPath) {
+ return commit(
+ tablePath,
+ fileIO,
+ partitionManager,
+ overwrite,
+ staticPartitions,
+ onlyValueInPath,
+ /* dynamicPartitionOverwrite */ true);
+ }
+
+ private FormatTableCommit commit(
+ Path tablePath,
+ FileIO fileIO,
+ FormatTablePartitionManager partitionManager,
+ boolean overwrite,
+ Map<String, String> staticPartitions,
+ boolean onlyValueInPath,
+ boolean dynamicPartitionOverwrite) {
return new FormatTableCommit(
tablePath.toString(),
PARTITION_KEYS,
@@ -595,7 +673,14 @@ class FormatTableCommitStatisticsTest {
staticPartitions,
null,
null,
- partitionManager);
+ partitionManager,
+ dynamicPartitionOverwrite);
+ }
+
+ /** An overwrite that names no partition: INSERT OVERWRITE without a
PARTITION clause. */
+ private FormatTableCommit overwritingTheWholeTable(
+ Path tablePath, FileIO fileIO, FormatTablePartitionManager
partitionManager) {
+ return commit(tablePath, fileIO, partitionManager, true, null, false,
false);
}
/** A file this commit wrote, with the counts its writer took. */
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
index db12129814..75b141d947 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -85,7 +85,8 @@ class FormatTableCommitTest {
null,
null,
null,
- partitionManager);
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
CommitMessage message = new TwoPhaseCommitMessage(committer);
assertThatThrownBy(() ->
commit.commit(Collections.singletonList(message)))
@@ -117,7 +118,8 @@ class FormatTableCommitTest {
null,
null,
null,
- partitionManager);
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
CommitMessage message = new TwoPhaseCommitMessage(committer);
assertThatThrownBy(() ->
commit.commit(Collections.singletonList(message)))
@@ -212,7 +214,8 @@ class FormatTableCommitTest {
staticPartition,
null,
null,
- null);
+ null,
+ /* dynamicPartitionOverwrite */ true);
commit.commit(Collections.singletonList(new
TwoPhaseCommitMessage(committer)));
@@ -260,7 +263,8 @@ class FormatTableCommitTest {
Collections.singletonMap("year", "2025"),
null,
null,
- null);
+ null,
+ /* dynamicPartitionOverwrite */ true);
commit.commit(Collections.emptyList());
@@ -303,7 +307,8 @@ class FormatTableCommitTest {
Collections.singletonMap("year", "2025"),
null,
null,
- null);
+ null,
+ /* dynamicPartitionOverwrite */ true);
commit.commit(Collections.emptyList());
@@ -350,7 +355,8 @@ class FormatTableCommitTest {
staticPartition,
null,
null,
- null);
+ null,
+ /* dynamicPartitionOverwrite */ true);
assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
.isInstanceOf(RuntimeException.class)
@@ -542,6 +548,113 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(defaultPartition)).isTrue();
}
+ @Test
+ void testOverwritingWithoutAStaticPartitionEmptiesAnUnpartitionedTable()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path previousData = new Path(tablePath, "data-1.csv");
+ fileIO.writeFile(previousData, "1", false);
+ // Another writer is mid-write; its staging tree is not table data.
+ Path stagingFile = new Path(tablePath,
"_temporary/attempt/part-00000.csv");
+ fileIO.writeFile(stagingFile, "2", false);
+ FormatTableCommit commit = overwritingCommit(tablePath, fileIO, true);
+
+ // Nothing to write: the query behind the statement returned no rows.
+ commit.commit(Collections.emptyList());
+
+ // An unpartitioned overwrite is about the table, so the files it
replaces are the table's,
+ // not the ones this commit happens to write.
+ assertThat(fileIO.exists(previousData)).isFalse();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ }
+
+ @Test
+ void testOverwritingAPartitionedTableFollowsDynamicPartitionOverwrite()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path dynamicTable = new Path(new Path(tempDir.toUri()), "dynamic");
+ Path staticTable = new Path(new Path(tempDir.toUri()), "static");
+ for (Path table : Arrays.asList(dynamicTable, staticTable)) {
+ fileIO.writeFile(new Path(table, "year=2025/month=10/data-1.csv"),
"1", false);
+ fileIO.writeFile(new Path(table, "year=2025/month=11/data-2.csv"),
"2", false);
+ }
+
+ overwritingCommit(dynamicTable, fileIO, true, "year", "month")
+ .commit(Collections.emptyList());
+ overwritingCommit(staticTable, fileIO, false, "year", "month")
+ .commit(Collections.emptyList());
+
+ // Dynamic overwrite selects the partitions written, and this commit
wrote none.
+ assertThat(fileIO.exists(new Path(dynamicTable,
"year=2025/month=10/data-1.csv"))).isTrue();
+ assertThat(fileIO.exists(new Path(dynamicTable,
"year=2025/month=11/data-2.csv"))).isTrue();
+ // With it off, the statement is about the whole table, whatever the
query returned.
+ assertThat(fileIO.exists(new Path(staticTable,
"year=2025/month=10/data-1.csv"))).isFalse();
+ assertThat(fileIO.exists(new Path(staticTable,
"year=2025/month=11/data-2.csv"))).isFalse();
+ }
+
+ @Test
+ void testDynamicOverwriteReplacesOnlyThePartitionsItWrites() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path rewritten = new Path(tablePath, "year=2025/month=10/data-1.csv");
+ Path untouched = new Path(tablePath, "year=2025/month=11/data-2.csv");
+ fileIO.writeFile(rewritten, "1", false);
+ fileIO.writeFile(untouched, "2", false);
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(
+ fileIO, new Path(tablePath,
"year=2025/month=10/data-new.csv"), false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer =
outputStream.closeForCommit();
+
+ overwritingCommit(tablePath, fileIO, true, "year", "month")
+ .commit(Collections.singletonList(new
TwoPhaseCommitMessage(committer)));
+
+ // The partitions a dynamic overwrite replaces are the ones it writes,
and only those.
+ assertThat(fileIO.exists(rewritten)).isFalse();
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=10/data-new.csv"))).isTrue();
+ assertThat(fileIO.exists(untouched)).isTrue();
+ }
+
+ @Test
+ void testOverwritingTheWholeTableLeavesADirectoryThatIsNoPartitionOfIt()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ fileIO.writeFile(new Path(tablePath, "year=2025/month=10/data-1.csv"),
"1", false);
+ // Neither of these parses into the partition keys, so no scan of the
table reads them:
+ // replacing what the table holds is none of their business.
+ fileIO.writeFile(new Path(tablePath, "year=2025/nomonth/data-2.csv"),
"2", false);
+ fileIO.writeFile(new Path(tablePath, "loose.csv"), "3", false);
+
+ overwritingCommit(tablePath, fileIO, false, "year", "month")
+ .commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=10/data-1.csv"))).isFalse();
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/nomonth/data-2.csv"))).isTrue();
+ assertThat(fileIO.exists(new Path(tablePath, "loose.csv"))).isTrue();
+ }
+
+ /**
+ * An overwrite that names no partition: what INSERT OVERWRITE without a
PARTITION clause is.
+ */
+ private FormatTableCommit overwritingCommit(
+ Path tableLocation,
+ LocalFileIO fileIO,
+ boolean dynamicPartitionOverwrite,
+ String... partitionKeys) {
+ return new FormatTableCommit(
+ tableLocation.toString(),
+ Arrays.asList(partitionKeys),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ null,
+ null,
+ null,
+ null,
+ dynamicPartitionOverwrite);
+ }
+
private static Map<String, String> partitionSpec(String year, String
month) {
LinkedHashMap<String, String> spec = new LinkedHashMap<>();
spec.put("year", year);
@@ -567,7 +680,8 @@ class FormatTableCommitTest {
null,
null,
null,
- partitionManager);
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
}
private FormatTablePartitionManager commitPartitionedFile(
@@ -591,7 +705,8 @@ class FormatTableCommitTest {
null,
null,
null,
- partitionManager);
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
commit.commit(Collections.singletonList(new
TwoPhaseCommitMessage(committer)));
return partitionManager;
}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
index c05349c504..430dbb4b2c 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
@@ -31,10 +31,9 @@ import org.apache.spark.sql.types.StructType
*/
class FormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType)
- extends FormatTableBatchWriteBase(table, overwriteDynamic,
overwritePartitions, writeSchema)
+ extends FormatTableBatchWriteBase(table, overwritePartitions, writeSchema)
with BatchWrite
with Serializable {
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 12dddf4db5..8cdc6e453d 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -232,10 +232,9 @@ class Spark4Shim extends SparkShim {
override def createFormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType): BatchWrite =
- new FormatTableBatchWrite(table, overwriteDynamic, overwritePartitions,
writeSchema)
+ new FormatTableBatchWrite(table, overwritePartitions, writeSchema)
override def createCTERelationRef(
cteId: Long,
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
index 853c7f4146..c7ab592a1b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
@@ -43,7 +43,6 @@ import scala.collection.JavaConverters._
*/
abstract class FormatTableBatchWriteBase(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType)
extends Logging
@@ -51,12 +50,10 @@ abstract class FormatTableBatchWriteBase(
protected val batchWriteBuilder: BatchWriteBuilder = {
val builder = table.newBatchWriteBuilder()
- // todo: add test for static overwrite the whole table
- if (overwriteDynamic.contains(true)) {
- builder.withOverwrite()
- } else {
- overwritePartitions.foreach(partitions =>
builder.withOverwrite(partitions.asJava))
- }
+ // Which partitions an overwrite that names none replaces is the table's
+ // `dynamic-partition-overwrite` to answer, and the builder carries the
mode Spark resolved
+ // in that option, so an empty spec here means the statement, not the mode.
+ overwritePartitions.foreach(partitions =>
builder.withOverwrite(partitions.asJava))
builder
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
index fea982207d..baa7d21e40 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
@@ -367,20 +367,30 @@ case class PaimonFormatTableWriterBuilder(table:
FormatTable, writeSchema: Struc
override def partitionRowType(): RowType = table.partitionType
- override def build: Write = new Write with RequiresDistributionAndOrdering {
- private val writeRequirement = PaimonWriteRequirement(table)
+ override def build: Write = {
+ // Which partitions an overwrite replaces is the table option's call, the
same as for a data
+ // table. Carrying the mode Spark resolved into that option is what keeps
a `STATIC` overwrite
+ // from being served as if it were `DYNAMIC`.
+ val writeTable = overwriteDynamic match {
+ case Some(dynamic) =>
+ table.copy(Map(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key ->
dynamic.toString).asJava)
+ case None => table
+ }
+ new Write with RequiresDistributionAndOrdering {
+ private val writeRequirement = PaimonWriteRequirement(writeTable)
- override def requiredDistribution(): Distribution =
writeRequirement.distribution
+ override def requiredDistribution(): Distribution =
writeRequirement.distribution
- override def requiredOrdering(): Array[SortOrder] =
writeRequirement.ordering
+ override def requiredOrdering(): Array[SortOrder] =
writeRequirement.ordering
- override def toBatch: BatchWrite = {
- SparkShimLoader.shim
- .createFormatTableBatchWrite(table, overwriteDynamic,
overwritePartitions, writeSchema)
- }
+ override def toBatch: BatchWrite = {
+ SparkShimLoader.shim
+ .createFormatTableBatchWrite(writeTable, overwritePartitions,
writeSchema)
+ }
- override def toStreaming: StreamingWrite = {
- throw new UnsupportedOperationException("FormatTable doesn't support
streaming write")
+ override def toStreaming: StreamingWrite = {
+ throw new UnsupportedOperationException("FormatTable doesn't support
streaming write")
+ }
}
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index 690c20b85b..8d0cd71579 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -151,7 +151,6 @@ trait SparkShim {
/** Same `BatchWrite` mixin problem as [[createPaimonBatchWrite]], but for
`FormatTable` writes. */
def createFormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType): BatchWrite
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
index dc523284bb..998c7c591c 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala
@@ -576,6 +576,54 @@ abstract class FormatTableTestBase extends
PaimonHiveTestBase with AdaptiveSpark
}
}
+ test(
+ "Format table: INSERT OVERWRITE empties an unpartitioned table when the
query returns nothing") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, payload STRING) USING CSV")
+ sql("INSERT INTO t VALUES (1, 'a'), (2, 'b')")
+
+ sql("INSERT OVERWRITE t SELECT * FROM t WHERE false")
+
+ // An unpartitioned overwrite replaces the table, and it replaces it
with nothing here. Left
+ // to the files this commit wrote, an empty query would leave the old
rows readable.
+ checkAnswer(sql("SELECT * FROM t"), Seq.empty)
+ }
+ }
+
+ test("Format table: static INSERT OVERWRITE replaces every partition of the
table") {
+ withTable("t") {
+ withSQLConf("spark.sql.sources.partitionOverwriteMode" -> "STATIC") {
+ sql("CREATE TABLE t (id INT, dt STRING) USING CSV PARTITIONED BY (dt)")
+ sql("INSERT INTO t VALUES (1, '20260101'), (2, '20260102')")
+
+ sql("INSERT OVERWRITE t VALUES (9, '20260101')")
+
+ // The statement names no partition and the mode is STATIC, so it is
about the whole
+ // table: the partition it does not write is replaced too, not left as
it was.
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(9, "20260101")))
+
+ sql("INSERT OVERWRITE t SELECT * FROM t WHERE false")
+ checkAnswer(sql("SELECT * FROM t"), Seq.empty)
+ }
+ }
+ }
+
+ test("Format table: dynamic INSERT OVERWRITE replaces only the partitions it
writes") {
+ withTable("t") {
+ withSQLConf("spark.sql.sources.partitionOverwriteMode" -> "DYNAMIC") {
+ sql("CREATE TABLE t (id INT, dt STRING) USING CSV PARTITIONED BY (dt)")
+ sql("INSERT INTO t VALUES (1, '20260101'), (2, '20260102')")
+
+ sql("INSERT OVERWRITE t VALUES (9, '20260101')")
+ checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(2,
"20260102"), Row(9, "20260101")))
+
+ // Nothing written means no partition selected, which is not the same
as selecting all.
+ sql("INSERT OVERWRITE t SELECT * FROM t WHERE false")
+ checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(2,
"20260102"), Row(9, "20260101")))
+ }
+ }
+ }
+
def collectFilteredInputSplits(plan: SparkPlan, tableName: String):
Seq[Split] = {
flatMap(plan) {
case s: BatchScanExec =>
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
index fedf8363d0..e554c72322 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
@@ -203,11 +203,21 @@ class PaimonFormatTableTest extends
PaimonSparkTestWithRestCatalogBase {
Row(1, 5, "Jerry") :: Row(1, 7, "Tom") :: Nil
)
spark.sql(s"INSERT INTO $tableName PARTITION (id = 3) VALUES (5,
'Alice')")
+ // No PARTITION clause under Spark's default STATIC mode: the statement
is about the whole
+ // table, so the partition it does not write is replaced too.
spark.sql(s"INSERT OVERWRITE $tableName VALUES (5, 'Jerry', 1), (7,
'Tom', 2)")
checkAnswer(
spark.sql(s"SELECT id, age, name FROM $tableName ORDER BY id, age"),
- Row(1, 5, "Jerry") :: Row(2, 7, "Tom") :: Row(3, 5, "Alice") :: Nil
+ Row(1, 5, "Jerry") :: Row(2, 7, "Tom") :: Nil
)
+ withSparkSQLConf("spark.sql.sources.partitionOverwriteMode" ->
"dynamic") {
+ spark.sql(s"INSERT INTO $tableName PARTITION (id = 3) VALUES (5,
'Alice')")
+ spark.sql(s"INSERT OVERWRITE $tableName VALUES (9, 'Jerry', 1)")
+ checkAnswer(
+ spark.sql(s"SELECT id, age, name FROM $tableName ORDER BY id, age"),
+ Row(1, 9, "Jerry") :: Row(2, 7, "Tom") :: Row(3, 5, "Alice") :: Nil
+ )
+ }
}
}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
index d13c737370..a00b8dd284 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
@@ -29,10 +29,9 @@ import org.apache.spark.sql.types.StructType
*/
class FormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType)
- extends FormatTableBatchWriteBase(table, overwriteDynamic,
overwritePartitions, writeSchema)
+ extends FormatTableBatchWriteBase(table, overwritePartitions, writeSchema)
with BatchWrite
with Serializable {
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index b568ce16c3..3ea437da35 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -218,10 +218,9 @@ class Spark3Shim extends SparkShim {
override def createFormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType): BatchWrite =
- new FormatTableBatchWrite(table, overwriteDynamic, overwritePartitions,
writeSchema)
+ new FormatTableBatchWrite(table, overwritePartitions, writeSchema)
override def createCTERelationRef(
cteId: Long,
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
index 4fa58c95a7..5a114f1300 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWrite.scala
@@ -31,10 +31,9 @@ import org.apache.spark.sql.types.StructType
*/
class FormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType)
- extends FormatTableBatchWriteBase(table, overwriteDynamic,
overwritePartitions, writeSchema)
+ extends FormatTableBatchWriteBase(table, overwritePartitions, writeSchema)
with BatchWrite
with Serializable {
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index eea503d7d5..4ae3bb1892 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -216,10 +216,9 @@ class Spark4Shim extends SparkShim {
override def createFormatTableBatchWrite(
table: FormatTable,
- overwriteDynamic: Option[Boolean],
overwritePartitions: Option[Map[String, String]],
writeSchema: StructType): BatchWrite =
- new FormatTableBatchWrite(table, overwriteDynamic, overwritePartitions,
writeSchema)
+ new FormatTableBatchWrite(table, overwritePartitions, writeSchema)
override def createCTERelationRef(
cteId: Long,
