Merge branch 'main' into PAIMON-110

Author: lucasfang

CMakeLists.txt

@@ -32,7 +32,7 @@ if(NOT CMAKE_BUILD_TYPE)
 endif()
 
 project(paimon
-        VERSION 0.9.0
+        VERSION 0.1.0
         DESCRIPTION "Paimon C++ Project")
 
 string(TOUPPER "${CMAKE_BUILD_TYPE}" UPPERCASE_BUILD_TYPE)

README.md

@@ -165,3 +165,7 @@ If you make improvements that could benefit all developers, please update the te
 ## License
 
 Licensed under the [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0)
+
+## Maintainership and Contributions
+
+This project is maintained by a core team from the Storage Service team at Alibaba, including [lxy-9602](https://github.com/lxy-9602) (maintainer), [lucasfang](https://github.com/lucasfang), [lszskye](https://github.com/lszskye), and [zjw1111](https://github.com/zjw1111). We sincerely appreciate contributions from the community — your feedback and patches are welcome and highly valued. For any questions, feature proposals, or code reviews, please feel free to reach out to us directly.

docs/source/conf.py

@@ -113,7 +113,7 @@
 # The master toctree document.
 master_doc = "index"
 
-version = "0.9.0"
+version = "0.1.0"
 
 html_theme_options = {
     "show_toc_level": 2,

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,39 +245,67 @@ 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());
-    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) {
-            used_files.insert(changelog_manifest_list.value());
-            return Status::NotImplemented("OrphanFilesCleaner do not support clean changelog");
-        }
-        const std::optional<std::string>& index_manifest_name = snapshot.IndexManifest();
-        if (index_manifest_name) {
-            return Status::NotImplemented("OrphanFilesCleaner do not support clean index manifest");
-            // 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());
+    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) {
+            const std::optional<std::string>& changelog_manifest_list =
+                snapshot.ChangelogManifestList();
+            if (changelog_manifest_list) {
+                used_files.insert(changelog_manifest_list.value());
+                return Status::NotImplemented("OrphanFilesCleaner do not support clean changelog");
             }
+            const std::optional<std::string>& index_manifest_name = snapshot.IndexManifest();
+            if (index_manifest_name) {
+                return Status::NotImplemented(
+                    "OrphanFilesCleaner do not support clean index manifest");
+                // TODO(jinli.zjw): support IndexManifestEntry and add tests
+                // used_files.insert(index_manifest_name.value());
+            }
+
+            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: