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