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 5d8110d660 [core] Preserve reused compaction files when aborting
commits (#10582)
5d8110d660 is described below
commit 5d8110d6600221f7cdb23a0ba618115e5eec2014
Author: Luo Zhengjie <[email protected]>
AuthorDate: Sun Oct 18 16:57:34 2026 +0800
[core] Preserve reused compaction files when aborting commits (#10582)
---
.../paimon/operation/FileStoreCommitImpl.java | 15 ++-
.../paimon/operation/FileStoreCommitTest.java | 101 +++++++++++++++++++++
2 files changed, 115 insertions(+), 1 deletion(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 2123517e13..ffe79cf4bb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -27,6 +27,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
@@ -725,13 +726,25 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
List<DataFileMeta> dataFilesToDelete = new ArrayList<>();
dataFilesToDelete.addAll(commitMessage.newFilesIncrement().newFiles());
dataFilesToDelete.addAll(commitMessage.newFilesIncrement().changelogFiles());
-
dataFilesToDelete.addAll(commitMessage.compactIncrement().compactAfter());
dataFilesToDelete.addAll(commitMessage.compactIncrement().changelogFiles());
+ // New files may also be compaction inputs, but abort still owns
their cleanup.
for (DataFileMeta file : dataFilesToDelete) {
file.collectFiles(dataPathFactory).forEach(fileIO::deleteQuietly);
}
+ // A level upgrade may reuse a main file while adding new
auxiliary files.
+ Set<Path> compactInputPaths =
+ commitMessage.compactIncrement().compactBefore().stream()
+ .flatMap(file ->
file.collectFiles(dataPathFactory).stream())
+ .collect(Collectors.toSet());
+
+ for (DataFileMeta file :
commitMessage.compactIncrement().compactAfter()) {
+ file.collectFiles(dataPathFactory).stream()
+ .filter(path -> !compactInputPaths.contains(path))
+ .forEach(fileIO::deleteQuietly);
+ }
+
List<IndexFileMeta> indexFilesToDelete = new ArrayList<>();
indexFilesToDelete.addAll(commitMessage.newFilesIncrement().newIndexFiles());
indexFilesToDelete.addAll(commitMessage.compactIncrement().newIndexFiles());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index f4501aacce..99fd1f6532 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1224,6 +1224,107 @@ public class FileStoreCommitTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAbortPreservesReusedCompactionFiles(boolean external)
throws Exception {
+ TestAppendFileStore store =
TestAppendFileStore.createAppendStore(tempDir, new HashMap<>());
+ BinaryRow partition = gen.getPartition(gen.next());
+ DataFilePathFactory pathFactory =
+ store.pathFactory().createDataFilePathFactory(partition, 0);
+ Path compactNewPath = pathFactory.newPath();
+ // External files may share a name with a different physical file in
the bucket directory.
+ Path reusedPath =
+ external
+ ? new
Path(tempDir.resolve(compactNewPath.getName()).toUri())
+ : pathFactory.newPath();
+ DataFileMeta reused = createDataFileWithExtraFile(store, reusedPath,
external);
+ DataFileMeta compactNew = createDataFileWithExtraFile(store,
compactNewPath, false);
+ Path newLookupPath = new Path(reusedPath.getParent(),
"new-lookup.sst");
+ store.fileIO().newOutputStream(newLookupPath, false).close();
+ List<String> upgradedExtraFiles = new ArrayList<>(reused.extraFiles());
+ upgradedExtraFiles.add(newLookupPath.getName());
+ DataFileMeta upgraded = reused.upgrade(1).copy(upgradedExtraFiles);
+ CommitMessage commitMessage =
+ new CommitMessageImpl(
+ partition,
+ 0,
+ store.options().bucket(),
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList()),
+ new CompactIncrement(
+ Collections.singletonList(reused),
+ Arrays.asList(upgraded, compactNew),
+ Collections.emptyList()));
+
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.abort(Collections.singletonList(commitMessage));
+ commit.abort(Collections.singletonList(commitMessage));
+ }
+
+ for (Path path : reused.collectFiles(pathFactory)) {
+ assertThat(store.fileIO().exists(path)).as("reused file %s",
path).isTrue();
+ }
+ for (Path path : compactNew.collectFiles(pathFactory)) {
+ assertThat(store.fileIO().exists(path)).as("new compaction file
%s", path).isFalse();
+ }
+ assertThat(store.fileIO().exists(newLookupPath))
+ .as("new lookup file %s", newLookupPath)
+ .isFalse();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"NONE", "INPUT", "FULL_COMPACTION"})
+ public void testAbortCleansUncommittedCompactionInputs(String
changelogProducer)
+ throws Exception {
+ TestFileStore store =
+ createStore(false, 1,
CoreOptions.ChangelogProducer.valueOf(changelogProducer));
+ KeyValue first = gen.nextInsert("20201110", 10, 1L, new int[] {1, 1},
"first");
+ BinaryRow partition = gen.getPartition(first);
+ DataFilePathFactory pathFactory =
+ store.pathFactory().createDataFilePathFactory(partition, 0);
+
+ AbstractFileStoreWrite<KeyValue> write = store.newWrite();
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ write.write(partition, 0, first);
+ write.compact(partition, 0, false);
+ write.write(partition, 0, first);
+ write.compact(partition, 0, true);
+ List<CommitMessage> messages = write.prepareCommit(true, 0);
+ assertThat(messages).hasSize(1);
+ CommitMessageImpl message = (CommitMessageImpl) messages.get(0);
+ assertThat(message.newFilesIncrement().newFiles()).hasSize(2);
+ assertThat(message.compactIncrement().compactBefore())
+
.containsExactlyInAnyOrderElementsOf(message.newFilesIncrement().newFiles());
+ assertThat(message.compactIncrement().compactAfter()).hasSize(1);
+
+ List<DataFileMeta> pendingFiles =
+ new ArrayList<>(message.newFilesIncrement().newFiles());
+ pendingFiles.addAll(message.newFilesIncrement().changelogFiles());
+ pendingFiles.addAll(message.compactIncrement().compactAfter());
+ pendingFiles.addAll(message.compactIncrement().changelogFiles());
+ List<Path> pendingPaths =
+ pendingFiles.stream()
+ .flatMap(file ->
file.collectFiles(pathFactory).stream())
+ .collect(Collectors.toList());
+ for (Path path : pendingPaths) {
+ assertThat(store.fileIO().exists(path)).as("pending file %s",
path).isTrue();
+ }
+ assertThat(store.snapshotManager().latestSnapshotId()).isNull();
+
+ commit.abort(messages);
+ commit.abort(messages);
+
+ for (Path path : pendingPaths) {
+ assertThat(store.fileIO().exists(path)).as("aborted file %s",
path).isFalse();
+ }
+ assertThat(store.snapshotManager().latestSnapshotId()).isNull();
+ } finally {
+ write.close();
+ }
+ }
+
private static DataFileMeta createDataFileWithExtraFile(
TestAppendFileStore store, Path path, boolean external) throws
Exception {
store.fileIO().newOutputStream(path, false).close();
