From 0b254fb5d29d71e896708690c248286548c79a4c Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Fri, 21 Aug 2026 00:56:53 +0800 Subject: [PATCH] [core][spark][flink] Support TRUNCATE TABLE on format tables TRUNCATE TABLE did not work on a Format Table. TRUNCATE TABLE t was rejected in the planner because PaimonFormatTable does not implement TruncatableTable, and TRUNCATE TABLE t PARTITION (...) reached PaimonPartitionManagement, which serves only FileStoreTable. Both FormatTableCommit entry points threw an empty UnsupportedOperationException, and FlinkFormatTableSink did not implement SupportsTruncate either. FormatTableCommit now removes the data files, reusing deletePreviousDataFile, the primitive a static INSERT OVERWRITE already clears partition directories with. Only data files go: the partition directories stay, and so do their catalog registrations, so SHOW PARTITIONS returns what it returned before (SPARK-34418). Staging trees of concurrent writers are left alone, on the same judgement FormatTableScan reads with. Which partitions the table has is answered by whatever the table reads its partitions from: the catalog when it manages them, the partition directories the scan parses otherwise. So truncating neither empties nor registers a directory still waiting for MSCK REPAIR TABLE, and never deletes a file the table cannot read. What the emptied partitions hold is reported to the catalog as an exact zero, carried with replaceStatistics, so a truncated partition stops describing files that are gone. An overwrite reports only the files it removed itself, so that concurrent writers do not each claim the whole subtree; truncation states that the partition holds nothing whoever deleted the files, so one that was already empty reports zero as well. A Format Table has no snapshot to make the whole truncation atomic, so a failure part-way reports what it emptied before propagating. SupportsTruncate arrived in Flink 1.18, so FlinkFormatTableSink is split the way FlinkTableSink already is and 1.16 and 1.17 get one that does not implement it. Flink has no TRUNCATE TABLE ... PARTITION, so only whole-table truncation is wired there. --- docs/docs/flink/sql-write.mdx | 7 + docs/docs/spark/sql-write.md | 10 + .../table/format/FormatTableCommit.java | 204 +++++++++++-- .../paimon/table/sink/BatchTableCommit.java | 6 +- .../FormatTableCommitStatisticsTest.java | 281 ++++++++++++++++++ .../table/format/FormatTableCommitTest.java | 211 +++++++++++++ .../flink/sink/FlinkFormatTableSink.java | 35 +++ .../flink/sink/FlinkFormatTableSink.java | 35 +++ .../flink/sink/FlinkFormatTableSink.java | 79 ++--- .../flink/sink/FlinkFormatTableSinkBase.java | 97 ++++++ .../flink/source/FormatTableITCase.java | 94 ++++-- .../spark/PaimonPartitionManagement.scala | 8 +- .../spark/format/PaimonFormatTable.scala | 96 +++++- .../format/CatalogManagedPartitionTest.scala | 23 ++ .../FormatTablePartitionManagementTest.scala | 110 ++++++- ...atalogManagedPartitionMsckRepairTest.scala | 33 ++ .../spark/sql/FormatTableTestBase.scala | 51 ++++ 17 files changed, 1260 insertions(+), 120 deletions(-) create mode 100644 paimon-flink/paimon-flink-1.16/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java create mode 100644 paimon-flink/paimon-flink-1.17/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSink.java create mode 100644 paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java diff --git a/docs/docs/flink/sql-write.mdx b/docs/docs/flink/sql-write.mdx index 8f1d8e3cfba7..aea1ba870c9b 100644 --- a/docs/docs/flink/sql-write.mdx +++ b/docs/docs/flink/sql-write.mdx @@ -142,6 +142,13 @@ TRUNCATE TABLE my_table; +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 1cc6c0bc691f..5ac34f083499 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 ebf5c39c5034..aec5f73e15d0 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.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 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 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> writtenPartitionSpecs, Map, PartitionStatistics> statisticsByPartition, Set clearedPartitionPaths, - long commitTime) { + long commitTime, + boolean replaceStatistics) { for (Path cleared : clearedPartitionPaths) { Map 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 orderedSpec = new LinkedHashMap<>(); for (int i = 0; i < partitionSpec.size(); i++) { String key = partitionKeys.get(i); @@ -442,16 +449,26 @@ private Set 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; @@ -459,12 +476,155 @@ private Set deletePreviousDataFile(Path partitionPath, int partitionLevels @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, 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> 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> complete = new ArrayList<>(); + for (Map partitionSpec : partitionSpecs) { + if (partitionSpec.size() == partitionKeys.size()) { + complete.add(partitionSpec); + } + } + Set> registered = + complete.isEmpty() + ? Collections.emptySet() + : partitionManager.listPartitionsByNames(complete).stream() + .map(Partition::spec) + .collect(Collectors.toSet()); + List> partitions = new ArrayList<>(); + for (Map 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> registeredPartitions(Map prefix) { + return partitionManager.listPartitions(prefix, null).stream() + .map(Partition::spec) + .collect(Collectors.toList()); + } + + private void truncate(List> partitionSpecs) { + long truncateTime = System.currentTimeMillis(); + Set 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, PartitionStatistics> emptied = new LinkedHashMap<>(); + RuntimeException failure = null; + for (Map 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 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 2cfc181eada2..52e11e3c84d0 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> 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 ed26d7074261..199b7b446b81 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.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 @@ void testStaticPrefixOverwriteZeroesAClearedPartitionAndKeepsItRegistered() thro }); } + @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 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 @@ private static void writeDataFile( } } + /** A partition the catalog holds, whose statistics the truncation is meant to replace. */ + private static Partition partition(Map 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... specs) { + List partitions = new ArrayList<>(); + for (Map 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> asked = invocation.getArgument(0); + List found = new ArrayList<>(); + for (Partition partition : partitions) { + if (asked.contains(partition.spec())) { + found.add(partition); + } + } + return found; + }); + } + private static Map spec(String year, String month) { LinkedHashMap spec = new LinkedHashMap<>(); spec.put("year", year); @@ -446,6 +696,37 @@ private static Reported capture(FormatTablePartitionManager partitionManager) { * 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 6d3d6da6f236..db1212981486 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.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.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 @@ void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception { 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 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 partitionSpec(String year, String month) { + LinkedHashMap 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 000000000000..f8f6f62cae97 --- /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 000000000000..b10e4cd91e60 --- /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 1c48602098e3..c31b723abff2 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 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 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/FlinkFormatTableSinkBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java new file mode 100644 index 000000000000..7add7f64c183 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java @@ -0,0 +1,97 @@ +/* + * 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.flink.PaimonDataStreamSinkProvider; +import org.apache.paimon.table.FormatTable; + +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.factories.DynamicTableFactory; + +import java.util.HashMap; +import java.util.Map; + +/** Table sink for format tables. */ +public abstract class FlinkFormatTableSinkBase + implements DynamicTableSink, SupportsOverwrite, SupportsPartitioning { + + protected final ObjectIdentifier tableIdentifier; + protected final FormatTable table; + protected final DynamicTableFactory.Context context; + protected Map staticPartitions = new HashMap<>(); + protected boolean overwrite = false; + + public FlinkFormatTableSinkBase( + 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 partition) { + table.partitionKeys() + .forEach( + partitionKey -> { + if (partition.containsKey(partitionKey)) { + this.staticPartitions.put( + partitionKey, partition.get(partitionKey)); + } + }); + } + + @Override + public void applyOverwrite(boolean overwrite) { + this.overwrite = overwrite; + } +} 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 43f47e433e9f..63da32b3fe4f 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 @@ 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 void testDiffFormat() { 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 void testDiffFormat() { 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 void testUnPartitionedTableInsertOverwrite() { 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 void testUnPartitionedTableInsertOverwrite() { 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 7bf95789899c..392b127ff1b2 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 6b5e5210f98a..131fc4e2b360 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 191fe2cb46b4..103aae3aeace 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 8febee0e0a0b..0488594fcc93 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 bee92ed56226..58c9537d7a90 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 652fb9be4db9..dc523284bbba 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)")