From e710b5c3971acfe78c7bdba192ab6fd6f1dfc0de Mon Sep 17 00:00:00 2001
From: Aitozi <aitozi@apache.org>
Date: Thu, 18 Apr 2024 12:31:21 +0800
Subject: [PATCH] fix

---
 .../paimon/operation/OrphanFilesClean.java    | 33 +++++++------------
 .../java/org/apache/paimon/TestFileStore.java |  6 ++--
 2 files changed, 15 insertions(+), 24 deletions(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
index 69271ed625602..16d854ea43a42 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -19,7 +19,6 @@
 package org.apache.paimon.operation;
 
 import org.apache.paimon.Changelog;
-import org.apache.paimon.CoreOptions;
 import org.apache.paimon.FileStore;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.VisibleForTesting;
@@ -96,13 +95,9 @@ public class OrphanFilesClean {
     private int deletedFilesNum = 0;
     private final List<Path> deleteFiles;
     private long olderThanMillis = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1);
-    private final boolean produceChangelog;
 
     public OrphanFilesClean(FileStoreTable table) {
         this.snapshotManager = table.snapshotManager();
-        this.produceChangelog =
-                new CoreOptions(table.options()).changelogProducer()
-                        != CoreOptions.ChangelogProducer.NONE;
         this.tagManager = table.tagManager();
         this.fileIO = table.fileIO();
         this.location = table.location();
@@ -220,6 +215,7 @@ private List<String> getUsedFilesForChangelog(Changelog changelog) {
         List<ManifestFileMeta> manifestFileMetas = new ArrayList<>();
         try {
             // try to read manifests
+            // changelog manifest
             List<ManifestFileMeta> changelogManifest = new ArrayList<>();
             if (changelog.changelogManifestList() != null) {
                 files.add(changelog.changelogManifestList());
@@ -233,10 +229,20 @@ private List<String> getUsedFilesForChangelog(Changelog changelog) {
                 }
             }
 
+            // base manifest
             if (manifestList.exists(changelog.baseManifestList())) {
                 files.add(changelog.baseManifestList());
+                List<ManifestFileMeta> baseManifest =
+                        retryReadingFiles(
+                                () ->
+                                        manifestList.readWithIOException(
+                                                changelog.baseManifestList()));
+                if (baseManifest != null) {
+                    manifestFileMetas.addAll(baseManifest);
+                }
             }
 
+            // delta manifest
             List<ManifestFileMeta> deltaManifest = null;
             if (manifestList.exists(changelog.deltaManifestList())) {
                 files.add(changelog.deltaManifestList());
@@ -245,26 +251,11 @@ private List<String> getUsedFilesForChangelog(Changelog changelog) {
                                 () ->
                                         manifestList.readWithIOException(
                                                 changelog.deltaManifestList()));
-                // delta
                 if (deltaManifest != null) {
                     manifestFileMetas.addAll(deltaManifest);
                 }
             }
 
-            List<ManifestFileMeta> baseManifest;
-            if (manifestList.exists(changelog.changelogManifestList())) {
-                files.add(changelog.changelogManifestList());
-                baseManifest =
-                        retryReadingFiles(
-                                () ->
-                                        manifestList.readWithIOException(
-                                                changelog.changelogManifestList()));
-                // delta
-                if (baseManifest != null) {
-                    manifestFileMetas.addAll(baseManifest);
-                }
-            }
-
             files.addAll(
                     manifestFileMetas.stream()
                             .map(ManifestFileMeta::fileName)
@@ -279,7 +270,7 @@ private List<String> getUsedFilesForChangelog(Changelog changelog) {
                                 : changelogManifest.stream()
                                         .map(ManifestFileMeta::fileName)
                                         .collect(Collectors.toList()));
-            } else if (changelog.commitKind() == Snapshot.CommitKind.APPEND) {
+            } else {
                 manifestFileName.addAll(
                         deltaManifest == null
                                 ? new ArrayList<>()
diff --git a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
index 2be95dde2a497..ab90c19fa8dd0 100644
--- a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
+++ b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
@@ -655,9 +655,9 @@ private static Set<Path> getSnapshotFileInUse(
                             entry.file().fileName()));
         }
 
-        // Add 'DELETE' 'APPEND' file
-        // These 'delete' files can be merged by the plan#splits, so it's not showed in the entries
-        // above.
+        // Add 'DELETE' 'APPEND' file in snapshot
+        // These 'delete' files can be merged by the plan#splits,
+        // so it's not shown in the entries above.
         // In other words, these files are not used (by snapshot or changelog) now,
         // but it can only be cleaned after this snapshot expired, so we should add it to the file
         // use list.