[core][spark][flink] Support TRUNCATE TABLE on format tables

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.

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. 

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java

@@ -30,11 +30,13 @@
 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.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 void commit(List<CommitMessage> commitMessages) {
             }
             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 void commit(List<CommitMessage> commitMessages) {
      * 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 @@ private void reportPartitions(
         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 @@ private static Path buildPartitionPath(
         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,29 +449,182 @@ private Set<Path> deletePreviousDataFile(Path partitionPath, int partitionLevels
                             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;
     }
 
     @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

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);