[core] Manage manifest extra file lifecycle

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java

@@ -339,6 +339,14 @@ public Path toPath(String fileName) {
         };
     }
 
+    /** 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 {
 

paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java

@@ -26,7 +26,6 @@
 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 Set<String> manifestSkippingSet(List<Snapshot> skippingSnapshots) {
             // 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

paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java

@@ -327,7 +327,11 @@ protected void collectUnusedManifestList(
             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 @@ private Set<String> manifestSkippingSet(Snapshot skippingSnapshot) {
         // 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 @@ private Set<String> manifestSkippingSet(Snapshot skippingSnapshot) {
         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);

paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java

@@ -95,7 +95,7 @@ static List<ManifestFileMeta> merge(
             // 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);

paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java

@@ -52,14 +52,14 @@ public void cleanUpReuseTmpManifests(
             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 void cleanUpNoReuseTmpManifests(
                         .collect(Collectors.toSet());
         for (ManifestFileMeta suspect : mergeAfterManifests) {
             if (!oldMetaSet.contains(suspect.fileName())) {
-                manifestFile.delete(suspect.fileName());
+                manifestFile.delete(suspect);
             }
         }
     }

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java

@@ -53,7 +53,8 @@ void testExtraFiles() throws IOException {
                 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(),

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java

@@ -100,6 +100,28 @@ public void testWriteAndReadManifestFile() {
         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();

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();
+    }
+}

paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java

@@ -38,8 +38,11 @@
 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 void testExpirePlansManifestsConcurrentlyWithSkippingSet() throws Excepti
         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 {

paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java

@@ -32,7 +32,9 @@
 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 void testNormallyRemoving() throws Throwable {
         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()));