optimize orphan files cleaner

Author: chongchongxiao

src/paimon/core/operation/metrics/clean_metrics.h

@@ -27,6 +27,7 @@ class CleanMetrics {
     static constexpr char CLEAN_LIST_FILE_STATUS_DURATION[] = "listFileStatusDuration";
     static constexpr char CLEAN_LIST_FILE_STATUS_TASKS[] = "listFileStatusTasks";
     static constexpr char CLEAN_LIST_USED_FILES_DURATION[] = "listUsedFilesDuration";
+    static constexpr char CLEAN_SNAPSHOT_FILES[] = "snapshotFiles";
     static constexpr char CLEAN_USED_FILES[] = "usedFiles";
     static constexpr char CLEAN_SCAN_ORPHAN_FILES_DURATION[] = "scanOrphanFilesDuration";
     static constexpr char CLEAN_ORPHAN_FILES[] = "orphanFiles";

src/paimon/core/operation/orphan_files_cleaner_impl.cpp

@@ -245,15 +245,12 @@ Result<std::set<std::string>> OrphanFilesCleanerImpl::GetUsedFiles() const {
     used_files.insert(SnapshotManager::LATEST);
     Duration duration;
     PAIMON_ASSIGN_OR_RAISE(std::vector<Snapshot> snapshots, snapshot_manager_->GetAllSnapshots());
+    std::vector<std::future<Result<std::set<std::string>>>> used_files_futures;
+    std::vector<Result<std::set<std::string>>> used_files_results;
+    ScopeGuard guard([&used_files_futures, &used_files_results]() {
+        used_files_results = CollectAll(used_files_futures);
+    });
     for (const auto& snapshot : snapshots) {
-        used_files.insert(SnapshotManager::SNAPSHOT_PREFIX + std::to_string(snapshot.Id()));
-        used_files.insert(snapshot.BaseManifestList());
-        used_files.insert(snapshot.DeltaManifestList());
-        std::vector<ManifestFileMeta> manifests;
-        PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.BaseManifestList(),
-                                                             /*filter=*/nullptr, &manifests));
-        PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.DeltaManifestList(),
-                                                             /*filter=*/nullptr, &manifests));
         const std::optional<std::string>& changelog_manifest_list =
             snapshot.ChangelogManifestList();
         if (changelog_manifest_list) {
@@ -266,18 +263,46 @@ Result<std::set<std::string>> OrphanFilesCleanerImpl::GetUsedFiles() const {
             // TODO(jinli.zjw): support IndexManifestEntry and add tests
             // used_files.insert(index_manifest_name.value());
         }
-        for (const auto& manifest : manifests) {
-            used_files.insert(manifest.FileName());
-            std::vector<ManifestEntry> manifest_entries;
-            PAIMON_RETURN_NOT_OK(manifest_file_->ReadIfFileExist(
-                manifest.FileName(), /*filter=*/nullptr, &manifest_entries));
-            for (const auto& manifest_entry : manifest_entries) {
-                used_files.insert(manifest_entry.FileName());
-            }
-        }
+
+        used_files_futures.emplace_back(
+            Via(executor_.get(), [this, snapshot] { return GetUsedFilesBySnapshot(snapshot); }));
+    }
+
+    for (const auto& used_files_result : used_files_results) {
+        PAIMON_RETURN_NOT_OK(used_files_result);
+        used_files.insert(used_files_result.value().begin(), used_files_result.value().end());
     }
+
     metrics_->SetCounter(CleanMetrics::CLEAN_LIST_USED_FILES_DURATION, duration.Get());
     metrics_->SetCounter(CleanMetrics::CLEAN_USED_FILES, static_cast<uint64_t>(used_files.size()));
+    metrics_->SetCounter(CleanMetrics::CLEAN_SNAPSHOT_FILES,
+                         static_cast<uint64_t>(snapshots.size()));
+    return used_files;
+}
+
+Result<std::set<std::string>> OrphanFilesCleanerImpl::GetUsedFilesBySnapshot(
+    const Snapshot& snapshot) const {
+    std::set<std::string> used_files;
+
+    used_files.insert(SnapshotManager::SNAPSHOT_PREFIX + std::to_string(snapshot.Id()));
+    used_files.insert(snapshot.BaseManifestList());
+    used_files.insert(snapshot.DeltaManifestList());
+    std::vector<ManifestFileMeta> manifests;
+    PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.BaseManifestList(),
+                                                         /*filter=*/nullptr, &manifests));
+    PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.DeltaManifestList(),
+                                                         /*filter=*/nullptr, &manifests));
+
+    for (const auto& manifest : manifests) {
+        used_files.insert(manifest.FileName());
+        std::vector<ManifestEntry> manifest_entries;
+        PAIMON_RETURN_NOT_OK(manifest_file_->ReadIfFileExist(
+            manifest.FileName(), /*filter=*/nullptr, &manifest_entries));
+        for (const auto& manifest_entry : manifest_entries) {
+            used_files.insert(manifest_entry.FileName());
+        }
+    }
+
     return used_files;
 }
 

src/paimon/core/operation/orphan_files_cleaner_impl.h

@@ -74,6 +74,7 @@ class OrphanFilesCleanerImpl : public OrphanFilesCleaner {
         const std::string& path) const;
     std::set<std::string> ListFileDirs(const std::string& path, int32_t max_level) const;
     Result<std::set<std::string>> GetUsedFiles() const;
+    Result<std::set<std::string>> GetUsedFilesBySnapshot(const Snapshot& snapshot) const;
     static bool SupportToClean(const std::string& file_name);
 
  private: