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 03a9fa2cdb [core][spark][flink] Support TRUNCATE TABLE on format
tables (#9330)
03a9fa2cdb is described below
commit 03a9fa2cdbe7c473e46d1e7ad4595780dd62d0d0
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Aug 21 21:53:41 2026 +0800
[core][spark][flink] Support TRUNCATE TABLE on format tables (#9330)
---
docs/docs/flink/sql-write.mdx | 7 +
docs/docs/spark/sql-write.md | 10 +
.../paimon/table/format/FormatTableCommit.java | 204 +++++++++++++--
.../apache/paimon/table/sink/BatchTableCommit.java | 6 +-
.../format/FormatTableCommitStatisticsTest.java | 281 +++++++++++++++++++++
.../paimon/table/format/FormatTableCommitTest.java | 211 ++++++++++++++++
.../paimon/flink/sink/FlinkFormatTableSink.java | 35 +++
.../paimon/flink/sink/FlinkFormatTableSink.java | 35 +++
.../paimon/flink/sink/FlinkFormatTableSink.java | 79 ++----
...ableSink.java => FlinkFormatTableSinkBase.java} | 12 +-
.../paimon/flink/source/FormatTableITCase.java | 94 +++++--
.../paimon/spark/PaimonPartitionManagement.scala | 8 +-
.../paimon/spark/format/PaimonFormatTable.scala | 96 ++++++-
.../spark/format/CatalogManagedPartitionTest.scala | 23 ++
.../FormatTablePartitionManagementTest.scala | 110 +++++++-
.../CatalogManagedPartitionMsckRepairTest.scala | 33 +++
.../paimon/spark/sql/FormatTableTestBase.scala | 51 ++++
17 files changed, 1169 insertions(+), 126 deletions(-)
diff --git a/docs/docs/flink/sql-write.mdx b/docs/docs/flink/sql-write.mdx
index 8f1d8e3cfb..aea1ba870c 100644
--- a/docs/docs/flink/sql-write.mdx
+++ b/docs/docs/flink/sql-write.mdx
@@ -142,6 +142,13 @@ TRUNCATE TABLE my_table;
</Tabs>
+On a Format Table read through Paimon (`format-table.implementation = paimon`,
the default),
+`TRUNCATE TABLE` deletes the data files and keeps the partitions: their
directories remain, and
+with `metastore.partitioned-table = true` so do their catalog registrations,
whose statistics are
+replaced with zero. That setting also makes the catalog the answer to which
partitions the table
+has, so truncating empties those and leaves an unregistered directory alone.
Flink has no
+`TRUNCATE TABLE ... PARTITION`; use Spark's.
+
## Purging Partitions
Currently, Paimon supports two ways to purge partitions.
diff --git a/docs/docs/spark/sql-write.md b/docs/docs/spark/sql-write.md
index 1cc6c0bc69..5ac34f0834 100644
--- a/docs/docs/spark/sql-write.md
+++ b/docs/docs/spark/sql-write.md
@@ -129,8 +129,18 @@ The `TRUNCATE TABLE` statement removes all the rows from a
table or partition(s)
```sql
TRUNCATE TABLE my_table;
+TRUNCATE TABLE my_table PARTITION (dt = '2025-01-01');
```
+On a Format Table read through Paimon (`format-table.implementation = paimon`,
the default),
+`TRUNCATE TABLE` deletes the data files of the table or of the named
partitions and keeps the
+partitions: their directories remain, and with `metastore.partitioned-table =
true` so do their
+catalog registrations, so `SHOW PARTITIONS` returns what it returned before.
That setting also
+makes the catalog the answer to which partitions the table has, so truncating
empties those, leaves
+a directory still waiting for `MSCK REPAIR TABLE` alone, and replaces their
statistics with zero. A
+spec that names only some of the partition keys empties the partitions it
covers; a complete spec
+the table does not have is an error.
+
## Update Table
Updates the column values for the rows that match a predicate. When no
predicate is provided, update the column values for all rows.
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 ebf5c39c50..aec5f73e15 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
@@ -30,11 +30,13 @@ import org.apache.paimon.fs.TwoPhaseOutputStream;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.TableCommit;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PartitionPathUtils;
import org.slf4j.Logger;
@@ -53,6 +55,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import static
org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
@@ -199,7 +202,11 @@ public class FormatTableCommit implements BatchTableCommit
{
}
if (reportsStatistics) {
reportPartitions(
- partitionSpecs, statisticsByPartition,
clearedPartitionPaths, commitTime);
+ partitionSpecs,
+ statisticsByPartition,
+ clearedPartitionPaths,
+ commitTime,
+ overwrite);
} else if (partitionManager != null && !partitionSpecs.isEmpty()) {
// Concurrent writers may touch the same partition, so
registration ignores the
// ones that already exist rather than failing the commit.
@@ -234,26 +241,20 @@ 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.
+ * its own request registers. A truncation writes nothing and reports
every partition it
+ * emptied.
*/
private void reportPartitions(
Set<Map<String, String>> writtenPartitionSpecs,
Map<Map<String, String>, PartitionStatistics>
statisticsByPartition,
Set<Path> clearedPartitionPaths,
- long commitTime) {
+ long commitTime,
+ boolean replaceStatistics) {
for (Path cleared : clearedPartitionPaths) {
Map<String, String> spec = clearedPartitionSpec(cleared);
if (spec != null) {
// Emptied and not written to: an exact zero, dated to the
commit that did it.
- statisticsByPartition.putIfAbsent(
- spec,
- new PartitionStatistics(
- spec,
- 0,
- 0,
- 0,
- commitTime,
- PartitionStatistics.UNKNOWN_TOTAL_BUCKETS));
+ statisticsByPartition.putIfAbsent(spec, emptyStatistics(spec,
commitTime));
}
}
@@ -263,13 +264,13 @@ public class FormatTableCommit implements
BatchTableCommit {
if (specs.isEmpty()) {
return;
}
- // An overwriting commit replaced what the partitions held, so what it
wrote is the total;
- // an appending one saw only its own files, so its numbers are an
increment.
+ // A commit that replaced what the partitions held reports a total; an
appending one saw
+ // only its own files, so its numbers are an increment.
partitionManager.createPartitions(
new ArrayList<>(specs),
true,
new ArrayList<>(statisticsByPartition.values()),
- overwrite);
+ replaceStatistics);
}
/** What one commit wrote into a partition, with one more of its files
folded in. */
@@ -388,6 +389,12 @@ public class FormatTableCommit implements BatchTableCommit
{
if (partitionSpec.isEmpty() || partitionKeys.isEmpty()) {
throw new IllegalArgumentException("partitionSpec or partitionKeys
is empty.");
}
+ if (partitionSpec.size() > partitionKeys.size()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition spec %s names more values than the
partition keys %s.",
+ partitionSpec, partitionKeys));
+ }
LinkedHashMap<String, String> orderedSpec = new LinkedHashMap<>();
for (int i = 0; i < partitionSpec.size(); i++) {
String key = partitionKeys.get(i);
@@ -442,16 +449,26 @@ public class FormatTableCommit implements
BatchTableCommit {
partitionLevels,
formatTablePartitionOnlyValueInPath,
defaultPartName)) {
+ boolean deleted;
try {
- // Only what this commit removed: a file another writer
deleted first would
- // have every concurrent writer report the whole subtree.
- if (fileIO.delete(file.getPath(), false)) {
- clearedPartitionPaths.add(file.getPath().getParent());
- }
+ deleted = fileIO.delete(file.getPath(), false);
} catch (FileNotFoundException ignore) {
+ continue;
} catch (IOException e) {
throw new RuntimeException(e);
}
+ if (deleted) {
+ // Only what this commit removed: a file another writer
deleted first would
+ // have every concurrent writer report the whole subtree.
+ clearedPartitionPaths.add(file.getPath().getParent());
+ } else if (fileIO.exists(file.getPath())) {
+ // A refusal is not that race: the file is still readable,
and going on would
+ // report the partition as holding nothing while its rows
are still there.
+ throw new IOException(
+ String.format(
+ "Failed to delete data file %s of table
%s.",
+ file.getPath(),
tableIdentifier.getFullName()));
+ }
}
}
return clearedPartitionPaths;
@@ -459,12 +476,155 @@ public class FormatTableCommit implements
BatchTableCommit {
@Override
public void truncateTable() {
- throw new UnsupportedOperationException();
+ // Data files only. The partition directories stay, and so do their
catalog registrations:
+ // emptying a table does not redefine which partitions it has.
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ try {
+ deletePreviousDataFile(new Path(location), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate table %s.",
tableIdentifier.getFullName()),
+ e);
+ }
+ return;
+ }
+ // Emptying the table is emptying every partition it has, and which
those are is answered
+ // by whatever the table reads its partitions from.
+ if (partitionManager != null) {
+ truncate(registeredPartitions(Collections.emptyMap()));
+ return;
+ }
+ // 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)) {
+ try {
+ deletePreviousDataFile(partition.getRight(), 0);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of table %s.",
+ partition.getLeft(),
tableIdentifier.getFullName()),
+ e);
+ }
+ }
}
@Override
public void truncatePartitions(List<Map<String, String>> partitionSpecs) {
- throw new UnsupportedOperationException();
+ if (partitionManager == null) {
+ truncate(partitionSpecs);
+ return;
+ }
+ // Complete specs are asked for in one request; only a prefix has to
be listed on its own.
+ List<Map<String, String>> complete = new ArrayList<>();
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ if (partitionSpec.size() == partitionKeys.size()) {
+ complete.add(partitionSpec);
+ }
+ }
+ Set<Map<String, String>> registered =
+ complete.isEmpty()
+ ? Collections.emptySet()
+ :
partitionManager.listPartitionsByNames(complete).stream()
+ .map(Partition::spec)
+ .collect(Collectors.toSet());
+ List<Map<String, String>> partitions = new ArrayList<>();
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ if (partitionSpec.size() == partitionKeys.size()) {
+ if (registered.contains(partitionSpec)) {
+ partitions.add(partitionSpec);
+ }
+ } else {
+ partitions.addAll(registeredPartitions(partitionSpec));
+ }
+ }
+ truncate(partitions);
+ }
+
+ /**
+ * The registered partitions named by {@code prefix}, which names only the
leading partition
+ * keys, or none of them. The catalog says which partitions a
catalog-managed table has, so
+ * truncating neither empties nor registers a directory still waiting for
MSCK REPAIR TABLE.
+ */
+ private List<Map<String, String>> registeredPartitions(Map<String, String>
prefix) {
+ return partitionManager.listPartitions(prefix, null).stream()
+ .map(Partition::spec)
+ .collect(Collectors.toList());
+ }
+
+ private void truncate(List<Map<String, String>> partitionSpecs) {
+ long truncateTime = System.currentTimeMillis();
+ Set<Path> clearedPartitionPaths = new HashSet<>();
+ // Statistics are keyed by the spec that named the partition, so only
a complete one can
+ // seed them; a prefix reaches here only for a table with nowhere to
report to.
+ Map<Map<String, String>, PartitionStatistics> emptied = new
LinkedHashMap<>();
+ RuntimeException failure = null;
+ for (Map<String, String> partitionSpec : partitionSpecs) {
+ Path partitionPath =
+ buildPartitionPath(
+ location,
+ partitionSpec,
+ formatTablePartitionOnlyValueInPath,
+ partitionKeys);
+ try {
+ clearedPartitionPaths.addAll(
+ deletePreviousDataFile(
+ partitionPath, partitionKeys.size() -
partitionSpec.size()));
+ } catch (Exception e) {
+ failure =
+ new RuntimeException(
+ String.format(
+ "Failed to truncate partition %s of
table %s.",
+ partitionSpec,
tableIdentifier.getFullName()),
+ e);
+ break;
+ }
+ if (partitionSpec.size() == partitionKeys.size()) {
+ emptied.put(partitionSpec, emptyStatistics(partitionSpec,
truncateTime));
+ }
+ }
+ if (partitionManager != null) {
+ // Truncating states that the partition holds nothing, whoever
deleted the files, so
+ // one that was already empty reports zero as well. An overwrite
reports only what it
+ // removed itself, so that concurrent writers do not each claim
the whole subtree;
+ // truncation makes the claim on purpose. What a failed truncation
emptied is reported
+ // too, so the catalog stops describing files that are gone.
+ try {
+ reportPartitions(
+ Collections.emptySet(),
+ emptied,
+ clearedPartitionPaths,
+ truncateTime,
+ /* replaceStatistics */ true);
+ } catch (RuntimeException e) {
+ if (failure == null) {
+ throw e;
+ }
+ // The deletion that failed first is the one that explains
what went wrong.
+ failure.addSuppressed(e);
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ /** What a partition holds once it has been emptied, dated to the commit
that emptied it. */
+ private static PartitionStatistics emptyStatistics(
+ Map<String, String> partitionSpec, long emptiedTime) {
+ return new PartitionStatistics(
+ partitionSpec, 0, 0, 0, emptiedTime,
PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
index 2cfc181ead..52e11e3c84 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
@@ -57,13 +57,15 @@ public interface BatchTableCommit extends TableCommit {
/**
* Truncate table, like normal {@link #commit}, files are not immediately
deleted, they are only
- * logically deleted and will be deleted after the snapshot expires.
+ * logically deleted and will be deleted after the snapshot expires. A
table that keeps no
+ * snapshots, such as a Format Table, deletes them right away.
*/
void truncateTable();
/**
* Truncate partitions, like normal {@link #commit}, files are not
immediately deleted, they are
- * only logically deleted and will be deleted after the snapshot expires.
+ * only logically deleted and will be deleted after the snapshot expires.
A table that keeps no
+ * snapshots, such as a Format Table, deletes them right away.
*/
void truncatePartitions(List<Map<String, String>> partitionSpecs);
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 ed26d70742..199b7b446b 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
@@ -67,6 +67,7 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Tests for the partition statistics a {@link FormatTableCommit} reports. */
class FormatTableCommitStatisticsTest {
@@ -222,6 +223,225 @@ class FormatTableCommitStatisticsTest {
});
}
+ @Test
+ void testTruncatingPartitionsReportsAnExactZeroAsTheTotal() 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", "data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv",
2048);
+
+ long before = System.currentTimeMillis();
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .truncatePartitions(Arrays.asList(spec("2025", "10"),
spec("2025", "11")));
+ long after = System.currentTimeMillis();
+
+ Reported reported = capture(partitionManager);
+ // What a truncated partition holds is zero, not zero fewer rows than
before.
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.specs)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ // Red line: emptying a partition zeroes its statistics, it never
unregisters it.
+ verify(partitionManager, never()).dropPartitions(anyList());
+ // One catalog request for the complete specs, not one per partition.
+ verify(partitionManager).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(any(), any());
+ // Statistics route by the spec they carry, so every partition needs
its own.
+ assertThat(reported.statistics)
+ .extracting(PartitionStatistics::spec)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ assertThat(reported.statistics)
+ .allSatisfy(
+ statistics -> {
+ assertThat(statistics.recordCount()).isZero();
+ assertThat(statistics.fileSizeInBytes()).isZero();
+ assertThat(statistics.fileCount()).isZero();
+ // Dated to the truncation. Reporting the time as
unknown would leave
+ // the stored one describing files that are gone,
since an unknown
+ // replaces nothing.
+
assertThat(statistics.lastFileCreationTime()).isBetween(before, after);
+ });
+ }
+
+ @Test
+ void testTruncatingAPartitionThatIsAlreadyEmptyStillReportsZero() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ // Registered, and its directory holds nothing to delete - whoever
emptied it, its stored
+ // statistics still describe the files that used to be there.
+ registered(partitionManager, spec("2025", "10"));
+ fileIO.mkdirs(new Path(tablePath, "year=2025/month=10"));
+
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .truncatePartitions(Collections.singletonList(spec("2025",
"10")));
+
+ Reported reported = capture(partitionManager);
+ // Unlike an overwrite, which reports only the files it removed
itself, truncation states
+ // that the partition holds nothing.
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.specs).containsExactly(spec("2025", "10"));
+ assertThat(reported.statistics).hasSize(1);
+ assertThat(reported.statistics.get(0).spec()).isEqualTo(spec("2025",
"10"));
+ assertThat(reported.statistics.get(0).recordCount()).isZero();
+ assertThat(reported.statistics.get(0).fileCount()).isZero();
+ }
+
+ @Test
+ void
testTruncatingTheTableReportsARegisteredPartitionWhoseDirectoryIsGone() 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", "data.csv",
4096);
+ // Registered but its directory is not there at all: someone deleted
it behind the catalog.
+ // A missing directory is drift to report on, not a reason to fail the
truncation of the
+ // partitions that do exist.
+
+ commit(tablePath, fileIO, partitionManager, false,
null).truncateTable();
+
+ Reported reported = capture(partitionManager);
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.specs)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ assertThat(reported.statistics)
+ .allSatisfy(
+ statistics -> {
+ assertThat(statistics.recordCount()).isZero();
+ assertThat(statistics.fileCount()).isZero();
+ });
+ }
+
+ @Test
+ void testTruncatingAPrefixReportsThePartitionsUnderneathIt() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ Map<String, String> prefix = Collections.singletonMap("year", "2025");
+ when(partitionManager.listPartitions(prefix, null))
+ .thenReturn(
+ Arrays.asList(
+ partition(spec("2025", "10")),
partition(spec("2025", "11"))));
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv",
2048);
+ writeDataFile(fileIO, tablePath, "year=2024/month=12", "data.csv",
1024);
+ // Under the prefix but not registered: a directory still waiting for
MSCK REPAIR TABLE is
+ // not a partition of the table, so truncating neither empties nor
registers it.
+ writeDataFile(fileIO, tablePath, "year=2025/month=12", "data.csv",
512);
+
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .truncatePartitions(Collections.singletonList(prefix));
+
+ Reported reported = capture(partitionManager);
+ // The prefix names no partition of its own; the partitions it empties
are the registered
+ // ones underneath it.
+ assertThat(reported.specs)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.statistics)
+ .allSatisfy(statistics ->
assertThat(statistics.recordCount()).isZero());
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=12/data.csv"))).isTrue();
+ }
+
+ @Test
+ void testTruncatingTheTableReportsEveryRegisteredPartition() 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", "data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv",
2048);
+ // Not registered, so not part of the table and not reported on.
+ writeDataFile(fileIO, tablePath, "year=2025/month=12", "data.csv",
512);
+
+ commit(tablePath, fileIO, partitionManager, false,
null).truncateTable();
+
+ Reported reported = capture(partitionManager);
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.specs)
+ .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025",
"11"));
+ assertThat(reported.statistics)
+ .allSatisfy(statistics ->
assertThat(statistics.fileCount()).isZero());
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=12/data.csv"))).isTrue();
+ }
+
+ @Test
+ void testAFailedTruncationReportsThePartitionsItAlreadyEmptied() throws
Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LocalFileIO fileIO = new UndeletableFileIO(new Path(tablePath,
"year=2025/month=11"));
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "10"), spec("2025", "11"));
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv",
2048);
+
+ assertThatThrownBy(
+ () ->
+ commit(tablePath, fileIO, partitionManager,
false, null)
+ .truncateTable())
+ .hasMessageContaining("month=11")
+ .hasMessageContaining(TABLE.getFullName());
+
+ // A Format Table has no snapshot to make the whole truncation atomic,
so what it emptied
+ // before the failure is reported anyway: the catalog must not keep
describing files that
+ // are gone.
+ Reported reported = capture(partitionManager);
+ assertThat(reported.specs).containsExactly(spec("2025", "10"));
+ assertThat(reported.statistics.get(0).fileCount()).isZero();
+ }
+
+ @Test
+ void testATruncationFailsWhenADeletionIsRefused() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LocalFileIO fileIO = new RefusingFileIO();
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "10"));
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv",
4096);
+
+ assertThatThrownBy(
+ () ->
+ commit(tablePath, fileIO, partitionManager,
false, null)
+ .truncateTable())
+ .hasMessageContaining("month=10")
+ .hasMessageContaining(TABLE.getFullName())
+ .hasStackTraceContaining("data.csv");
+
+ // A refused deletion is not a concurrent one: the rows are still
readable, so reporting
+ // the partition as holding nothing would hide them.
+ verify(partitionManager, never())
+ .createPartitions(anyList(), anyBoolean(), anyList(),
anyBoolean());
+ assertThat(fileIO.exists(new Path(tablePath,
"year=2025/month=10/data.csv"))).isTrue();
+ }
+
+ @Test
+ void testAFailedReportDoesNotHideTheDeletionThatFailedFirst() throws
Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LocalFileIO fileIO = new UndeletableFileIO(new Path(tablePath,
"year=2025/month=11"));
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "10"), spec("2025", "11"));
+ doThrow(new RuntimeException("the catalog is unreachable"))
+ .when(partitionManager)
+ .createPartitions(anyList(), anyBoolean(), anyList(),
anyBoolean());
+ writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv",
4096);
+ writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv",
2048);
+
+ assertThatThrownBy(
+ () ->
+ commit(tablePath, fileIO, partitionManager,
false, null)
+ .truncateTable())
+ // The deletion that failed first is what explains the
failure; the report that
+ // could not record the rest is attached to it.
+ .hasMessageContaining("month=11")
+ .satisfies(
+ thrown ->
+ assertThat(thrown.getSuppressed())
+ .anySatisfy(
+ suppressed ->
+ assertThat(suppressed)
+
.hasMessageContaining(
+
"unreachable")));
+ }
+
@Test
void
testAClearedPartitionIsFoundEvenWhenTheListingAnswersUnderAnotherScheme()
throws Exception {
@@ -416,6 +636,36 @@ class FormatTableCommitStatisticsTest {
}
}
+ /** A partition the catalog holds, whose statistics the truncation is
meant to replace. */
+ private static Partition partition(Map<String, String> spec) {
+ return new Partition(spec, 3, 4096, 1, 0,
PartitionStatistics.UNKNOWN_TOTAL_BUCKETS, false);
+ }
+
+ /** Which partitions the catalog says the table has. */
+ @SafeVarargs
+ private static void registered(
+ FormatTablePartitionManager partitionManager, Map<String,
String>... specs) {
+ List<Partition> partitions = new ArrayList<>();
+ for (Map<String, String> spec : specs) {
+ partitions.add(partition(spec));
+ when(partitionManager.listPartitions(spec, null))
+ .thenReturn(Collections.singletonList(partition(spec)));
+ }
+ when(partitionManager.listPartitions(Collections.emptyMap(),
null)).thenReturn(partitions);
+ when(partitionManager.listPartitionsByNames(anyList()))
+ .thenAnswer(
+ invocation -> {
+ List<Map<String, String>> asked =
invocation.getArgument(0);
+ List<Partition> found = new ArrayList<>();
+ for (Partition partition : partitions) {
+ if (asked.contains(partition.spec())) {
+ found.add(partition);
+ }
+ }
+ return found;
+ });
+ }
+
private static Map<String, String> spec(String year, String month) {
LinkedHashMap<String, String> spec = new LinkedHashMap<>();
spec.put("year", year);
@@ -446,6 +696,37 @@ class FormatTableCommitStatisticsTest {
* A {@link FileIO} that answers a listing with paths stripped of their
scheme, the way a
* delegating one does when it resolves the caller's scheme to the one it
really uses.
*/
+ /** A file IO whose deletions all report failure, leaving the files in
place. */
+ private static class RefusingFileIO extends LocalFileIO {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ return false;
+ }
+ }
+
+ /** A file IO that refuses to delete anything under one directory. */
+ private static class UndeletableFileIO extends LocalFileIO {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String directory;
+
+ private UndeletableFileIO(Path directory) {
+ this.directory = directory.toString();
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) throws IOException
{
+ if (path.toString().startsWith(directory)) {
+ throw new IOException("Refused to delete " + path);
+ }
+ return super.delete(path, recursive);
+ }
+ }
+
private static class RescopingFileIO extends LocalFileIO {
private static final long serialVersionUID = 1L;
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 6d3d6da6f2..db12129814 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
@@ -23,6 +23,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.RenamingTwoPhaseOutputStream;
import org.apache.paimon.fs.TwoPhaseOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.partition.Partition;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.utils.PartitionPathUtils;
@@ -49,6 +50,7 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Tests for {@link FormatTableCommit}. */
class FormatTableCommitTest {
@@ -359,6 +361,215 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(siblingPath)).isTrue();
}
+ @Test
+ void
testTruncateTableEmptiesEveryPartitionAndLeavesThePartitionsThemselves() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path october = new Path(tablePath, "year=2025/month=10");
+ Path november = new Path(tablePath, "year=2025/month=11");
+ Path octoberData = new Path(october, "data-1.csv");
+ Path novemberData = new Path(november, "data-2.csv");
+ fileIO.writeFile(octoberData, "1", false);
+ fileIO.writeFile(novemberData, "2", false);
+ // Another writer is mid-write in this partition; its staging tree is
not table data.
+ Path stagingFile = new Path(october,
"_temporary/attempt/part-00000.csv");
+ fileIO.writeFile(stagingFile, "3", false);
+ FormatTableCommit commit =
+ truncatingCommit(tablePath, fileIO, false, null, "year",
"month");
+
+ commit.truncateTable();
+
+ assertThat(fileIO.exists(octoberData)).isFalse();
+ assertThat(fileIO.exists(novemberData)).isFalse();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ // Emptying a table does not redefine which partitions it has: the
directories stay.
+ assertThat(fileIO.exists(october)).isTrue();
+ assertThat(fileIO.exists(november)).isTrue();
+ }
+
+ @Test
+ void
testTruncateTableEmptiesTheRegisteredPartitionsOfACatalogManagedTable() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path registered = new Path(tablePath, "year=2025/month=10");
+ Path registeredData = new Path(registered, "data-1.csv");
+ fileIO.writeFile(registeredData, "1", false);
+ // Dropped there by something outside Paimon and not registered yet,
so it is not part of
+ // the table: MSCK REPAIR TABLE is what would make it so.
+ Path awaitingRepair = new Path(tablePath, "year=2025/month=11");
+ Path awaitingRepairData = new Path(awaitingRepair, "data-2.csv");
+ fileIO.writeFile(awaitingRepairData, "2", false);
+ FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitions(Collections.emptyMap(), null))
+ .thenReturn(
+ Collections.singletonList(
+ new Partition(partitionSpec("2025", "10"), 0,
0, 0, 0, -1, false)));
+ FormatTableCommit commit =
+ truncatingCommit(tablePath, fileIO, false, partitionManager,
"year", "month");
+
+ commit.truncateTable();
+
+ assertThat(fileIO.exists(registeredData)).isFalse();
+ assertThat(fileIO.exists(registered)).isTrue();
+ assertThat(fileIO.exists(awaitingRepairData)).isTrue();
+ }
+
+ @Test
+ void testTruncateTableOnlyEmptiesTheDirectoriesThatAreItsPartitions()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path partitionData = new Path(tablePath,
"year=2025/month=10/data-1.csv");
+ fileIO.writeFile(partitionData, "1", false);
+ // Neither of these is a partition: the scan reads the directories
that parse into the
+ // partition keys, so nothing else under the table directory is table
data.
+ Path atTheTableRoot = new Path(tablePath, "notes.csv");
+ fileIO.writeFile(atTheTableRoot, "2", false);
+ Path outsideThePartitionLayout = new Path(tablePath,
"tmp/unknown/x.csv");
+ fileIO.writeFile(outsideThePartitionLayout, "3", false);
+ FormatTableCommit commit =
+ truncatingCommit(tablePath, fileIO, false, null, "year",
"month");
+
+ commit.truncateTable();
+
+ assertThat(fileIO.exists(partitionData)).isFalse();
+ assertThat(fileIO.exists(atTheTableRoot)).isTrue();
+ assertThat(fileIO.exists(outsideThePartitionLayout)).isTrue();
+ }
+
+ @Test
+ void
testTruncateTableClearsAValueOnlyDefaultPartitionBelowTheTableDirectory()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ // Value-only layout: a null month is the default partition name,
which starts with '_'
+ // without being a staging directory, and here it sits below the
listed directory instead
+ // of being it.
+ Path defaultPartition =
+ new Path(tablePath, "2025/" +
PARTITION_DEFAULT_NAME.defaultValue());
+ Path staleFile = new Path(defaultPartition, "data-old.csv");
+ fileIO.writeFile(staleFile, "1", false);
+ Path stagingFile = new Path(tablePath,
"2025/_temporary/attempt/part-00000.csv");
+ fileIO.writeFile(stagingFile, "2", false);
+ FormatTableCommit commit = truncatingCommit(tablePath, fileIO, true,
null, "year", "month");
+
+ commit.truncateTable();
+
+ assertThat(fileIO.exists(staleFile)).isFalse();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ assertThat(fileIO.exists(defaultPartition)).isTrue();
+ }
+
+ @Test
+ void testTruncateTableOfAnUnpartitionedTableClearsTheTableDirectory()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = new Path(tablePath, "data-1.csv");
+ fileIO.writeFile(dataFile, "1", false);
+ Path stagingFile = new Path(tablePath,
"_temporary/attempt/part-00000.csv");
+ fileIO.writeFile(stagingFile, "2", false);
+ FormatTableCommit commit = truncatingCommit(tablePath, fileIO, false,
null);
+
+ commit.truncateTable();
+
+ assertThat(fileIO.exists(dataFile)).isFalse();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ assertThat(fileIO.exists(tablePath)).isTrue();
+ }
+
+ @Test
+ void testTruncatePartitionsStaysInsideThePartitionsItNames() throws
Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path october = new Path(tablePath, "year=2025/month=10");
+ Path november = new Path(tablePath, "year=2025/month=11");
+ Path octoberData = new Path(october, "data-1.csv");
+ Path novemberData = new Path(november, "data-2.csv");
+ fileIO.writeFile(octoberData, "1", false);
+ fileIO.writeFile(novemberData, "2", false);
+ Map<String, String> october2025 = new LinkedHashMap<>();
+ october2025.put("year", "2025");
+ october2025.put("month", "10");
+ FormatTableCommit commit =
+ truncatingCommit(tablePath, fileIO, false, null, "year",
"month");
+
+ commit.truncatePartitions(Collections.singletonList(october2025));
+
+ assertThat(fileIO.exists(octoberData)).isFalse();
+ assertThat(fileIO.exists(october)).isTrue();
+ assertThat(fileIO.exists(novemberData)).isTrue();
+ }
+
+ @Test
+ void testTruncatingAPrefixClearsThePartitionsBelowItButNotStagingTrees()
throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path staleFile = new Path(tablePath,
"year=2025/month=10/data-old.csv");
+ fileIO.writeFile(staleFile, "1", false);
+ // A job writing this prefix with the month left dynamic stages
exactly where the month
+ // directories sit, so a directory at a partition level is not
automatically partition data.
+ Path stagingFile =
+ new Path(tablePath,
"year=2025/_temporary/attempt/month=11/part-00011.csv");
+ fileIO.writeFile(stagingFile, "2", false);
+ Path otherYear = new Path(tablePath,
"year=2024/month=10/data-old.csv");
+ fileIO.writeFile(otherYear, "3", false);
+ FormatTableCommit commit =
+ truncatingCommit(tablePath, fileIO, false, null, "year",
"month");
+
+ commit.truncatePartitions(
+ Collections.singletonList(Collections.singletonMap("year",
"2025")));
+
+ assertThat(fileIO.exists(staleFile)).isFalse();
+ assertThat(fileIO.exists(stagingFile)).isTrue();
+ assertThat(fileIO.exists(otherYear)).isTrue();
+ }
+
+ @Test
+ void testTruncatingTheValueOnlyDefaultPartitionClearsIt() throws Exception
{
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ // Value-only layout: a null partition value is the default partition
name, which starts
+ // with '_' without being a staging directory. The scan reads it, so
TRUNCATE clears it.
+ Path defaultPartition = new Path(tablePath,
PARTITION_DEFAULT_NAME.defaultValue());
+ Path staleFile = new Path(defaultPartition, "data-old.csv");
+ fileIO.writeFile(staleFile, "1", false);
+ FormatTableCommit commit = truncatingCommit(tablePath, fileIO, true,
null, "year");
+
+ commit.truncatePartitions(
+ Collections.singletonList(
+ Collections.singletonMap("year",
PARTITION_DEFAULT_NAME.defaultValue())));
+
+ assertThat(fileIO.exists(staleFile)).isFalse();
+ assertThat(fileIO.exists(defaultPartition)).isTrue();
+ }
+
+ private static Map<String, String> partitionSpec(String year, String
month) {
+ LinkedHashMap<String, String> spec = new LinkedHashMap<>();
+ spec.put("year", year);
+ spec.put("month", month);
+ return spec;
+ }
+
+ /** The commit TRUNCATE makes: nothing to write, so no overwrite and no
static partition. */
+ private FormatTableCommit truncatingCommit(
+ Path tableLocation,
+ LocalFileIO fileIO,
+ boolean onlyValueInPath,
+ FormatTablePartitionManager partitionManager,
+ String... partitionKeys) {
+ return new FormatTableCommit(
+ tableLocation.toString(),
+ Arrays.asList(partitionKeys),
+ fileIO,
+ onlyValueInPath,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ Identifier.create("truncate_db", "truncate_table"),
+ null,
+ null,
+ null,
+ partitionManager);
+ }
+
private FormatTablePartitionManager commitPartitionedFile(
Path tableLocation, boolean onlyValueInPath, String partitionDir)
throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
diff --git
a/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
new file mode 100644
index 0000000000..f8f6f62cae
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
@@ -0,0 +1,35 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.table.FormatTable;
+
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.factories.DynamicTableFactory;
+
+/** Table sink for format tables. Flink 1.16 has no SupportsTruncate. */
+public class FlinkFormatTableSink extends FlinkFormatTableSinkBase {
+
+ public FlinkFormatTableSink(
+ ObjectIdentifier tableIdentifier,
+ FormatTable table,
+ DynamicTableFactory.Context context) {
+ super(tableIdentifier, table, context);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-1.17/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
b/paimon-flink/paimon-flink-1.17/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
new file mode 100644
index 0000000000..b10e4cd91e
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.17/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
@@ -0,0 +1,35 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.table.FormatTable;
+
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.factories.DynamicTableFactory;
+
+/** Table sink for format tables. Flink 1.17 has no SupportsTruncate. */
+public class FlinkFormatTableSink extends FlinkFormatTableSinkBase {
+
+ public FlinkFormatTableSink(
+ ObjectIdentifier tableIdentifier,
+ FormatTable table,
+ DynamicTableFactory.Context context) {
+ super(tableIdentifier, table, context);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
index 1c48602098..c31b723abf 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
@@ -18,80 +18,37 @@
package org.apache.paimon.flink.sink;
-import org.apache.paimon.flink.PaimonDataStreamSinkProvider;
import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.flink.table.catalog.ObjectIdentifier;
-import org.apache.flink.table.connector.ChangelogMode;
-import org.apache.flink.table.connector.sink.DynamicTableSink;
-import org.apache.flink.table.connector.sink.abilities.SupportsOverwrite;
-import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.connector.sink.abilities.SupportsTruncate;
import org.apache.flink.table.factories.DynamicTableFactory;
-import java.util.HashMap;
-import java.util.Map;
-
/** Table sink for format tables. */
-public class FlinkFormatTableSink
- implements DynamicTableSink, SupportsOverwrite, SupportsPartitioning {
-
- private final ObjectIdentifier tableIdentifier;
- private final FormatTable table;
- private final DynamicTableFactory.Context context;
- private Map<String, String> staticPartitions = new HashMap<>();
- protected boolean overwrite = false;
+public class FlinkFormatTableSink extends FlinkFormatTableSinkBase implements
SupportsTruncate {
public FlinkFormatTableSink(
ObjectIdentifier tableIdentifier,
FormatTable table,
DynamicTableFactory.Context context) {
- this.tableIdentifier = tableIdentifier;
- this.table = table;
- this.context = context;
- }
-
- @Override
- public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
- throw new UnsupportedOperationException("Format Table doesn't support
changelog mode.");
- }
-
- @Override
- public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
- return new PaimonDataStreamSinkProvider(
- (dataStream) ->
- new FlinkFormatTableDataStreamSink(table, overwrite,
staticPartitions)
- .sinkFrom(dataStream),
- tableIdentifier.asSummaryString(),
- table);
- }
-
- @Override
- public DynamicTableSink copy() {
- FlinkFormatTableSink copied = new
FlinkFormatTableSink(tableIdentifier, table, context);
- copied.staticPartitions = new HashMap<>(staticPartitions);
- copied.overwrite = overwrite;
- return copied;
- }
-
- @Override
- public String asSummaryString() {
- return "PaimonFormatTableSink";
- }
-
- @Override
- public void applyStaticPartition(Map<String, String> partition) {
- table.partitionKeys()
- .forEach(
- partitionKey -> {
- if (partition.containsKey(partitionKey)) {
- this.staticPartitions.put(
- partitionKey,
partition.get(partitionKey));
- }
- });
+ super(tableIdentifier, table, context);
}
+ /**
+ * Removes the data of the whole table - of its registered partitions,
when the catalog manages
+ * them. The partition directories stay, and so do their catalog
registrations: emptying a table
+ * does not redefine which partitions it has.
+ */
@Override
- public void applyOverwrite(boolean overwrite) {
- this.overwrite = overwrite;
+ public void executeTruncation() {
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.truncateTable();
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to truncate table %s.",
tableIdentifier.asSummaryString()),
+ e);
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
similarity index 91%
copy from
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
copy to
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
index 1c48602098..7add7f64c1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
@@ -32,16 +32,16 @@ import java.util.HashMap;
import java.util.Map;
/** Table sink for format tables. */
-public class FlinkFormatTableSink
+public abstract class FlinkFormatTableSinkBase
implements DynamicTableSink, SupportsOverwrite, SupportsPartitioning {
- private final ObjectIdentifier tableIdentifier;
- private final FormatTable table;
- private final DynamicTableFactory.Context context;
- private Map<String, String> staticPartitions = new HashMap<>();
+ protected final ObjectIdentifier tableIdentifier;
+ protected final FormatTable table;
+ protected final DynamicTableFactory.Context context;
+ protected Map<String, String> staticPartitions = new HashMap<>();
protected boolean overwrite = false;
- public FlinkFormatTableSink(
+ public FlinkFormatTableSinkBase(
ObjectIdentifier tableIdentifier,
FormatTable table,
DynamicTableFactory.Context context) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
index 43f47e433e..63da32b3fe 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
@@ -21,11 +21,14 @@ package org.apache.paimon.flink.source;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.flink.RESTCatalogITCaseBase;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.rest.RESTToken;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
@@ -42,19 +45,10 @@ public class FormatTableITCase extends
RESTCatalogITCaseBase {
Decimal decimal = Decimal.fromBigDecimal(new
BigDecimal(bigDecimalStr), 8, 3);
for (String format : new String[] {"parquet", "csv", "json"}) {
String tableName = "format_table_parquet_" + format.toLowerCase();
- Identifier identifier = Identifier.create("default", tableName);
sql(
"CREATE TABLE %s (a DECIMAL(8, 3), b INT, c INT) WITH
('file.format'='%s', 'type'='format-table')",
tableName, format);
- RESTToken expiredDataToken =
- new RESTToken(
- ImmutableMap.of(
- "akId",
- "akId-expire",
- "akSecret",
- UUID.randomUUID().toString()),
- System.currentTimeMillis() + 1000_000);
- restCatalogServer.setDataToken(identifier, expiredDataToken);
+ setDataToken(tableName);
sql("INSERT INTO %s VALUES (%s, 1, 1), (%s, 2, 2)", tableName,
decimal, decimal);
assertThat(sql("SELECT a, b FROM %s", tableName))
.containsExactlyInAnyOrder(
@@ -68,16 +62,10 @@ public class FormatTableITCase extends
RESTCatalogITCaseBase {
public void testPartitionedTableInsertOverwrite() {
String ptTableName = "format_table_overwrite";
- Identifier ptIdentifier = Identifier.create("default", ptTableName);
sql(
"CREATE TABLE %s (a DECIMAL(8, 3), b INT, c INT) PARTITIONED
BY (c) WITH ('file.format'='parquet', 'type'='format-table')",
ptTableName);
- RESTToken expiredDataToken =
- new RESTToken(
- ImmutableMap.of(
- "akId", "akId-expire", "akSecret",
UUID.randomUUID().toString()),
- System.currentTimeMillis() + 1000_000);
- restCatalogServer.setDataToken(ptIdentifier, expiredDataToken);
+ setDataToken(ptTableName);
String ptBigDecimalStr1 = "10.001";
String ptBigDecimalStr2 = "12.345";
@@ -125,16 +113,10 @@ public class FormatTableITCase extends
RESTCatalogITCaseBase {
Decimal decimal1 = Decimal.fromBigDecimal(new
BigDecimal(bigDecimalStr1), 8, 3);
Decimal decimal2 = Decimal.fromBigDecimal(new
BigDecimal(bigDecimalStr2), 8, 3);
- Identifier identifier = Identifier.create("default", tableName);
sql(
"CREATE TABLE %s (a DECIMAL(8, 3), b INT, c INT) WITH
('file.format'='parquet', 'type'='format-table')",
tableName);
- RESTToken expiredDataToken =
- new RESTToken(
- ImmutableMap.of(
- "akId", "akId-expire", "akSecret",
UUID.randomUUID().toString()),
- System.currentTimeMillis() + 1000_000);
- restCatalogServer.setDataToken(identifier, expiredDataToken);
+ setDataToken(tableName);
sql("INSERT INTO %s VALUES (%s, 1, 1), (%s, 2, 2)", tableName,
decimal1, decimal1);
assertThat(sql("SELECT a, b FROM %s", tableName))
@@ -150,4 +132,68 @@ public class FormatTableITCase extends
RESTCatalogITCaseBase {
sql("Drop TABLE %s", tableName);
}
+
+ @Test
+ public void testTruncateTable() {
+ String tableName = "format_table_truncate";
+ sql(
+ "CREATE TABLE %s (a INT, b INT) WITH ('file.format'='parquet',
'type'='format-table')",
+ tableName);
+ setDataToken(tableName);
+
+ sql("INSERT INTO %s VALUES (1, 11), (2, 22)", tableName);
+ assertThat(sql("SELECT * FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(1, 11), Row.of(2, 22));
+
+ assertThat(sql("TRUNCATE TABLE %s", tableName))
+ .containsExactly(Row.ofKind(RowKind.INSERT, "OK"));
+ assertThat(sql("SELECT * FROM %s", tableName)).isEmpty();
+
+ // The table is empty, not gone: it takes writes again.
+ sql("INSERT INTO %s VALUES (3, 33)", tableName);
+ assertThat(sql("SELECT * FROM %s",
tableName)).containsExactly(Row.of(3, 33));
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ @Test
+ public void testTruncatePartitionedTable() throws Exception {
+ String tableName = "format_table_truncate_partitioned";
+ sql(
+ "CREATE TABLE %s (a INT, b INT) PARTITIONED BY (b) WITH
('file.format'='parquet', 'type'='format-table')",
+ tableName);
+ setDataToken(tableName);
+
+ sql("INSERT INTO %s PARTITION (b = 1) VALUES (10)", tableName);
+ sql("INSERT INTO %s PARTITION (b = 2) VALUES (20)", tableName);
+ assertThat(sql("SELECT a, b FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(10, 1), Row.of(20, 2));
+
+ sql("TRUNCATE TABLE %s", tableName);
+ assertThat(sql("SELECT a, b FROM %s", tableName)).isEmpty();
+
+ // Emptied, not dropped: the partition directory of the one nothing
was written back to
+ // is still there, and all that is left in it is a writer's staging
tree.
+ Path emptied = new Path(dataPath, "default.db/" + tableName + "/b=2");
+ LocalFileIO fileIO = LocalFileIO.create();
+ assertThat(fileIO.exists(emptied)).isTrue();
+ assertThat(fileIO.listStatus(emptied))
+ .extracting(status -> status.getPath().getName())
+ .allMatch(name -> name.startsWith("_"));
+
+ // Every partition was emptied, and each of them takes writes again.
+ sql("INSERT INTO %s PARTITION (b = 1) VALUES (100)", tableName);
+ assertThat(sql("SELECT a, b FROM %s",
tableName)).containsExactly(Row.of(100, 1));
+
+ sql("DROP TABLE %s", tableName);
+ }
+
+ private void setDataToken(String tableName) {
+ restCatalogServer.setDataToken(
+ Identifier.create("default", tableName),
+ new RESTToken(
+ ImmutableMap.of(
+ "akId", "akId-expire", "akSecret",
UUID.randomUUID().toString()),
+ System.currentTimeMillis() + 1000_000));
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
index 7bf9578989..392b127ff1 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
@@ -58,10 +58,10 @@ trait PaimonPartitionManagement extends
SupportsAtomicPartitionManagement with L
toPaimonPartition(r, partitionKeys.take(r.numFields))
}
case _: FormatTable =>
- // Reached by the partition operations this trait still serves
directly, such as TRUNCATE
- // PARTITION. Saying that only a FileStoreTable has partitions would
be wrong for a Format
- // Table with catalog-managed partitions, which has them and lists
them here; a Format
- // Table is still a Paimon table, just not a native one.
+ // Reached by the partition operations this trait still serves
directly. Saying that only
+ // a FileStoreTable has partitions would be wrong for a Format Table
with catalog-managed
+ // partitions, which has them and lists them here; a Format Table is
still a Paimon table,
+ // just not a native one.
throw new UnsupportedOperationException(
s"This partition operation is supported only for a native Paimon
table; " +
s"${table.name()} is a Format Table, which manages its partitions
through " +
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 6b5e5210f9..131fc4e2b3 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
@@ -25,11 +25,13 @@ import org.apache.paimon.spark.{BaseTable,
FormatTableScanBuilder}
import org.apache.paimon.spark.write.{BaseV2WriteBuilder,
PaimonWriteRequirement}
import org.apache.paimon.table.FormatTable
import org.apache.paimon.table.format.FormatTablePartitionManager
+import org.apache.paimon.table.sink.BatchTableCommit
import org.apache.paimon.types.RowType
import org.apache.paimon.utils.PartitionPathUtils
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
TableCapability, TableCatalog}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException,
NoSuchPartitionsException}
+import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
TableCapability, TableCatalog, TruncatableTable}
import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ,
BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
import org.apache.spark.sql.connector.distributions.Distribution
import org.apache.spark.sql.connector.expressions.SortOrder
@@ -49,7 +51,8 @@ import scala.collection.mutable.{ArrayBuffer, HashSet}
case class PaimonFormatTable(table: FormatTable)
extends BaseTable
with SupportsRead
- with SupportsWrite {
+ with SupportsWrite
+ with TruncatableTable {
// A Format Table uses catalog-managed partitions exactly when the catalog
gave it a partition
// manager; tables using filesystem partition discovery return null and rely
on the directory
@@ -92,6 +95,90 @@ case class PaimonFormatTable(table: FormatTable)
PaimonFormatTableWriterBuilder(table, info.schema)
}
+ /**
+ * Removes the data of the whole table - of its registered partitions, when
the catalog manages
+ * them. The partition directories stay, and so do their catalog
registrations: emptying a table
+ * does not redefine which partitions it has (SPARK-34418).
+ */
+ override def truncateTable(): Boolean = {
+ withCommit(_.truncateTable())
+ true
+ }
+
+ override def truncatePartitions(idents: Array[InternalRow]): Boolean = {
+ truncateFormatTablePartitions(
+ idents,
+ missing => new NoSuchPartitionsException(name(), missing.toSeq,
partitionSchema))
+ }
+
+ override def truncatePartition(ident: InternalRow): Boolean = {
+ truncateFormatTablePartitions(
+ Array(ident),
+ missing => new NoSuchPartitionException(name(), missing.head,
partitionSchema))
+ }
+
+ /**
+ * Removes the data of the given partitions, keeping the partitions
themselves (see
+ * [[truncateTable]]). Spark resolves a partial spec to the partitions it
covers before calling
+ * either entry point, so every spec arriving here is complete.
+ *
+ * A partition the table does not have cannot be truncated, and each entry
point reports that the
+ * way Spark expects it to. What the table has is answered by the catalog
for catalog-managed
+ * partitions and by the directory for filesystem partition discovery - each
kind is asked the
+ * same source it reads its partitions from, so data merely awaiting
registration is never emptied
+ * behind MSCK REPAIR TABLE's back.
+ *
+ * The partitions are truncated one after another. A Format Table has no
snapshot to make that
+ * atomic, so a failure part-way leaves the partitions handled before it
empty - as a failing
+ * `INSERT OVERWRITE` of several partitions does.
+ */
+ private def truncateFormatTablePartitions(
+ idents: Array[InternalRow],
+ noSuchPartitions: Array[InternalRow] => Throwable): Boolean = {
+ if (idents.isEmpty) {
+ return true
+ }
+ val partitionKeys = table.partitionKeys().asScala.toSeq
+ val specs = idents.map {
+ ident =>
+ require(
+ ident.numFields == partitionKeys.size,
+ s"Truncating a partition of Format Table ${table.fullName()} needs a
complete spec " +
+ s"for partition keys ${partitionKeys.mkString("[", ", ", "]")}, " +
+ s"but got ${ident.numFields} values."
+ )
+ toPaimonPartition(ident, partitionKeys)
+ }
+ val onlyValueInPath =
+
CoreOptions.fromMap(table.options()).formatTablePartitionOnlyValueInPath()
+ // Resolve (and path-safety validate) every directory before deleting
anything, as ADD and DROP
+ // PARTITION do.
+ val partitionPaths =
+ specs.map(spec => resolvePartitionPathWithinTable(orderedSpec(spec),
onlyValueInPath))
+ val exists = if (hasCatalogManagedPartitions) {
+ val partitionNames = partitionKeys.toArray
+ formatTablePartitionsRegistered(idents.map(_ => partitionNames), idents)
+ } else {
+ val fileIO = table.fileIO()
+ partitionPaths.map(fileIO.exists)
+ }
+ val missing = idents.zip(exists).collect { case (ident, false) => ident }
+ if (missing.nonEmpty) {
+ throw noSuchPartitions(missing)
+ }
+ withCommit(_.truncatePartitions(specs.toSeq.asJava))
+ true
+ }
+
+ private def withCommit(operation: BatchTableCommit => Unit): Unit = {
+ val commit = table.newBatchWriteBuilder().newCommit()
+ try {
+ operation(commit)
+ } finally {
+ commit.close()
+ }
+ }
+
/**
* Resolves, with a single catalog list-by-names lookup, which of the given
complete partition
* specs are registered. The result is aligned with the input arrays.
@@ -233,8 +320,9 @@ case class PaimonFormatTable(table: FormatTable)
/**
* Build the partition directory for a spec and verify it stays strictly
under the table location.
* Value-only path components are validated (including rejecting '.'/'..'),
and the normalized
- * path is checked against the table location so neither DROP (recursive
delete) nor ADD (mkdirs)
- * can escape the table directory via crafted or corrupt partition values.
+ * path is checked against the table location so no DROP (recursive delete),
ADD (mkdirs) or
+ * TRUNCATE (delete of the files below it) can escape the table directory
via crafted or corrupt
+ * partition values.
*/
private def resolvePartitionPathWithinTable(
orderedSpec: util.LinkedHashMap[String, String],
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala
index 191fe2cb46..103aae3aea 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionTest.scala
@@ -45,6 +45,29 @@ class CatalogManagedPartitionTest extends
PaimonSparkTestWithRestCatalogBase {
}
}
+ test("TRUNCATE on a catalog-managed Format Table empties data and keeps the
registrations") {
+ val tableName = "catalog_partition_truncate"
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, dt STRING)
+ |USING CSV
+ |PARTITIONED BY (dt)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $tableName VALUES (1, '20260715'), (2, '20260716')")
+
+ sql(s"TRUNCATE TABLE $tableName PARTITION (dt = '20260715')")
+ checkAnswer(sql(s"SELECT id FROM $tableName"), Seq(Row(2)))
+ checkAnswer(sql(s"SHOW PARTITIONS $tableName"), Seq(Row("dt=20260715"),
Row("dt=20260716")))
+
+ sql(s"TRUNCATE TABLE $tableName")
+ checkAnswer(sql(s"SELECT id FROM $tableName"), Seq.empty)
+ // The registrations outlive the data they pointed at: TRUNCATE is not
DROP PARTITION.
+ checkAnswer(sql(s"SHOW PARTITIONS $tableName"), Seq(Row("dt=20260715"),
Row("dt=20260716")))
+ }
+ }
+
test("catalog-managed Format Table sends partition predicate to REST
filtered listing") {
val tableName = "catalog_partition_filter_pushdown"
withTable(tableName) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
index 8febee0e0a..0488594fcc 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
@@ -30,6 +30,7 @@ import org.apache.paimon.types.DataTypes
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException,
NoSuchPartitionsException}
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
import org.apache.spark.unsafe.types.UTF8String
@@ -233,6 +234,7 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
// partition stays registered; re-running ADD ... IF NOT EXISTS then
creates the directory.
assert(error eq failure)
assert(gateway.partitions == Seq(Map("dt" -> "20260715", "hh" -> "10")))
+ assert(gateway.listCalls == 0)
} finally {
delegate.delete(tablePath, true)
}
@@ -585,6 +587,7 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
// call already created: resolving duplicates is the catalog's job, not
Spark's.
assert(gateway.createRequests == Seq((Seq(first, second), true),
(Seq(second, third), true)))
assert(gateway.lookupCalls == 0)
+ assert(gateway.listCalls == 0)
}
test("catalog-managed strict ADD asks the catalog to reject duplicates,
without looking first") {
@@ -601,9 +604,10 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
}
assert(error.partition == existing)
- // Rejecting the batch is the catalog's decision; Spark must not pre-empt
it with a lookup,
- // which is what would break the batch's atomicity.
+ // Rejecting the batch is the catalog's decision; Spark must not pre-empt
it with a lookup
+ // or an enumeration, which is what would break the batch's atomicity.
assert(gateway.lookupCalls == 0)
+ assert(gateway.listCalls == 0)
}
test("catalog-managed ADD preserves typed special partition values") {
@@ -623,6 +627,100 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
assert(
gateway.createRequests == Seq(
(values.map(value => Map("dt" -> value)) :+ Map("dt" ->
"__DEFAULT_PARTITION__"), true)))
+ assert(gateway.listCalls == 0)
+ }
+
+ test("catalog-managed TRUNCATE PARTITION empties the partition without
unregistering it") {
+ val fileIO = LocalFileIO.create()
+ val tablePath = new
Path(Files.createTempDirectory("catalog-partition-format-truncate").toUri)
+ val registered = Seq(partitionSpec(20260715, 10), partitionSpec(20260716,
11))
+ val gateway = new InMemoryPartitionManager(registered)
+ val table = formatTableWithCatalogManagedPartitions(fileIO,
tablePath.toString, gateway)
+ try {
+ val truncated = new Path(tablePath, "dt=20260715/hh=10")
+ val kept = new Path(tablePath, "dt=20260716/hh=11")
+ val truncatedData = new Path(truncated, "data.csv")
+ val keptData = new Path(kept, "data.csv")
+ fileIO.writeFile(truncatedData, "1,20260715,10", false)
+ fileIO.writeFile(keptData, "2,20260716,11", false)
+
+ assert(new
PaimonFormatTable(table).truncatePartition(partitionRow(20260715, 10)))
+
+ assert(!fileIO.exists(truncatedData))
+ // The partition survives being emptied: its directory stays and so does
its registration,
+ // which is what makes TRUNCATE different from DROP PARTITION.
+ assert(fileIO.exists(truncated))
+ assert(gateway.partitions == registered)
+ assert(fileIO.exists(keptData))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed TRUNCATE PARTITION refuses a partition the catalog
does not know") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new
Path(Files.createTempDirectory("catalog-partition-format-truncate-unknown").toUri)
+ val gateway = new InMemoryPartitionManager
+ val table = formatTableWithCatalogManagedPartitions(fileIO,
tablePath.toString, gateway)
+ try {
+ val awaitingRepair = new Path(tablePath, "dt=20260715/hh=10")
+ val data = new Path(awaitingRepair, "data.csv")
+ fileIO.writeFile(data, "1,20260715,10", false)
+
+ intercept[NoSuchPartitionException] {
+ new PaimonFormatTable(table).truncatePartition(partitionRow(20260715,
10))
+ }
+ intercept[NoSuchPartitionsException] {
+ new
PaimonFormatTable(table).truncatePartitions(Array(partitionRow(20260715, 10)))
+ }
+
+ // Data that MSCK REPAIR TABLE has yet to register is not this table's
to empty.
+ assert(fileIO.exists(data))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed TRUNCATE TABLE empties the registered partitions and
nothing else") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new
Path(Files.createTempDirectory("catalog-partition-format-truncate-table").toUri)
+ val registered = Seq(partitionSpec(20260715, 10))
+ val gateway = new InMemoryPartitionManager(registered)
+ val table = formatTableWithCatalogManagedPartitions(fileIO,
tablePath.toString, gateway)
+ try {
+ val registeredData = new Path(new Path(tablePath, "dt=20260715/hh=10"),
"data.csv")
+ val awaitingRepairData = new Path(new Path(tablePath,
"dt=20260716/hh=11"), "data.csv")
+ fileIO.writeFile(registeredData, "1,20260715,10", false)
+ fileIO.writeFile(awaitingRepairData, "2,20260716,11", false)
+
+ assert(new PaimonFormatTable(table).truncateTable())
+
+ assert(!fileIO.exists(registeredData))
+ // The catalog says which partitions this table has, and the table reads
only those. A
+ // directory MSCK REPAIR TABLE has yet to register is not the table's to
empty.
+ assert(fileIO.exists(awaitingRepairData))
+ assert(gateway.partitions == registered)
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed TRUNCATE PARTITION rejects a partition value that
would escape the table") {
+ val gateway = new InMemoryPartitionManager(Seq(Map("dt" -> "..")))
+ val sparkTable =
+ new PaimonFormatTable(
+ stringFormatTableWithCatalogManagedPartitions(gateway, onlyValueInPath
= true))
+
+ val error = intercept[IllegalArgumentException] {
+ sparkTable.truncatePartition(new
GenericInternalRow(Array[Any](UTF8String.fromString(".."))))
+ }
+
+ // The value is rejected while the directory is resolved, before the
catalog is asked whether
+ // the partition exists and so before anything is deleted.
+ assert(error.getMessage.contains(".."))
+ assert(gateway.lookupCalls == 0)
}
private def emptyGateway: FormatTablePartitionManager =
@@ -651,6 +749,7 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
private val storedPartitions = mutable.LinkedHashSet(initialPartitions: _*)
private var requests = Seq.empty[(Seq[Map[String, String]], Boolean)]
var lookupCalls = 0
+ var listCalls = 0
def createRequests: Seq[(Seq[Map[String, String]], Boolean)] =
synchronized(requests)
@@ -683,7 +782,12 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
}
override def listPartitions(prefix: JMap[String, String], filter:
Predicate): JList[Partition] =
- throw new AssertionError("ADD must not enumerate catalog or filesystem
partitions")
+ synchronized {
+ listCalls += 1
+ val required = prefix.asScala.toSet
+ registeredPartitions(
+ storedPartitions.filter(spec =>
required.subsetOf(spec.toSet)).toSeq: _*)
+ }
}
private class TestPartitionAlreadyExistsException(val partition: Map[String,
String])
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
index bee92ed562..58c9537d7a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
@@ -448,6 +448,39 @@ class CatalogManagedPartitionMsckRepairTest extends
PaimonSparkTestWithRestCatal
}
}
+ test("a measuring MSCK keeps the exact zero a TRUNCATE reported") {
+ val tableName = "msck_statistics_after_truncate"
+ val partition = "20260722"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ writeCsvPartition(tableName, partition, 22, "to-be-truncated")
+
+ val collectStatistics =
+
s"spark.paimon.${SparkConnectorOptions.FORMAT_TABLE_REPAIR_COLLECT_STATISTICS.key()}"
+ withSQLConf(collectStatistics -> "true") {
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE
paimon.$dbName0.$tableName")
+ }
+ assert(statisticsOf(tableName, partition).fileCount() == 1L)
+
+ sql(s"TRUNCATE TABLE paimon.$dbName0.$tableName PARTITION (dt =
'$partition')")
+ val truncated = statisticsOf(tableName, partition)
+ assert(truncated.fileCount() == 0L, truncated.toString)
+ assert(truncated.recordCount() == 0L, truncated.toString)
+
+ withSQLConf(collectStatistics -> "true") {
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE
paimon.$dbName0.$tableName")
+ }
+
+ // Measuring an empty partition confirms it is empty; it does not
un-measure it.
+ val remeasured = statisticsOf(tableName, partition)
+ assert(remeasured.fileCount() == 0L, remeasured.toString)
+ assert(remeasured.fileSizeInBytes() == 0L, remeasured.toString)
+ assert(remeasured.recordCount() == 0L, remeasured.toString)
+ assertPartitionState(tableName, Set(partition))
+ }
+ }
+
private def statisticsOf(tableName: String, partition: String): Partition =
paimonCatalog
.listPartitions(tableIdentifier(tableName))
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 652fb9be4d..dc523284bb 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
@@ -28,6 +28,7 @@ import org.apache.paimon.table.source.Split
import org.apache.paimon.utils.{CompressUtils, PartitionPathUtils}
import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException
import org.apache.spark.sql.connector.read.InputPartition
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -144,6 +145,56 @@ abstract class FormatTableTestBase extends
PaimonHiveTestBase with AdaptiveSpark
}
}
+ test("Format table: truncate table") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, p1 INT, p2 STRING) USING csv PARTITIONED BY
(p1, p2)")
+ sql("INSERT INTO t VALUES (1, 1, '1'), (2, 2, '1'), (3, 2, '2')")
+
+ sql("TRUNCATE TABLE t")
+
+ checkAnswer(sql("SELECT * FROM t"), Seq.empty)
+ // Emptying a table does not redefine which partitions it has
(SPARK-34418). Here they are
+ // the directories, and those stay.
+ checkAnswer(
+ sql("SHOW PARTITIONS t"),
+ Seq(Row("p1=1/p2=1"), Row("p1=2/p2=1"), Row("p1=2/p2=2")))
+ }
+ }
+
+ test("Format table: truncate partition") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, p1 INT, p2 STRING) USING csv PARTITIONED BY
(p1, p2)")
+ sql("INSERT INTO t VALUES (1, 1, '1'), (2, 2, '1'), (3, 2, '2')")
+
+ sql("TRUNCATE TABLE t PARTITION (p1 = 2, p2 = '2')")
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, 1, "1"), Row(2, 2, "1")))
+
+ // A partial spec truncates the partitions it covers.
+ sql("TRUNCATE TABLE t PARTITION (p1 = 2)")
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, 1, "1")))
+
+ checkAnswer(
+ sql("SHOW PARTITIONS t"),
+ Seq(Row("p1=1/p2=1"), Row("p1=2/p2=1"), Row("p1=2/p2=2")))
+ }
+ }
+
+ test("Format table: truncate a partition the table does not have") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, p1 INT, p2 STRING) USING csv PARTITIONED BY
(p1, p2)")
+ sql("INSERT INTO t VALUES (1, 1, '1')")
+
+ // A complete spec matching nothing is an error, as for any other Spark
table; a partial one
+ // matching nothing does nothing.
+ intercept[NoSuchPartitionException] {
+ sql("TRUNCATE TABLE t PARTITION (p1 = 9, p2 = '9')")
+ }
+ sql("TRUNCATE TABLE t PARTITION (p1 = 9)")
+
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, 1, "1")))
+ }
+ }
+
test("Format table: CTAS with partitioned table") {
withTable("t1", "t2") {
sql("CREATE TABLE t1 (id INT, p1 INT, p2 INT) USING csv PARTITIONED BY
(p1, p2)")
