This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 301713fe6cb8106be9caf353048aff6443f81bb0 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; + } +}
