This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 02302d8597 [core] Repeated append-only compaction retains intermediate
files (#10448)
02302d8597 is described below
commit 02302d859727e96b6eae570e6c6ad32c19edcb61
Author: Ran Tao <[email protected]>
AuthorDate: Fri Oct 9 21:56:06 2026 +0800
[core] Repeated append-only compaction retains intermediate files (#10448)
---
.../org/apache/paimon/append/AppendOnlyWriter.java | 50 ++++++----
.../apache/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 906573ad73..c26376395f 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 @@ package org.apache.paimon.append;
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 javax.annotation.Nullable;
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<DataFileMeta> newFiles;
private final List<DataFileMeta> deletedFiles;
- private final List<DataFileMeta> compactBefore;
- private final List<DataFileMeta> compactAfter;
+ private final LinkedHashMap<String, DataFileMeta> compactBefore;
+ private final LinkedHashMap<String, DataFileMeta> compactAfter;
private final Supplier<LongCounter> seqNumCounterProvider;
private final String fileCompression;
private final CompressOptions spillCompression;
@@ -160,8 +162,8 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
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 class AppendOnlyWriter implements
BatchRecordWriter, MemoryOwner {
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 class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
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 @@ public class AppendOnlyWriter implements
BatchRecordWriter, MemoryOwner {
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 @@ public class AppendOnlyWriter implements BatchRecordWriter,
MemoryOwner {
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 3b2464d311..742d219063 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.Path;
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.schema.TableSchema;
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 class AppendOnlyWriterTest {
assertThat(secInc.newFilesIncrement().newFiles()).hasSize(1);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRepeatedCompaction(boolean commitBetweenCompactions)
throws Exception {
+ List<DataFileMeta> 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<CommitIncrement> 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<ManifestEntry> 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 @@ public class AppendOnlyWriterTest {
return Pair.of(writer, compactManager.allFiles());
}
+ private List<DataFileMeta> createCompactionInputs(int count) throws
IOException {
+ List<DataFileMeta> 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<DataFileMeta> toCompact)
throws IOException {
int size = toCompact.size();
long minSeq = toCompact.get(0).minSequenceNumber();
