Skip to content

[core] Repeated append-only compaction retains intermediate files #10448

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Oct 9, 2026
Merged

[core] Repeated append-only compaction retains intermediate files #10448

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 33 additions & 17 deletions paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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());
}
}
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -367,14 +375,22 @@ private RollingFileWriter<InternalRow, DataFileMeta> 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) {
Expand All @@ -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;

Expand Down
109 changes: 109 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,18 +45,23 @@
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;
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;
Expand Down Expand Up @@ -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<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);
Expand Down Expand Up @@ -1317,6 +1395,37 @@ private Pair<AppendOnlyWriter, List<DataFileMeta>> createWriterBase(
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();
Expand Down
Loading