From 4d6a2f60347f28767ba2000d0434304ffb4a36d1 Mon Sep 17 00:00:00 2001 From: Ran Tao Date: Thu, 8 Oct 2026 19:14:39 +0800 Subject: [PATCH] [core] Repeated append-only compaction retains intermediate files Clean up intermediate compaction outputs while preserving files already handed to a committer. --- .../paimon/append/AppendOnlyWriter.java | 50 +++++--- .../paimon/append/AppendOnlyWriterTest.java | 109 ++++++++++++++++++ 2 files changed, 142 insertions(+), 17 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java index 906573ad73ca..c26376395f15 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java @@ -21,6 +21,7 @@ import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.compact.CompactDeletionFile; import org.apache.paimon.compact.CompactManager; +import org.apache.paimon.compact.CompactResult; import org.apache.paimon.compression.CompressOptions; import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.IOManager; @@ -58,6 +59,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.function.Supplier; @@ -92,8 +94,8 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner { private final boolean omitAllNonDedicatedWriteCols; private final List newFiles; private final List deletedFiles; - private final List compactBefore; - private final List compactAfter; + private final LinkedHashMap compactBefore; + private final LinkedHashMap compactAfter; private final Supplier seqNumCounterProvider; private final String fileCompression; private final CompressOptions spillCompression; @@ -160,8 +162,8 @@ public AppendOnlyWriter( this.omitAllNonDedicatedWriteCols = omitAllNonDedicatedWriteCols; this.newFiles = new ArrayList<>(); this.deletedFiles = new ArrayList<>(); - this.compactBefore = new ArrayList<>(); - this.compactAfter = new ArrayList<>(); + this.compactBefore = new LinkedHashMap<>(); + this.compactAfter = new LinkedHashMap<>(); final LongCounter seqNumCounter = new LongCounter(maxSequenceNumber + 1); this.seqNumCounterProvider = dataEvolutionEnabled ? () -> new LongCounter(0) : () -> seqNumCounter; @@ -180,8 +182,14 @@ public AppendOnlyWriter( if (increment != null) { newFiles.addAll(increment.newFilesIncrement().newFiles()); deletedFiles.addAll(increment.newFilesIncrement().deletedFiles()); - compactBefore.addAll(increment.compactIncrement().compactBefore()); - compactAfter.addAll(increment.compactIncrement().compactAfter()); + increment + .compactIncrement() + .compactBefore() + .forEach(file -> compactBefore.put(file.fileName(), file)); + increment + .compactIncrement() + .compactAfter() + .forEach(file -> compactAfter.put(file.fileName(), file)); updateCompactDeletionFile(increment.compactDeletionFile()); } } @@ -285,7 +293,7 @@ public void close() throws Exception { sync(); compactManager.close(); - for (DataFileMeta file : compactAfter) { + for (DataFileMeta file : compactAfter.values()) { // appendOnlyCompactManager will rewrite the file and no file upgrade will occur, so we // can directly delete the file in compactAfter. file.collectFiles(pathFactory).forEach(fileIO::deleteQuietly); @@ -367,14 +375,22 @@ private RollingFileWriter createRollingRowWriter() { private void trySyncLatestCompaction(boolean blocking) throws ExecutionException, InterruptedException { - compactManager - .getCompactionResult(blocking) - .ifPresent( - result -> { - compactBefore.addAll(result.before()); - compactAfter.addAll(result.after()); - updateCompactDeletionFile(result.deletionFile()); - }); + compactManager.getCompactionResult(blocking).ifPresent(this::updateCompactResult); + } + + private void updateCompactResult(CompactResult result) { + for (DataFileMeta file : result.before()) { + DataFileMeta intermediate = compactAfter.remove(file.fileName()); + if (intermediate != null) { + // Append compaction rewrites files. An output compacted again before the next + // prepareCommit has not been handed to a committer and can be deleted directly. + intermediate.collectFiles(pathFactory).forEach(fileIO::deleteQuietly); + } else { + compactBefore.put(file.fileName(), file); + } + } + result.after().forEach(file -> compactAfter.put(file.fileName(), file)); + updateCompactDeletionFile(result.deletionFile()); } private void updateCompactDeletionFile(@Nullable CompactDeletionFile newDeletionFile) { @@ -394,8 +410,8 @@ private CommitIncrement drainIncrement() { Collections.emptyList()); CompactIncrement compactIncrement = new CompactIncrement( - new ArrayList<>(compactBefore), - new ArrayList<>(compactAfter), + new ArrayList<>(compactBefore.values()), + new ArrayList<>(compactAfter.values()), Collections.emptyList()); CompactDeletionFile drainDeletionFile = compactDeletionFile; diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java index 3b2464d311d3..742d219063b3 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java @@ -45,11 +45,15 @@ import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.manifest.FileEntry; +import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.memory.HeapMemorySegmentPool; import org.apache.paimon.memory.MemoryPoolFactory; import org.apache.paimon.operation.BaseAppendFileStoreWrite; import org.apache.paimon.operation.BlobFileContext; +import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.schema.Schema; @@ -57,6 +61,7 @@ import org.apache.paimon.stats.SimpleStatsConverter; import org.apache.paimon.table.AppendOnlyFileStoreTable; import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.types.BlobType; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; @@ -351,6 +356,79 @@ public void testRollingWrite() throws Exception { assertThat(secInc.newFilesIncrement().newFiles()).hasSize(1); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testRepeatedCompaction(boolean commitBetweenCompactions) throws Exception { + List inputs = createCompactionInputs(5); + AppendOnlyWriter writer = createWriter(1024 * 1024L, false, inputs.subList(0, 3)).getLeft(); + try { + // The first compaction produces a small file which is eligible for compaction again. + writer.flush(false, false); + writer.sync(); + assertThat(writer.dataFiles()).hasSize(1); + DataFileMeta intermediate = writer.dataFiles().iterator().next(); + assertThat(intermediate.extraFiles()).hasSize(1); + + List increments = new ArrayList<>(); + if (commitBetweenCompactions) { + CommitIncrement first = writer.prepareCommit(true); + assertThat(first.compactIncrement().compactBefore()) + .containsExactlyElementsOf(inputs.subList(0, 3)); + assertThat(first.compactIncrement().compactAfter()).containsExactly(intermediate); + increments.add(first); + } + + // Add two more inputs so the next compaction rewrites the intermediate file. + writer.addNewFiles(inputs.subList(3, 5)); + writer.flush(false, false); + writer.sync(); + CommitIncrement last = writer.prepareCommit(true); + increments.add(last); + + assertThat(last.compactIncrement().compactBefore()) + .containsExactlyInAnyOrderElementsOf( + commitBetweenCompactions + ? Arrays.asList(intermediate, inputs.get(3), inputs.get(4)) + : inputs); + assertThat(last.compactIncrement().compactAfter()).hasSize(1); + DataFileMeta output = last.compactIncrement().compactAfter().get(0); + assertThat(output.rowCount()).isEqualTo(5); + + // A file handed to a previous commit must remain available even before that commit + // finishes. Otherwise, both the intermediate data file and its index can be removed. + for (Path path : intermediate.collectFiles(pathFactory)) { + assertThat(LocalFileIO.create().exists(path)).isEqualTo(commitBetweenCompactions); + } + for (DataFileMeta file : inputs) { + assertThat(LocalFileIO.create().exists(pathFactory.toPath(file))).isTrue(); + } + for (Path path : output.collectFiles(pathFactory)) { + assertThat(LocalFileIO.create().exists(path)).isTrue(); + } + + List entries = new ArrayList<>(); + for (DataFileMeta file : inputs) { + entries.add(ManifestEntry.create(FileKind.ADD, BinaryRow.EMPTY_ROW, 0, 1, file)); + } + ManifestEntryChanges changes = new ManifestEntryChanges(1); + for (CommitIncrement increment : increments) { + changes.collect( + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + 1, + increment.newFilesIncrement(), + increment.compactIncrement())); + } + entries.addAll(changes.compactTableFiles); + assertThat(FileEntry.mergeEntries(entries)) + .extracting(ManifestEntry::file) + .containsExactly(output); + } finally { + writer.close(); + } + } + @Test public void testCloseUnexpectedly() throws Exception { final CountDownLatch latch = new CountDownLatch(1); @@ -1317,6 +1395,37 @@ private Pair> createWriterBase( return Pair.of(writer, compactManager.allFiles()); } + private List createCompactionInputs(int count) throws IOException { + List files = new ArrayList<>(); + for (int i = 0; i < count; i++) { + Path path = pathFactory.newPath("input-"); + LocalFileIO.create().newOutputStream(path, false).close(); + String name = String.format("%03d", i); + files.add( + DataFileMeta.forAppend( + path.getName(), + 10L, + 1L, + STATS_SERIALIZER.toBinaryAllMode( + new SimpleColStats[] { + initStats(i, i, 0), + initStats(name, name, 0), + initStats(PART, PART, 0) + }), + i, + i, + SCHEMA_ID, + Collections.emptyList(), + null, + FileSource.APPEND, + null, + null, + null, + null)); + } + return files; + } + private DataFileMeta generateCompactAfter(List toCompact) throws IOException { int size = toCompact.size(); long minSeq = toCompact.get(0).minSequenceNumber();