[core] Fix that snapshot expire might delete files used by tag mistakenly (apache#5237)

Author: JingsongLi

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

@@ -340,9 +340,11 @@ public Predicate<ExpireFileEntry> createDataFileSkipperForTags(
         if (index >= 0) {
             Snapshot previousTag = taggedSnapshots.get(index);
             if (previousTag.id() != cachedTag) {
-                cachedTag = previousTag.id();
+                cachedTag = 0;
                 cachedTagDataFiles.clear();
                 addMergedDataFiles(cachedTagDataFiles, previousTag);
+                // update cachedTag after read tag successfully
+                cachedTag = previousTag.id();
             }
             return entry -> containsDataFile(cachedTagDataFiles, entry);
         }
@@ -359,7 +361,7 @@ protected List<ManifestFileMeta> tryReadManifestList(String manifestListName) {
         try {
             return manifestList.read(manifestListName);
         } catch (Exception e) {
-            LOG.warn("Failed to read manifest list file " + manifestListName, e);
+            LOG.warn("Failed to read manifest list file {}", manifestListName, e);
             return Collections.emptyList();
         }
     }
@@ -371,22 +373,17 @@ protected List<ManifestFileMeta> tryReadManifestList(String manifestListName) {
     protected void addMergedDataFiles(
             Map<BinaryRow, Map<Integer, Set<String>>> dataFiles, Snapshot snapshot)
             throws IOException {
-        for (ExpireFileEntry entry : readMergedDataFiles(snapshot)) {
+        for (ExpireFileEntry entry :
+                readMergedDataFiles(manifestList.readDataManifests(snapshot))) {
             dataFiles
                     .computeIfAbsent(entry.partition(), p -> new HashMap<>())
                     .computeIfAbsent(entry.bucket(), b -> new HashSet<>())
                     .add(entry.fileName());
         }
     }
 
-    protected Collection<ExpireFileEntry> readMergedDataFiles(Snapshot snapshot)
+    protected Collection<ExpireFileEntry> readMergedDataFiles(List<ManifestFileMeta> manifests)
             throws IOException {
-        // read data manifests
-
-        List<ManifestFileMeta> manifests = tryReadManifestList(snapshot.baseManifestList());
-        manifests.addAll(tryReadManifestList(snapshot.deltaManifestList()));
-
-        // read and merge manifest entries
         Map<Identifier, ExpireFileEntry> map = new HashMap<>();
         for (ManifestFileMeta manifest : manifests) {
             List<ExpireFileEntry> entries =

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

@@ -26,6 +26,7 @@
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.manifest.ExpireFileEntry;
 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.DataFilePathFactories;
@@ -73,7 +74,10 @@ public TagDeletion(
     public void cleanUnusedDataFiles(Snapshot taggedSnapshot, Predicate<ExpireFileEntry> skipper) {
         Collection<ExpireFileEntry> manifestEntries;
         try {
-            manifestEntries = readMergedDataFiles(taggedSnapshot);
+            List<ManifestFileMeta> manifests =
+                    tryReadManifestList(taggedSnapshot.baseManifestList());
+            manifests.addAll(tryReadManifestList(taggedSnapshot.deltaManifestList()));
+            manifestEntries = readMergedDataFiles(manifests);
         } catch (IOException e) {
             LOG.info("Skip data file clean for the tag of id {}.", taggedSnapshot.id(), e);
             return;