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 f218528437 [core] Clean up index files on aborted commits (#8572)
f218528437 is described below
commit f218528437bbd697f3ff63261ebefb7d414a1c0e
Author: QuakeWang <[email protected]>
AuthorDate: Mon Jul 13 10:17:48 2026 +0800
[core] Clean up index files on aborted commits (#8572)
`FileStoreCommitImpl.abort` deleted uncommitted data and changelog files
but ignored `newIndexFiles` from both data and compact increments.
Aborted commits could therefore leave index files behind, including
files stored at external paths.
Clean up both sets of new index files through `IndexFilePathFactories`.
Keep `deletedIndexFiles` untouched because they still belong to the
active snapshot when the commit is aborted.
---
.../paimon/operation/FileStoreCommitImpl.java | 34 ++++++++++----
.../paimon/operation/FileStoreCommitTest.java | 53 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 10 deletions(-)
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 cb2357eabe..07505fa3e4 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,8 @@ 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.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.FileEntry;
@@ -75,6 +77,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.DataFilePathFactories;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.ListUtils;
import org.apache.paimon.utils.Pair;
@@ -670,18 +673,29 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
@Override
public void abort(List<CommitMessage> commitMessages) {
- DataFilePathFactories factories = new
DataFilePathFactories(pathFactory);
+ DataFilePathFactories dataFactories = new
DataFilePathFactories(pathFactory);
+ IndexFilePathFactories indexFactories = new
IndexFilePathFactories(pathFactory);
for (CommitMessage message : commitMessages) {
- DataFilePathFactory pathFactory =
factories.get(message.partition(), message.bucket());
+ DataFilePathFactory dataPathFactory =
+ dataFactories.get(message.partition(), message.bucket());
+ IndexPathFactory indexPathFactory =
+ indexFactories.get(message.partition(), message.bucket());
CommitMessageImpl commitMessage = (CommitMessageImpl) message;
- List<DataFileMeta> toDelete = new ArrayList<>();
- toDelete.addAll(commitMessage.newFilesIncrement().newFiles());
-
toDelete.addAll(commitMessage.newFilesIncrement().changelogFiles());
- toDelete.addAll(commitMessage.compactIncrement().compactAfter());
- toDelete.addAll(commitMessage.compactIncrement().changelogFiles());
-
- for (DataFileMeta file : toDelete) {
- fileIO.deleteQuietly(pathFactory.toPath(file));
+ 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());
+
+ for (DataFileMeta file : dataFilesToDelete) {
+ fileIO.deleteQuietly(dataPathFactory.toPath(file));
+ }
+
+ List<IndexFileMeta> indexFilesToDelete = new ArrayList<>();
+
indexFilesToDelete.addAll(commitMessage.newFilesIncrement().newIndexFiles());
+
indexFilesToDelete.addAll(commitMessage.compactIncrement().newIndexFiles());
+ for (IndexFileMeta file : indexFilesToDelete) {
+ fileIO.deleteQuietly(indexPathFactory.toPath(file));
}
}
}
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 e3903b5077..97f4a5b804 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
@@ -34,6 +34,7 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
@@ -960,6 +961,58 @@ public class FileStoreCommitTest {
assertThat(dvs.get("f2").isDeleted(3)).isTrue();
}
+ @Test
+ public void testAbortIndexFiles() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.INDEX_FILE_IN_DATA_FILE_DIR.key(), "true");
+ TestAppendFileStore store =
TestAppendFileStore.createAppendStore(tempDir, options);
+ BinaryRow partition = gen.getPartition(gen.next());
+ IndexPathFactory indexPathFactory =
store.pathFactory().indexFileFactory(partition, 0);
+
+ Path dataNewPath = indexPathFactory.newPath();
+ IndexFileMeta dataNew = createIndexFile(store, dataNewPath, false);
+ Path compactNewPath = new
Path(tempDir.resolve("external-new-index").toUri());
+ IndexFileMeta compactNew = createIndexFile(store, compactNewPath,
true);
+ Path dataDeletedPath = indexPathFactory.newPath();
+ IndexFileMeta dataDeleted = createIndexFile(store, dataDeletedPath,
false);
+ Path compactDeletedPath = indexPathFactory.newPath();
+ IndexFileMeta compactDeleted = createIndexFile(store,
compactDeletedPath, false);
+
+ CommitMessage commitMessage =
+ new CommitMessageImpl(
+ partition,
+ 0,
+ store.options().bucket(),
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonList(dataNew),
+ Collections.singletonList(dataDeleted)),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonList(compactNew),
+ Collections.singletonList(compactDeleted)));
+
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ commit.abort(Collections.singletonList(commitMessage));
+ }
+
+ assertThat(store.fileIO().exists(dataNewPath)).isFalse();
+ assertThat(store.fileIO().exists(compactNewPath)).isFalse();
+ assertThat(store.fileIO().exists(dataDeletedPath)).isTrue();
+ assertThat(store.fileIO().exists(compactDeletedPath)).isTrue();
+ }
+
+ private static IndexFileMeta createIndexFile(
+ TestAppendFileStore store, Path path, boolean external) throws
Exception {
+ store.fileIO().newOutputStream(path, false).close();
+ return new IndexFileMeta(
+ HASH_INDEX, path.getName(), 0, 0, null, external ?
path.toString() : null, null);
+ }
+
@Test
public void testRollbackToAsLatestFileLevelDeleteIsVisibleToStreaming()
throws Exception {
// Contrast with the DV-only case: when a rollback removes whole data
files, the delete is
