From 6cb6319952e024c0380d31c5984303546700cf8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Sun, 9 Aug 2026 01:57:08 +0800 Subject: [PATCH] [core] Support reporting partition statistics from format table commits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The writer already counted the rows and the bytes, the commit already knows which partitions it wrote, and the catalog now has somewhere to put both. This connects them, behind format-table.commit.report-partition-statistics. Whether a commit replaces or adds follows from what it did to the partition. An appending commit saw only its own files, so it reports an increment: a Flink sink commits once per writer subtask, and N increments over one partition add up to what the job wrote. An overwriting commit replaced everything the partitions held, so what it wrote is the total and it reports that. Static prefix overwrite is why a pure increment cannot express this. Clearing a prefix empties every partition beneath it, including ones this commit writes nothing to; their old data is gone and no increment says so. Those partitions report zero — an exact zero, they really are empty — dated to this commit, since a time reported as unknown would leave the stored one describing files that are gone. They ride in the same create request as the written ones, since a statistic can only be reported for a partition its own request registers. Nothing here unregisters a partition, whatever the numbers say. Only the files this commit actually deleted count as emptying a partition: one another writer removed first is not this commit's doing, and counting it as such would have every concurrent writer report the whole subtree. What a commit wrote into a partition is folded the way PartitionEntry already folds the same five fields: an immutable PartitionStatistics per file, merged into a map by spec. A count nobody took leaves that field unknown for the whole partition rather than reporting the sum of the rest as exact. Reporting is unconditional, the way a Paimon table's commit reports through commitSnapshot. An increment can drift, from a job that retries or a writer that is not Paimon, and a field cannot be written back to unknown once it is set; what converges it is a later full measurement over the same partition. Tests: FormatTableCommitStatisticsTest covers append, dynamic overwrite, static prefix overwrite of a partition this commit does not write, a directory that is no partition of this table being left alone rather than failing the commit, the summation of the independent increments of concurrent writers of one partition, a listing that answers under another scheme, an uncounted file between two counted ones so that unknown has to stay unknown, a report the catalog refuses taking the commit down with it and the written file with it, the same for an overwrite that has already deleted what the partition held, and the numbers reaching the catalog through the write builder. --- .../table/format/FormatTableCommit.java | 195 ++++- .../FormatTableCommitStatisticsTest.java | 676 ++++++++++++++++++ .../table/format/FormatTableCommitTest.java | 13 +- 3 files changed, 862 insertions(+), 22 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java 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 152baa3ba010..ebf5c39c5034 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,12 +30,16 @@ import org.apache.paimon.metrics.MetricRegistry; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.Options; +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.PartitionPathUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import javax.annotation.Nullable; import java.io.FileNotFoundException; @@ -45,6 +49,7 @@ import java.util.Collections; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -54,6 +59,8 @@ /** Commit for Format Table. */ public class FormatTableCommit implements BatchTableCommit { + private static final Logger LOG = LoggerFactory.getLogger(FormatTableCommit.class); + private String location; private final boolean formatTablePartitionOnlyValueInPath; private final String defaultPartName; @@ -107,10 +114,13 @@ public FormatTableCommit( @Override public void commit(List commitMessages) { try { - List committers = new ArrayList<>(); + // One reading for the whole commit: stat-ing each file back costs a request per + // file for a coarser number. + long commitTime = System.currentTimeMillis(); + List messages = new ArrayList<>(); for (CommitMessage commitMessage : commitMessages) { if (commitMessage instanceof TwoPhaseCommitMessage) { - committers.add(((TwoPhaseCommitMessage) commitMessage).getCommitter()); + messages.add((TwoPhaseCommitMessage) commitMessage); } else { throw new RuntimeException( "Unsupported commit message type: " @@ -119,6 +129,7 @@ public void commit(List commitMessages) { } Set> partitionSpecs = new HashSet<>(); + Set clearedPartitionPaths = new HashSet<>(); if (staticPartitions != null && !staticPartitions.isEmpty()) { Path partitionPath = @@ -133,39 +144,63 @@ public void commit(List commitMessages) { if (overwrite) { // A static partition may name only the leading keys, in which case the path // is a prefix and the partition directories of the remaining keys sit below. - deletePreviousDataFile( - partitionPath, partitionKeys.size() - staticPartitions.size()); + clearedPartitionPaths.addAll( + deletePreviousDataFile( + partitionPath, partitionKeys.size() - staticPartitions.size())); } if (!fileIO.exists(partitionPath)) { fileIO.mkdirs(partitionPath); } } else if (overwrite) { Set partitionPaths = new HashSet<>(); - for (TwoPhaseOutputStream.Committer c : committers) { - partitionPaths.add(c.targetPath().getParent()); + for (TwoPhaseCommitMessage message : messages) { + partitionPaths.add(message.getCommitter().targetPath().getParent()); } for (Path p : partitionPaths) { // The parent of a written file is a complete partition directory - the table // directory itself when the table is unpartitioned - so there is no partition - // level below it to descend. + // level below it to descend, and it is a partition this commit writes anyway. deletePreviousDataFile(p, 0); } } - for (TwoPhaseOutputStream.Committer committer : committers) { + boolean registersPartitions = + partitionKeys != null + && !partitionKeys.isEmpty() + && (hiveCatalog != null || partitionManager != null); + boolean reportsStatistics = registersPartitions && partitionManager != null; + Map, PartitionStatistics> statisticsByPartition = + new LinkedHashMap<>(); + for (TwoPhaseCommitMessage message : messages) { + TwoPhaseOutputStream.Committer committer = message.getCommitter(); committer.commit(this.fileIO); - if (partitionKeys != null - && !partitionKeys.isEmpty() - && (hiveCatalog != null || partitionManager != null)) { - partitionSpecs.add( + if (registersPartitions) { + // Extracted once: registration and statistics must key on the same spec. + Map spec = extractPartitionSpecFromPath( - committer.targetPath().getParent(), partitionKeys)); + committer.targetPath().getParent(), partitionKeys); + partitionSpecs.add(spec); + if (reportsStatistics) { + statisticsByPartition.merge( + spec, + new PartitionStatistics( + spec, + message.recordCount(), + message.fileSizeInBytes(), + 1, + commitTime, + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS), + FormatTableCommit::sum); + } } } - for (TwoPhaseOutputStream.Committer committer : committers) { - committer.clean(this.fileIO); + for (TwoPhaseCommitMessage message : messages) { + message.getCommitter().clean(this.fileIO); } - if (partitionManager != null && !partitionSpecs.isEmpty()) { + if (reportsStatistics) { + reportPartitions( + partitionSpecs, statisticsByPartition, clearedPartitionPaths, commitTime); + } 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. partitionManager.createPartitions(new ArrayList<>(partitionSpecs), true); @@ -195,6 +230,120 @@ 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. + */ + private void reportPartitions( + Set> writtenPartitionSpecs, + Map, PartitionStatistics> statisticsByPartition, + Set clearedPartitionPaths, + long commitTime) { + 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)); + } + } + + // Statistics are matched by spec, not by position: the specs need only be a superset. + Set> specs = new LinkedHashSet<>(writtenPartitionSpecs); + specs.addAll(statisticsByPartition.keySet()); + 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. + partitionManager.createPartitions( + new ArrayList<>(specs), + true, + new ArrayList<>(statisticsByPartition.values()), + overwrite); + } + + /** What one commit wrote into a partition, with one more of its files folded in. */ + private static PartitionStatistics sum(PartitionStatistics summed, PartitionStatistics file) { + return new PartitionStatistics( + summed.spec(), + add(summed.recordCount(), file.recordCount()), + add(summed.fileSizeInBytes(), file.fileSizeInBytes()), + summed.fileCount() + file.fileCount(), + summed.lastFileCreationTime(), + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS); + } + + /** A count nobody took leaves that field unknown for the whole partition. */ + private static long add(long sum, long value) { + return PartitionStatistics.isKnown(sum) && PartitionStatistics.isKnown(value) + ? sum + value + : PartitionStatistics.UNKNOWN; + } + + /** + * The partition a cleared directory belongs to, or null when it is none of this table's. + * Requiring the spec to rebuild the same directory rules out one nested below a partition, + * whose trailing components would otherwise read as some other partition; such a directory is + * left alone, since stale statistics beat statistics of the wrong partition. + */ + @Nullable + private Map clearedPartitionSpec(Path clearedPath) { + LinkedHashMap spec = + formatTablePartitionOnlyValueInPath + ? PartitionPathUtils.extractPartitionSpecFromPathOnlyValue( + clearedPath, partitionKeys) + : PartitionPathUtils.extractPartitionSpecFromPath( + clearedPath, partitionKeys); + if (spec == null) { + LOG.warn( + "Cleared directory {} of table {} is not one of its partition directories; " + + "its partition statistics are left unchanged.", + clearedPath, + tableIdentifier.getFullName()); + return null; + } + Path rebuilt = + buildPartitionPath( + location, spec, formatTablePartitionOnlyValueInPath, partitionKeys); + if (!samePathComponent(rebuilt, clearedPath)) { + LOG.warn( + "Cleared directory {} of table {} does not rebuild from partition spec {}; " + + "its partition statistics are left unchanged.", + clearedPath, + tableIdentifier.getFullName(), + spec); + return null; + } + return spec; + } + + /** + * Whether two paths name the same directory, ignoring scheme and authority: a {@link FileIO} + * that delegates answers a listing under the scheme it used, not the one it was asked with. + */ + private static boolean samePathComponent(Path left, Path right) { + return trimTrailingSeparators(left.toUri().normalize().getPath()) + .equals(trimTrailingSeparators(right.toUri().normalize().getPath())); + } + + private static String trimTrailingSeparators(String path) { + String trimmed = path; + while (trimmed.length() > 1 && trimmed.endsWith(Path.SEPARATOR)) { + trimmed = trimmed.substring(0, trimmed.length() - 1); + } + return trimmed; + } + private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException { Method hiveCreatePartitionsInHmsMethod = hiveCatalog @@ -276,8 +425,13 @@ public void abort(List commitMessages) { @Override public void close() throws Exception {} - private void deletePreviousDataFile(Path partitionPath, int partitionLevels) + /** + * Deletes the data files below a path and returns the directories they sat in, which for a + * static prefix overwrite can be partitions this commit never writes. + */ + private Set deletePreviousDataFile(Path partitionPath, int partitionLevels) throws IOException { + Set clearedPartitionPaths = new HashSet<>(); if (fileIO.exists(partitionPath)) { // Committed data files only: what sits under a staging directory is another writer's // uncommitted output, whatever its name looks like. @@ -289,13 +443,18 @@ private void deletePreviousDataFile(Path partitionPath, int partitionLevels) formatTablePartitionOnlyValueInPath, defaultPartName)) { try { - fileIO.delete(file.getPath(), false); + // 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()); + } } catch (FileNotFoundException ignore) { } catch (IOException e) { throw new RuntimeException(e); } } } + return clearedPartitionPaths; } @Override 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 new file mode 100644 index 000000000000..ed26d7074261 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java @@ -0,0 +1,676 @@ +/* + * 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.table.format; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +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.partition.PartitionStatistics; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** Tests for the partition statistics a {@link FormatTableCommit} reports. */ +class FormatTableCommitStatisticsTest { + + private static final List PARTITION_KEYS = Arrays.asList("year", "month"); + private static final String DEFAULT_PART_NAME = PARTITION_DEFAULT_NAME.defaultValue(); + private static final Identifier TABLE = + Identifier.create("statistics_db", "statistics_format_table"); + + @TempDir java.nio.file.Path tempDir; + + @Test + void testAppendReportsWhatItWroteAsAnIncrement() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + + long before = System.currentTimeMillis(); + commit(tablePath, fileIO, partitionManager, false, null) + .commit(Collections.singletonList(message)); + long after = System.currentTimeMillis(); + + Reported reported = capture(partitionManager); + assertThat(reported.replaceStatistics).isFalse(); + assertThat(reported.specs).containsExactly(spec("2025", "10")); + assertThat(reported.statistics).hasSize(1); + PartitionStatistics statistics = reported.statistics.get(0); + assertThat(statistics.spec()).isEqualTo(spec("2025", "10")); + assertThat(statistics.recordCount()).isEqualTo(3); + assertThat(statistics.fileSizeInBytes()).isEqualTo(128); + assertThat(statistics.fileCount()).isEqualTo(1); + // The contract is the commit's wall clock, so bounds pin it where positivity cannot. + assertThat(statistics.lastFileCreationTime()).isBetween(before, after); + assertThat(statistics.totalBuckets()).isEqualTo(PartitionStatistics.UNKNOWN_TOTAL_BUCKETS); + } + + @Test + void testFilesOfOnePartitionAreSummed() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + List messages = + Arrays.asList( + writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128), + writtenFile(fileIO, tablePath, "year=2025/month=10", 4, 256), + writtenFile(fileIO, tablePath, "year=2025/month=11", 5, 512)); + + commit(tablePath, fileIO, partitionManager, false, null).commit(messages); + + Reported reported = capture(partitionManager); + assertThat(reported.statistics) + .hasSize(2) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "10")); + assertThat(statistics.recordCount()).isEqualTo(7); + assertThat(statistics.fileSizeInBytes()).isEqualTo(384); + assertThat(statistics.fileCount()).isEqualTo(2); + }) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "11")); + assertThat(statistics.recordCount()).isEqualTo(5); + assertThat(statistics.fileCount()).isEqualTo(1); + }); + } + + @Test + void testAFileNobodyCountedMakesThePartitionUnknown() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + List messages = + Arrays.asList( + writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128), + // An older writer produced this one and counted nothing. + uncountedFile(fileIO, tablePath, "year=2025/month=10"), + // Counted, and after the one that was not: unknown has to stay unknown, + // or a partition missing a file comes out as an exact count of the rest. + writtenFile(fileIO, tablePath, "year=2025/month=10", 5, 512)); + + commit(tablePath, fileIO, partitionManager, false, null).commit(messages); + + PartitionStatistics statistics = capture(partitionManager).statistics.get(0); + // A sum missing a file must not be presented as an exact count. + assertThat(statistics.recordCount()).isEqualTo(PartitionStatistics.UNKNOWN); + assertThat(statistics.fileSizeInBytes()).isEqualTo(PartitionStatistics.UNKNOWN); + // The file count is still exact: it is counted here, not reported by the writer. + assertThat(statistics.fileCount()).isEqualTo(3); + } + + @Test + void testDynamicOverwriteReportsTheWholePartition() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + // Something was there before this commit replaced it. + writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + + commit(tablePath, fileIO, partitionManager, true, null) + .commit(Collections.singletonList(message)); + + Reported reported = capture(partitionManager); + assertThat(reported.replaceStatistics).isTrue(); + assertThat(reported.statistics).hasSize(1); + PartitionStatistics statistics = reported.statistics.get(0); + assertThat(statistics.recordCount()).isEqualTo(3); + assertThat(statistics.fileSizeInBytes()).isEqualTo(128); + assertThat(statistics.fileCount()).isEqualTo(1); + } + + @Test + void testStaticPrefixOverwriteZeroesAClearedPartitionAndKeepsItRegistered() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + // Two sibling partitions hold data; the overwrite writes only one of them. + writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096); + writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + + long before = System.currentTimeMillis(); + commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025")) + .commit(Collections.singletonList(message)); + long after = System.currentTimeMillis(); + + Reported reported = capture(partitionManager); + long commitTime = + reported.statistics.stream() + .filter(s -> s.spec().equals(spec("2025", "10"))) + .findFirst() + .orElseThrow(AssertionError::new) + .lastFileCreationTime(); + assertThat(commitTime).isBetween(before, after); + assertThat(reported.replaceStatistics).isTrue(); + // Red line: emptying a partition zeroes its statistics, it never unregisters it. + assertThat(reported.specs) + .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11")); + verify(partitionManager, never()).dropPartitions(anyList()); + assertThat(reported.statistics) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "11")); + assertThat(statistics.recordCount()).isZero(); + assertThat(statistics.fileSizeInBytes()).isZero(); + assertThat(statistics.fileCount()).isZero(); + // Emptying is dated to the commit that did it. Reporting the time as + // unknown would leave the stored one describing files that are gone, + // since an unknown replaces nothing. + assertThat(statistics.lastFileCreationTime()).isEqualTo(commitTime); + }); + } + + @Test + void testAClearedPartitionIsFoundEvenWhenTheListingAnswersUnderAnotherScheme() + throws Exception { + RescopingFileIO fileIO = new RescopingFileIO(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096); + writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + + commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025")) + .commit(Collections.singletonList(message)); + + // A listing does not have to answer under the URI it was asked with, and matching whole + // paths would then throw away a directory this very listing produced — leaving an emptied + // partition holding stale statistics. + assertThat(capture(partitionManager).statistics) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "11")); + assertThat(statistics.recordCount()).isZero(); + assertThat(statistics.fileCount()).isZero(); + }); + } + + @Test + void testADirectoryThatIsNoPartitionOfThisTableIsNotReported() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + // The key=value layout, where a directory that is not a partition of this table has no + // spec at all rather than a plausible wrong one: the prefix directory itself, and a + // directory nested below a partition. Clearing the prefix deletes the files in both, + // because the listing collects data files at every level, not only the partition one. + writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048); + writeDataFile(fileIO, tablePath, "year=2025", "orphan.csv", 512); + writeDataFile(fileIO, tablePath, "year=2025/month=10/nested", "old-data.csv", 1024); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + + commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025")) + .commit(Collections.singletonList(message)); + + // The commit succeeds and reports only the two real partitions. A directory with no spec + // is left alone: its statistics go stale, which beats failing the commit that just wrote + // the data, or accounting the files to a partition that does not exist. + Reported reported = capture(partitionManager); + assertThat(reported.specs) + .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11")); + assertThat(reported.statistics).hasSize(2); + } + + @Test + void testADirectoryBelowThePartitionIsNotReadAsAPartitionOfItsOwn() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + // In the value-only layout a partition directory is the bare value, so the trailing two + // components of 2025/10/nested read as the plausible partition {year=10, month=nested}. + writeDataFile(fileIO, tablePath, "2025/10", "old-data.csv", 4096); + writeDataFile(fileIO, tablePath, "2025/10/nested", "old-data.csv", 4096); + writeDataFile(fileIO, tablePath, "2025/11", "old-data.csv", 2048); + CommitMessage message = writtenFile(fileIO, tablePath, "2025/10", 3, 128); + + commit( + tablePath, + fileIO, + partitionManager, + true, + Collections.singletonMap("year", "2025"), + true) + .commit(Collections.singletonList(message)); + + // Only directories the spec rebuilds are reported: accounting 2025/10/nested to a partition + // named {year=10, month=nested} would zero a partition this commit never touched. + Reported reported = capture(partitionManager); + assertThat(reported.specs) + .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11")); + assertThat(reported.statistics) + .hasSize(2) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "10")); + assertThat(statistics.recordCount()).isEqualTo(3); + assertThat(statistics.fileCount()).isEqualTo(1); + }) + .anySatisfy( + statistics -> { + assertThat(statistics.spec()).isEqualTo(spec("2025", "11")); + assertThat(statistics.recordCount()).isZero(); + assertThat(statistics.fileCount()).isZero(); + }); + } + + @Test + void testTheIncrementsOfConcurrentWritersOfOnePartitionSum() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + // A Flink sink commits once per writer subtask, in the subtask's own close(): writing one + // partition at parallelism 3 is three independent commits against the same catalog, and + // the partition total is whatever the catalog makes of the three reports. + AccumulatingPartitionManager catalog = new AccumulatingPartitionManager(); + long[][] perSubtask = {{3, 128}, {4, 256}, {5, 512}}; + + for (long[] subtask : perSubtask) { + commit(tablePath, fileIO, catalog, false, null) + .commit( + Collections.singletonList( + writtenFile( + fileIO, + tablePath, + "year=2025/month=10", + subtask[0], + subtask[1]))); + } + + // Each subtask saw only its own files, so each reports an increment. Reporting the whole + // partition instead would make the last subtask to close the only one that counted. + assertThat(catalog.replaceFlags).containsExactly(false, false, false); + assertThat(catalog.registered).containsOnly(spec("2025", "10")); + PartitionStatistics total = catalog.stored.get(spec("2025", "10")); + assertThat(total).isNotNull(); + assertThat(total.recordCount()).isEqualTo(12); + assertThat(total.fileSizeInBytes()).isEqualTo(896); + assertThat(total.fileCount()).isEqualTo(3); + } + + private FormatTableCommit commit( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + boolean overwrite, + Map staticPartitions) { + return commit(tablePath, fileIO, partitionManager, overwrite, staticPartitions, false); + } + + private FormatTableCommit commit( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + boolean overwrite, + Map staticPartitions, + boolean onlyValueInPath) { + return new FormatTableCommit( + tablePath.toString(), + PARTITION_KEYS, + fileIO, + onlyValueInPath, + DEFAULT_PART_NAME, + overwrite, + TABLE, + staticPartitions, + null, + null, + partitionManager); + } + + /** A file this commit wrote, with the counts its writer took. */ + private CommitMessage writtenFile( + FileIO fileIO, + Path tablePath, + String partitionDir, + long recordCount, + long fileSizeInBytes) + throws Exception { + return new TwoPhaseCommitMessage( + stage(fileIO, tablePath, partitionDir), recordCount, fileSizeInBytes); + } + + /** A file committed by a writer that reported no counts. */ + private CommitMessage uncountedFile(LocalFileIO fileIO, Path tablePath, String partitionDir) + throws Exception { + return new TwoPhaseCommitMessage(stage(fileIO, tablePath, partitionDir)); + } + + private TwoPhaseOutputStream.Committer stage(FileIO fileIO, Path tablePath, String partitionDir) + throws Exception { + Path targetPath = + new Path(new Path(tablePath, partitionDir), "data-" + UUID.randomUUID() + ".csv"); + RenamingTwoPhaseOutputStream outputStream = + new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); + outputStream.write(1); + return outputStream.closeForCommit(); + } + + private static void writeDataFile( + FileIO fileIO, Path tablePath, String partitionDir, String name, int bytes) + throws Exception { + Path path = new Path(new Path(tablePath, partitionDir), name); + fileIO.mkdirs(path.getParent()); + try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { + out.write(new byte[bytes]); + } + } + + private static Map spec(String year, String month) { + LinkedHashMap spec = new LinkedHashMap<>(); + spec.put("year", year); + spec.put("month", month); + return spec; + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static Reported capture(FormatTablePartitionManager partitionManager) { + ArgumentCaptor>> specs = + ArgumentCaptor.forClass((Class) List.class); + ArgumentCaptor> statistics = + ArgumentCaptor.forClass((Class) List.class); + ArgumentCaptor replaceStatistics = ArgumentCaptor.forClass(Boolean.class); + verify(partitionManager) + .createPartitions( + specs.capture(), + eq(true), + statistics.capture(), + replaceStatistics.capture()); + return new Reported( + new ArrayList<>(specs.getValue()), + new ArrayList<>(statistics.getValue()), + replaceStatistics.getValue()); + } + + /** + * 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. + */ + private static class RescopingFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + FileStatus[] statuses = super.listStatus(path); + for (int i = 0; i < statuses.length; i++) { + statuses[i] = new RescopedFileStatus(statuses[i]); + } + return statuses; + } + } + + private static class RescopedFileStatus implements FileStatus { + + private final FileStatus delegate; + + private RescopedFileStatus(FileStatus delegate) { + this.delegate = delegate; + } + + @Override + public long getLen() { + return delegate.getLen(); + } + + @Override + public boolean isDir() { + return delegate.isDir(); + } + + @Override + public Path getPath() { + return new Path(delegate.getPath().toUri().getPath()); + } + + @Override + public long getModificationTime() { + return delegate.getModificationTime(); + } + } + + /** + * A partition manager that folds the reports it receives the way a catalog does: ADD + * accumulates onto what is held, SET replaces it. It holds what several independent commits + * against one table add up to. + */ + private static class AccumulatingPartitionManager implements FormatTablePartitionManager { + + private static final long serialVersionUID = 1L; + + private final List> registered = new ArrayList<>(); + private final Map, PartitionStatistics> stored = new LinkedHashMap<>(); + private final List replaceFlags = new ArrayList<>(); + + @Override + public void createPartitions( + List> partitions, + boolean ignoreIfExists, + @Nullable List statistics, + boolean replaceStatistics) { + createPartitions(partitions, ignoreIfExists); + if (statistics == null) { + return; + } + replaceFlags.add(replaceStatistics); + for (PartitionStatistics reported : statistics) { + PartitionStatistics held = stored.get(reported.spec()); + if (held == null || replaceStatistics) { + stored.put(reported.spec(), reported); + continue; + } + stored.put( + reported.spec(), + new PartitionStatistics( + reported.spec(), + held.recordCount() + reported.recordCount(), + held.fileSizeInBytes() + reported.fileSizeInBytes(), + held.fileCount() + reported.fileCount(), + Math.max( + held.lastFileCreationTime(), + reported.lastFileCreationTime()), + PartitionStatistics.UNKNOWN_TOTAL_BUCKETS)); + } + } + + @Override + public void createPartitions(List> partitions, boolean ignoreIfExists) { + registered.addAll(partitions); + } + + @Override + public List listPartitions( + Map prefix, @Nullable Predicate filter) { + throw new UnsupportedOperationException(); + } + + @Override + public List listPartitionsByNames(List> partitions) { + throw new UnsupportedOperationException(); + } + + @Override + public void dropPartitions(List> partitions) { + throw new UnsupportedOperationException(); + } + } + + @Test + void testTheNumbersReachTheCatalogThroughTheWriteBuilder() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + Map options = new HashMap<>(); + options.put(CoreOptions.PATH.key(), tablePath.toString()); + options.put(CoreOptions.FILE_FORMAT.key(), "csv"); + FormatTable table = + FormatTable.builder() + .fileIO(fileIO) + .identifier(Identifier.create("test_db", "test_table")) + .rowType( + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.STRING()}, + new String[] {"id", "year"})) + .partitionKeys(Collections.singletonList("year")) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options(options) + .partitionManager(partitionManager) + .build(); + + // The whole path rather than a commit built by hand: the write builder, the commit it + // builds, and the numbers the writer counted on the way. + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + BatchTableWrite write = writeBuilder.newWrite(); + write.write(GenericRow.of(1, BinaryString.fromString("2025"))); + write.write(GenericRow.of(2, BinaryString.fromString("2025"))); + List messages = write.prepareCommit(); + writeBuilder.newCommit().commit(messages); + + Reported reported = capture(partitionManager); + assertThat(reported.replaceStatistics).isFalse(); + assertThat(reported.specs).containsExactly(Collections.singletonMap("year", "2025")); + assertThat(reported.statistics).hasSize(1); + PartitionStatistics statistics = reported.statistics.get(0); + assertThat(statistics.recordCount()).isEqualTo(2); + assertThat(statistics.fileCount()).isEqualTo(1); + // The byte size is the writer's own count, so it has to match what landed on disk. + long onDisk = 0; + for (FileStatus file : fileIO.listStatus(new Path(tablePath, "year=2025"))) { + if (!file.isDir()) { + onDisk += file.getLen(); + } + } + assertThat(statistics.fileSizeInBytes()).isEqualTo(onDisk); + } + + @Test + void testAFailedReportFailsTheCommitAndDiscardsWhatItWrote() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + RuntimeException failure = new RuntimeException("catalog says 429"); + doThrow(failure) + .when(partitionManager) + .createPartitions(anyList(), anyBoolean(), any(), anyBoolean()); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + Path written = ((TwoPhaseCommitMessage) message).getCommitter().targetPath(); + + // Registration and statistics ride in one request, so a failure says nothing about + // whether the partition was registered. Committing anyway would leave data behind that + // nothing points at. + assertThatThrownBy( + () -> + commit(tablePath, fileIO, partitionManager, false, null) + .commit(Collections.singletonList(message))) + .hasRootCause(failure); + + assertThat(fileIO.exists(written)).isFalse(); + } + + @Test + void testAFailedReportOfAnOverwriteLeavesThePartitionEmpty() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + doThrow(new RuntimeException("catalog says 429")) + .when(partitionManager) + .createPartitions(anyList(), anyBoolean(), any(), anyBoolean()); + writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096); + CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128); + Path written = ((TwoPhaseCommitMessage) message).getCommitter().targetPath(); + + assertThatThrownBy( + () -> + commit( + tablePath, + fileIO, + partitionManager, + true, + Collections.singletonMap("year", "2025")) + .commit(Collections.singletonList(message))) + .hasRootCauseMessage("catalog says 429"); + + // The state this leaves is worth stating rather than discovering: the overwrite already + // deleted what the partition held, and the abort takes back what it wrote, so the + // partition is empty on disk while the catalog still describes what used to be there. + assertThat(fileIO.exists(written)).isFalse(); + assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=10/old-data.csv"))).isFalse(); + } + + /** What one call reported to the catalog. */ + private static class Reported { + private final List> specs; + private final List statistics; + private final boolean replaceStatistics; + + private Reported( + List> specs, + List statistics, + boolean replaceStatistics) { + this.specs = specs; + this.statistics = statistics; + this.replaceStatistics = replaceStatistics; + } + } +} 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 c7a557b4498a..6d3d6da6f236 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 @@ -41,6 +41,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.entry; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; @@ -65,7 +67,9 @@ void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); RuntimeException registrationFailure = new RuntimeException("Catalog partition registration unavailable"); - doThrow(registrationFailure).when(partitionManager).createPartitions(anyList(), eq(true)); + doThrow(registrationFailure) + .when(partitionManager) + .createPartitions(anyList(), eq(true), any(), anyBoolean()); FormatTableCommit commit = new FormatTableCommit( @@ -89,7 +93,7 @@ void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception // A failed write leaves nothing behind, whichever step failed: rerunning it converges, // and an idempotent registration makes a partition that was registered anyway harmless. assertThat(fileIO.exists(targetPath)).isFalse(); - verify(partitionManager).createPartitions(anyList(), eq(true)); + verify(partitionManager).createPartitions(anyList(), eq(true), any(), anyBoolean()); } @Test @@ -119,7 +123,8 @@ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception { .hasRootCauseMessage("data commit failed"); verify(committer).discard(fileIO); - verify(partitionManager, never()).createPartitions(anyList(), eq(true)); + verify(partitionManager, never()) + .createPartitions(anyList(), eq(true), any(), anyBoolean()); } @Test @@ -385,7 +390,7 @@ private static Map registeredSpec( FormatTablePartitionManager partitionManager) { ArgumentCaptor>> captor = ArgumentCaptor.forClass((Class) List.class); - verify(partitionManager).createPartitions(captor.capture(), eq(true)); + verify(partitionManager).createPartitions(captor.capture(), eq(true), any(), anyBoolean()); assertThat(captor.getValue()).hasSize(1); return captor.getValue().get(0); }