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 c872898bc5 [core] Manage manifest extra file lifecycle (#9886)
c872898bc5 is described below
commit c872898bc547dee93b0e650644e7310b7905022e
Author: YeJunHao <[email protected]>
AuthorDate: Wed Sep 16 17:25:40 2026 +0800
[core] Manage manifest extra file lifecycle (#9886)
---
.../org/apache/paimon/manifest/ManifestFile.java | 8 ++
.../apache/paimon/operation/ChangelogDeletion.java | 13 ++-
.../apache/paimon/operation/FileDeletionBase.java | 20 +++-
.../paimon/operation/ManifestFileMerger.java | 2 +-
.../paimon/operation/commit/CommitCleaner.java | 6 +-
.../manifest/ManifestFileMetaSerializerTest.java | 3 +-
.../apache/paimon/manifest/ManifestFileTest.java | 22 +++++
.../paimon/manifest/ManifestIndexTestUtils.java | 99 +++++++++++++++++++
.../paimon/operation/ExpireSnapshotsTest.java | 72 ++++++++++++++
.../operation/LocalOrphanFilesCleanTest.java | 35 +++++++
.../operation/ManifestSidecarLifecycleTest.java | 106 +++++++++++++++++++++
11 files changed, 370 insertions(+), 16 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
index 0dc99a0470..9781be4b90 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
@@ -339,6 +339,14 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
};
}
+ /** Deletes an unreferenced manifest and its explicitly referenced extra
files. */
+ public void delete(ManifestFileMeta manifest) {
+ delete(manifest.fileName());
+ if (manifest.extraFiles() != null) {
+ manifest.extraFiles().forEach(this::delete);
+ }
+ }
+
/** Creator of {@link ManifestFile}. */
public static class Factory {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
index 9689f272e2..eec6090a36 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
@@ -26,7 +26,6 @@ import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.ExpireFileEntry;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
-import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -100,17 +99,17 @@ public class ChangelogDeletion extends
FileDeletionBase<Changelog> {
// base manifests
if (manifestList.exists(skippingSnapshot.baseManifestList())) {
skippingSet.add(skippingSnapshot.baseManifestList());
- manifestList.read(skippingSnapshot.baseManifestList()).stream()
- .map(ManifestFileMeta::fileName)
- .forEach(skippingSet::add);
+ manifestList
+ .read(skippingSnapshot.baseManifestList())
+ .forEach(manifest ->
addManifestToSkippingSet(skippingSet, manifest));
}
// delta manifests
if (manifestList.exists(skippingSnapshot.deltaManifestList())) {
skippingSet.add(skippingSnapshot.deltaManifestList());
-
manifestList.read(skippingSnapshot.deltaManifestList()).stream()
- .map(ManifestFileMeta::fileName)
- .forEach(skippingSet::add);
+ manifestList
+ .read(skippingSnapshot.deltaManifestList())
+ .forEach(manifest ->
addManifestToSkippingSet(skippingSet, manifest));
}
// index manifests
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
index 6494411a8c..ad544af0ea 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
@@ -327,7 +327,11 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
if (skippingSet.add(fileName)) {
manifests.add(fileName);
if (manifest.extraFiles() != null) {
- manifests.addAll(manifest.extraFiles());
+ for (String extraFile : manifest.extraFiles()) {
+ if (skippingSet.add(extraFile)) {
+ manifests.add(extraFile);
+ }
+ }
}
}
}
@@ -489,9 +493,9 @@ public abstract class FileDeletionBase<T extends Snapshot> {
// data manifests
skippingSet.add(skippingSnapshot.baseManifestList());
skippingSet.add(skippingSnapshot.deltaManifestList());
- manifestList.readDataManifests(skippingSnapshot).stream()
- .map(ManifestFileMeta::fileName)
- .forEach(skippingSet::add);
+ manifestList
+ .readDataManifests(skippingSnapshot)
+ .forEach(manifest -> addManifestToSkippingSet(skippingSet,
manifest));
// index manifests
String indexManifest = skippingSnapshot.indexManifest();
@@ -511,6 +515,14 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
return skippingSet;
}
+ protected static void addManifestToSkippingSet(
+ Set<String> skippingSet, ManifestFileMeta manifest) {
+ skippingSet.add(manifest.fileName());
+ if (manifest.extraFiles() != null) {
+ skippingSet.addAll(manifest.extraFiles());
+ }
+ }
+
private boolean tryDeleteEmptyDirectory(Path path) {
try {
fileIO.delete(path, false);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index 79d21e025e..eff0d00908 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -95,7 +95,7 @@ public class ManifestFileMerger {
// exception occurs, clean up and rethrow
for (ManifestFileMeta manifest : newFilesForAbort) {
try {
- manifestFile.delete(manifest.fileName());
+ manifestFile.delete(manifest);
} catch (Throwable cleanupFailure) {
primaryFailure =
ExceptionUtils.firstOrSuppressed(cleanupFailure,
primaryFailure);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java
index a24b5c4c6e..735a706937 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java
@@ -52,14 +52,14 @@ public class CommitCleaner {
String newIndexManifest) {
if (deltaManifestList != null) {
for (ManifestFileMeta manifest :
manifestList.read(deltaManifestList.getKey())) {
- manifestFile.delete(manifest.fileName());
+ manifestFile.delete(manifest);
}
manifestList.delete(deltaManifestList.getKey());
}
if (changelogManifestList != null) {
for (ManifestFileMeta manifest :
manifestList.read(changelogManifestList.getKey())) {
- manifestFile.delete(manifest.fileName());
+ manifestFile.delete(manifest);
}
manifestList.delete(changelogManifestList.getKey());
}
@@ -80,7 +80,7 @@ public class CommitCleaner {
.collect(Collectors.toSet());
for (ManifestFileMeta suspect : mergeAfterManifests) {
if (!oldMetaSet.contains(suspect.fileName())) {
- manifestFile.delete(suspect.fileName());
+ manifestFile.delete(suspect);
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
index 46ff2067d2..04f41aa820 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
@@ -53,7 +53,8 @@ public class ManifestFileMetaSerializerTest extends
ObjectSerializerTestBase<Man
Arrays.asList(
null,
Collections.<String>emptyList(),
- Arrays.asList("extra-1", "extra-2"))) {
+ Arrays.asList("extra-1", "extra-2"),
+ Arrays.asList("partition-index", "manifest" +
ManifestSidecar.SUFFIX))) {
ManifestFileMeta meta =
new ManifestFileMeta(
original.fileName(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
index 5fb0a7e049..50097b041a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
@@ -100,6 +100,28 @@ public class ManifestFileTest {
assertThat(actualEntries).isEqualTo(entries);
}
+ @Test
+ void testDeleteManifestAndOnlyReferencedExtraFiles() throws Exception {
+ ManifestFile manifests = createManifestFile(tempDir.toString());
+ ManifestFileMeta meta =
manifests.write(Collections.singletonList(gen.next())).get(0);
+ java.nio.file.Path manifestDir = tempDir.resolve("manifest");
+ String sidecar = "custom-index" + ManifestSidecar.SUFFIX;
+ String extra = "other-extra";
+ String unreferenced = meta.fileName() + ManifestSidecar.SUFFIX;
+ for (String name : Arrays.asList(sidecar, extra, unreferenced)) {
+ Files.createFile(manifestDir.resolve(name));
+ }
+ manifests.delete(
+ ManifestIndexTestUtils.withExtraFiles(meta,
Arrays.asList(sidecar, extra)));
+
assertThat(Files.exists(manifestDir.resolve(meta.fileName()))).isFalse();
+ assertThat(Files.exists(manifestDir.resolve(sidecar))).isFalse();
+ assertThat(Files.exists(manifestDir.resolve(extra))).isFalse();
+ assertThat(Files.exists(manifestDir.resolve(unreferenced))).isTrue();
+ // Deleting already removed files is harmless.
+ manifests.delete(
+ ManifestIndexTestUtils.withExtraFiles(meta,
Arrays.asList(sidecar, extra)));
+ }
+
@Test
void testWriteManifestFileToExplicitPath() throws Exception {
List<ManifestEntry> entries = generateData();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java
new file mode 100644
index 0000000000..32a94c9a5b
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.manifest;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.SnapshotManager;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Synthetic index references for manifest serialization and lifecycle tests.
*/
+public final class ManifestIndexTestUtils {
+ private ManifestIndexTestUtils() {}
+
+ public static ManifestFileMeta withIndexFileName(ManifestFileMeta meta,
String indexFileName) {
+ return withExtraFiles(
+ meta, indexFileName == null ? null :
Collections.singletonList(indexFileName));
+ }
+
+ public static ManifestFileMeta withExtraFiles(ManifestFileMeta meta,
List<String> extraFiles) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ meta.fileSize(),
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ meta.minBucket(),
+ meta.maxBucket(),
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ meta.totalBuckets(),
+ extraFiles);
+ }
+
+ /** Replaces only synthetic snapshot fixtures, using newly written
manifest lists. */
+ public static void registerIndexReferences(FileStore<?> store, long
snapshotId)
+ throws IOException {
+ SnapshotManager manager = store.snapshotManager();
+ FileIO io = manager.fileIO();
+ Path snapshotPath = manager.snapshotPath(snapshotId);
+ ObjectNode snapshot =
+ (ObjectNode)
+ JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(
+ io.readFileUtf8(snapshotPath));
+ ManifestList lists = store.manifestListFactory().create();
+ for (String field :
+ new String[] {"baseManifestList", "deltaManifestList",
"changelogManifestList"}) {
+ JsonNode value = snapshot.get(field);
+ if (value == null || value.isNull()) {
+ continue;
+ }
+ List<ManifestFileMeta> indexed = new ArrayList<>();
+ for (ManifestFileMeta meta : lists.read(value.asText())) {
+ // Deliberately use a name which cannot be derived by
appending the sidecar suffix.
+ String name = "index-for-" + meta.fileName() +
ManifestSidecar.SUFFIX;
+ Path index = store.pathFactory().toManifestFilePath(name);
+ if (!io.exists(index)) {
+ // GC treats index bytes as opaque; unsupported/partial
files are still owned.
+ io.newOutputStream(index, false).close();
+ }
+ indexed.add(withIndexFileName(meta, name));
+ }
+ Pair<String, Long> replacement = lists.write(indexed);
+ snapshot.put(field, replacement.getLeft());
+ snapshot.put(field + "Size", replacement.getRight());
+ }
+ io.overwriteFileUtf8(
+ snapshotPath,
JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.writeValueAsString(snapshot));
+ manager.invalidateCache();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
index 24a6e0eaea..957e4b7317 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
@@ -38,8 +38,11 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestIndexTestUtils;
+import org.apache.paimon.manifest.ManifestSidecar;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.options.ExpireConfig;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
@@ -854,6 +857,75 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ @Test
+ void testSidecarsFollowSnapshotAndTagRetention() throws Exception {
+
store.options().toConfiguration().set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 2);
+ store.options()
+ .toConfiguration()
+ .set(CoreOptions.MANIFEST_TARGET_FILE_SIZE,
MemorySize.parse("8 mb"));
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(8, allData, snapshotPositions);
+ int latest =
requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+ Set<Path> manifests = new HashSet<>();
+ Set<Path> retainedManifests = new HashSet<>();
+ for (int snapshotId = 1; snapshotId <= latest; snapshotId++) {
+ rewriteSnapshotTime(snapshotId, 0);
+ ManifestIndexTestUtils.registerIndexReferences(store, snapshotId);
+ snapshotManager.invalidateCache();
+ for (ManifestFileMeta meta :
+ store.manifestListFactory()
+ .create()
+
.readDataManifests(snapshotManager.snapshot(snapshotId))) {
+ Path manifest =
store.pathFactory().toManifestFilePath(meta.fileName());
+ manifests.add(manifest);
+ if (snapshotId == 3 || snapshotId == latest) {
+ retainedManifests.add(manifest);
+ }
+ }
+ }
+ store.newTagManager()
+ .createTag(
+ snapshotManager.snapshot(3),
+ "keep-sidecars",
+ store.options().tagDefaultTimeRetained(),
+ Collections.emptyList(),
+ false);
+ Set<Path> expiredManifests = new HashSet<>(manifests);
+ expiredManifests.removeAll(retainedManifests);
+ assertThat(expiredManifests).isNotEmpty();
+ ExpireSnapshotsImpl expire =
+ (ExpireSnapshotsImpl)
store.newExpire(expireAllButLatestConfig());
+ expire.setCurrentTimeMillis(() -> 1000L);
+ expire.expire();
+ for (Path manifest : manifests) {
+ boolean retained = retainedManifests.contains(manifest);
+ assertThat(fileIO.exists(manifest)).as("manifest %s",
manifest).isEqualTo(retained);
+ assertThat(
+ fileIO.exists(
+ new Path(
+ manifest.getParent(),
+ "index-for-"
+ + manifest.getName()
+ + ManifestSidecar.SUFFIX)))
+ .as("sidecar for %s", manifest)
+ .isEqualTo(retained);
+ }
+ for (ManifestFileMeta meta :
+ store.manifestListFactory()
+ .create()
+ .readDataManifests(
+ store.newTagManager()
+ .getOrThrow("keep-sidecars")
+ .trimToSnapshot())) {
+ assertThat(
+ fileIO.exists(
+ store.pathFactory()
+
.toManifestFilePath(ManifestSidecar.fileName(meta))))
+ .isTrue();
+ }
+ }
+
@Test
public void
testExpireWithTagsAndConcurrentPlanningKeepsTaggedSnapshotsReadable()
throws Exception {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
index 9e3bf05b69..49e22ee551 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java
@@ -32,7 +32,9 @@ import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestIndexTestUtils;
import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.manifest.ManifestSidecar;
import org.apache.paimon.mergetree.compact.ConcatRecordReader;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.ReaderSupplier;
@@ -158,6 +160,39 @@ public class LocalOrphanFilesCleanTest {
normallyRemoving(tablePath);
}
+ @Test
+ void testOrphanCleanupProtectsReferencedSidecars() throws Exception {
+ commit(Collections.singletonList(TestPojo.next()));
+ ManifestIndexTestUtils.registerIndexReferences(
+ table.store(), table.snapshotManager().latestSnapshotId());
+ table.snapshotManager().invalidateCache();
+ table.createTag("sidecar-tag",
table.snapshotManager().latestSnapshotId());
+ List<Path> sidecars = new ArrayList<>();
+ List<Path> unreferenced = new ArrayList<>();
+ for (ManifestFileMeta meta :
+ table.store()
+ .manifestListFactory()
+ .create()
+
.readDataManifests(table.snapshotManager().latestSnapshot())) {
+ Path sidecar = new Path(manifestDir,
ManifestSidecar.fileName(meta));
+ sidecars.add(sidecar);
+ Path guessed = new Path(manifestDir, meta.fileName() +
ManifestSidecar.SUFFIX);
+ fileIO.newOutputStream(guessed, false).close();
+ unreferenced.add(guessed);
+ }
+ Path orphan = new Path(manifestDir, "manifest-orphan" +
ManifestSidecar.SUFFIX);
+ fileIO.newOutputStream(orphan, false).close();
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ assertThat(fileIO.exists(orphan)).isFalse();
+ assertThat(sidecars).isNotEmpty();
+ for (Path sidecar : sidecars) {
+ assertThat(fileIO.exists(sidecar)).isTrue();
+ }
+ for (Path guessed : unreferenced) {
+ assertThat(fileIO.exists(guessed)).isFalse();
+ }
+ }
+
@Test
public void testKeepManagedBlobPack() throws Exception {
commit(Collections.singletonList(TestPojo.next()));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestSidecarLifecycleTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestSidecarLifecycleTest.java
new file mode 100644
index 0000000000..be9e199427
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestSidecarLifecycleTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.manifest.IndexManifestFile;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.commit.CommitCleaner;
+import org.apache.paimon.utils.Pair;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/** Tests ownership of manifest extra files during expiration and aborted
commits. */
+class ManifestSidecarLifecycleTest {
+
+ @Test
+ void retainedManifestProtectsSharedExtraFiles() {
+ ManifestFileMeta retained = manifest("retained",
"shared.avro.sidecar", "retained-extra");
+ ManifestFileMeta expired = manifest("expired", "shared.avro.sidecar",
"expired-extra");
+ FileDeletionBase<?> deletion = mock(FileDeletionBase.class,
CALLS_REAL_METHODS);
+ doReturn(Arrays.asList(retained,
expired)).when(deletion).tryReadManifestList("old-list");
+ Set<String> skipping = new HashSet<>();
+ FileDeletionBase.addManifestToSkippingSet(skipping, retained);
+ Set<String> removed = new HashSet<>();
+ deletion.collectUnusedManifestList("old-list", skipping, removed);
+ assertThat(removed).containsExactlyInAnyOrder("expired",
"expired-extra", "old-list");
+ }
+
+ @Test
+ void failedCommitDeletesOnlyNewManifestsWithTheirMetadata() {
+ ManifestList lists = mock(ManifestList.class);
+ ManifestFile files = mock(ManifestFile.class);
+ IndexManifestFile indexes = mock(IndexManifestFile.class);
+ ManifestFileMeta reused = manifest("reused", "reused.avro.sidecar");
+ ManifestFileMeta added = manifest("added", "added.avro.sidecar");
+ CommitCleaner cleaner = new CommitCleaner(lists, files, indexes);
+ cleaner.cleanUpNoReuseTmpManifests(
+ Pair.of("new-base-list", 1L),
+ Collections.singletonList(reused),
+ Arrays.asList(reused, added));
+ verify(files).delete(added);
+ verify(files, never()).delete(reused);
+ verify(lists).delete("new-base-list");
+ verifyNoInteractions(indexes);
+ }
+
+ @Test
+ void failedCommitDeletesDeltaAndChangelogManifestExtras() {
+ ManifestList lists = mock(ManifestList.class);
+ ManifestFile files = mock(ManifestFile.class);
+ IndexManifestFile indexes = mock(IndexManifestFile.class);
+ ManifestFileMeta delta = manifest("delta", "delta.avro.sidecar");
+ ManifestFileMeta changelog = manifest("changelog",
"changelog.avro.sidecar");
+
when(lists.read("delta-list")).thenReturn(Collections.singletonList(delta));
+
when(lists.read("changelog-list")).thenReturn(Collections.singletonList(changelog));
+ new CommitCleaner(lists, files, indexes)
+ .cleanUpReuseTmpManifests(
+ Pair.of("delta-list", 1L),
+ Pair.of("changelog-list", 1L),
+ "same-index",
+ "same-index");
+ verify(files).delete(delta);
+ verify(files).delete(changelog);
+ verify(lists).delete("delta-list");
+ verify(lists).delete("changelog-list");
+ verifyNoInteractions(indexes);
+ }
+
+ private static ManifestFileMeta manifest(String name, String... extras) {
+ ManifestFileMeta meta = mock(ManifestFileMeta.class);
+ when(meta.fileName()).thenReturn(name);
+ when(meta.extraFiles()).thenReturn(Arrays.asList(extras));
+ return meta;
+ }
+}
