Copilot commented on code in PR #57: URL: https://github.com/apache/paimon-cpp/pull/57#discussion_r3371990039
########## src/paimon/core/operation/orphan_files_cleaner_impl.cpp: ########## @@ -0,0 +1,314 @@ +/* + * 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. + */ + +#include "paimon/core/operation/orphan_files_cleaner_impl.h" + +#include <algorithm> +#include <future> +#include <map> +#include <optional> +#include <queue> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/operation/metrics/clean_metrics.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/duration.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/status.h" + +namespace paimon { +class MemoryPool; + +const int64_t OrphanFilesCleanerImpl::MIN_VALID_FILE_MODIFICATION_MS = 10 * 1000 * 1000 * 1000L; + +OrphanFilesCleanerImpl::OrphanFilesCleanerImpl( + const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor, + const std::shared_ptr<arrow::Schema>& schema, const std::string& root_path, + const CoreOptions& options, const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::vector<std::string>& partition_keys, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<ManifestList>& manifest_list, int64_t older_than_ms, + std::function<bool(const std::string&)> should_be_retained) + : memory_pool_(memory_pool), + executor_(executor), + schema_(schema), + root_path_(root_path), + options_(options), + fs_(options.GetFileSystem()), + snapshot_manager_(snapshot_manager), + partition_keys_(partition_keys), + manifest_file_(manifest_file), + manifest_list_(manifest_list), + older_than_ms_(older_than_ms), + should_be_retained_(should_be_retained), + metrics_(std::make_shared<MetricsImpl>()) {} + +bool OrphanFilesCleanerImpl::SupportToClean(const std::string& file_name) { + static std::vector<std::pair<std::string, std::string>> supported_pattern = { + {"manifest-", ""}, {"manifest-list-", ""}, {".", ".tmp"}}; + for (const auto& pattern : supported_pattern) { + if (StringUtils::StartsWith(file_name, pattern.first) && + StringUtils::EndsWith(file_name, pattern.second)) { + return true; + } + } + static std::vector<std::string> supported_formats = {".orc", ".parquet", ".avro", ".lance"}; + for (const auto& format : supported_formats) { + if (StringUtils::StartsWith(file_name, "data-") && + StringUtils::EndsWith(file_name, format)) { + return true; + } + } + return false; +} + +Result<std::set<std::string>> OrphanFilesCleanerImpl::Clean() { + Duration main_duration; + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "tag")).empty()) { + return Status::NotImplemented("OrphanFilesCleaner do not support cleaning table with tag"); + } + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "branch")).empty()) { + return Status::NotImplemented( + "OrphanFilesCleaner do not support cleaning table with branch"); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> all_dirs, ListPaimonFileDirs()); + std::vector<std::future<std::vector<std::unique_ptr<FileStatus>>>> file_statuses_futures; + for (const auto& dir : all_dirs) { + file_statuses_futures.push_back( + Via(executor_.get(), [this, dir] { return TryBestListingDirs(dir); })); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> used_file_names, GetUsedFiles()); + + Duration duration; Review Comment: `file_statuses_duration` is captured immediately after constructing `Duration duration` and calling `Reset()`, which is likely to record ~0 rather than the directory listing time. If this metric is intended to measure the time spent listing file statuses, move the timing boundary to wrap the `CollectAll(file_statuses_futures)` (or the listing phase) and record the elapsed time after that step completes. ########## src/paimon/core/operation/orphan_files_cleaner_impl.cpp: ########## @@ -0,0 +1,314 @@ +/* + * 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. + */ + +#include "paimon/core/operation/orphan_files_cleaner_impl.h" + +#include <algorithm> +#include <future> +#include <map> +#include <optional> +#include <queue> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/operation/metrics/clean_metrics.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/duration.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/status.h" + +namespace paimon { +class MemoryPool; + +const int64_t OrphanFilesCleanerImpl::MIN_VALID_FILE_MODIFICATION_MS = 10 * 1000 * 1000 * 1000L; + +OrphanFilesCleanerImpl::OrphanFilesCleanerImpl( + const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor, + const std::shared_ptr<arrow::Schema>& schema, const std::string& root_path, + const CoreOptions& options, const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::vector<std::string>& partition_keys, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<ManifestList>& manifest_list, int64_t older_than_ms, + std::function<bool(const std::string&)> should_be_retained) + : memory_pool_(memory_pool), + executor_(executor), + schema_(schema), + root_path_(root_path), + options_(options), + fs_(options.GetFileSystem()), + snapshot_manager_(snapshot_manager), + partition_keys_(partition_keys), + manifest_file_(manifest_file), + manifest_list_(manifest_list), + older_than_ms_(older_than_ms), + should_be_retained_(should_be_retained), + metrics_(std::make_shared<MetricsImpl>()) {} + +bool OrphanFilesCleanerImpl::SupportToClean(const std::string& file_name) { + static std::vector<std::pair<std::string, std::string>> supported_pattern = { + {"manifest-", ""}, {"manifest-list-", ""}, {".", ".tmp"}}; + for (const auto& pattern : supported_pattern) { + if (StringUtils::StartsWith(file_name, pattern.first) && + StringUtils::EndsWith(file_name, pattern.second)) { + return true; + } + } + static std::vector<std::string> supported_formats = {".orc", ".parquet", ".avro", ".lance"}; + for (const auto& format : supported_formats) { + if (StringUtils::StartsWith(file_name, "data-") && + StringUtils::EndsWith(file_name, format)) { + return true; + } + } + return false; +} + +Result<std::set<std::string>> OrphanFilesCleanerImpl::Clean() { + Duration main_duration; + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "tag")).empty()) { + return Status::NotImplemented("OrphanFilesCleaner do not support cleaning table with tag"); + } + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "branch")).empty()) { + return Status::NotImplemented( + "OrphanFilesCleaner do not support cleaning table with branch"); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> all_dirs, ListPaimonFileDirs()); + std::vector<std::future<std::vector<std::unique_ptr<FileStatus>>>> file_statuses_futures; + for (const auto& dir : all_dirs) { + file_statuses_futures.push_back( + Via(executor_.get(), [this, dir] { return TryBestListingDirs(dir); })); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> used_file_names, GetUsedFiles()); + + Duration duration; + std::set<std::string> need_to_deletes; + std::vector<std::future<void>> futures; + ScopeGuard guard([&futures]() { Wait(futures); }); + uint64_t file_statuses_duration = duration.Reset(); + for (const auto& file_statuses : CollectAll(file_statuses_futures)) { Review Comment: `file_statuses_duration` is captured immediately after constructing `Duration duration` and calling `Reset()`, which is likely to record ~0 rather than the directory listing time. If this metric is intended to measure the time spent listing file statuses, move the timing boundary to wrap the `CollectAll(file_statuses_futures)` (or the listing phase) and record the elapsed time after that step completes. ########## src/paimon/core/operation/orphan_files_cleaner_impl.h: ########## @@ -0,0 +1,100 @@ +/* + * 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. + */ + +#pragma once + +#include <cstdint> +#include <functional> +#include <map> +#include <memory> +#include <set> +#include <string> +#include <vector> + +#include "paimon/core/core_options.h" +#include "paimon/core/snapshot.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/metrics.h" +#include "paimon/orphan_files_cleaner.h" +#include "paimon/result.h" + +struct ArrowSchema; + +namespace arrow { +class Schema; +} // namespace arrow + +namespace paimon { + +class SnapshotManager; +class FileStorePathFactory; +class FileSystem; +class ManifestFile; +class ManifestList; +class Executor; +class MemoryPool; + +class OrphanFilesCleanerImpl : public OrphanFilesCleaner { + public: + OrphanFilesCleanerImpl(const std::shared_ptr<MemoryPool>& memory_pool, + const std::shared_ptr<Executor>& executor, + const std::shared_ptr<arrow::Schema>& schema, + const std::string& root_path, const CoreOptions& options, + const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::vector<std::string>& partition_keys, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<ManifestList>& manifest_list, + int64_t older_than_ms, + std::function<bool(const std::string&)> should_be_retained); + + Result<std::set<std::string>> Clean() override; + + std::shared_ptr<Metrics> GetMetrics() const override { + return metrics_; + } + + private: + Result<std::set<std::string>> ListPaimonFileDirs() const; + std::vector<std::unique_ptr<FileStatus>> TryBestListingDirs(const std::string& path) const; + std::vector<std::unique_ptr<BasicFileStatus>> MinimalTryBestListingDirs( + 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); Review Comment: `SupportToClean` is declared under `private:` but the new test calls `OrphanFilesCleanerImpl::SupportToClean(...)` directly. This will fail to compile due to private access. Make this helper `public` (or move the test to validate behavior through the public `Clean()` API instead of calling private internals). ########## src/paimon/core/operation/orphan_files_cleaner_impl.cpp: ########## @@ -0,0 +1,314 @@ +/* + * 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. + */ + +#include "paimon/core/operation/orphan_files_cleaner_impl.h" + +#include <algorithm> +#include <future> +#include <map> +#include <optional> +#include <queue> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/operation/metrics/clean_metrics.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/duration.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/status.h" + +namespace paimon { +class MemoryPool; + +const int64_t OrphanFilesCleanerImpl::MIN_VALID_FILE_MODIFICATION_MS = 10 * 1000 * 1000 * 1000L; + +OrphanFilesCleanerImpl::OrphanFilesCleanerImpl( + const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor, + const std::shared_ptr<arrow::Schema>& schema, const std::string& root_path, + const CoreOptions& options, const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::vector<std::string>& partition_keys, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<ManifestList>& manifest_list, int64_t older_than_ms, + std::function<bool(const std::string&)> should_be_retained) + : memory_pool_(memory_pool), + executor_(executor), + schema_(schema), + root_path_(root_path), + options_(options), + fs_(options.GetFileSystem()), + snapshot_manager_(snapshot_manager), + partition_keys_(partition_keys), + manifest_file_(manifest_file), + manifest_list_(manifest_list), + older_than_ms_(older_than_ms), + should_be_retained_(should_be_retained), + metrics_(std::make_shared<MetricsImpl>()) {} + +bool OrphanFilesCleanerImpl::SupportToClean(const std::string& file_name) { + static std::vector<std::pair<std::string, std::string>> supported_pattern = { + {"manifest-", ""}, {"manifest-list-", ""}, {".", ".tmp"}}; + for (const auto& pattern : supported_pattern) { + if (StringUtils::StartsWith(file_name, pattern.first) && + StringUtils::EndsWith(file_name, pattern.second)) { + return true; + } + } + static std::vector<std::string> supported_formats = {".orc", ".parquet", ".avro", ".lance"}; + for (const auto& format : supported_formats) { + if (StringUtils::StartsWith(file_name, "data-") && + StringUtils::EndsWith(file_name, format)) { + return true; + } + } + return false; +} + +Result<std::set<std::string>> OrphanFilesCleanerImpl::Clean() { + Duration main_duration; + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "tag")).empty()) { + return Status::NotImplemented("OrphanFilesCleaner do not support cleaning table with tag"); + } + if (!MinimalTryBestListingDirs(PathUtil::JoinPath(root_path_, "branch")).empty()) { + return Status::NotImplemented( + "OrphanFilesCleaner do not support cleaning table with branch"); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> all_dirs, ListPaimonFileDirs()); + std::vector<std::future<std::vector<std::unique_ptr<FileStatus>>>> file_statuses_futures; + for (const auto& dir : all_dirs) { + file_statuses_futures.push_back( + Via(executor_.get(), [this, dir] { return TryBestListingDirs(dir); })); + } + PAIMON_ASSIGN_OR_RAISE(std::set<std::string> used_file_names, GetUsedFiles()); + + Duration duration; + std::set<std::string> need_to_deletes; + std::vector<std::future<void>> futures; + ScopeGuard guard([&futures]() { Wait(futures); }); + uint64_t file_statuses_duration = duration.Reset(); + for (const auto& file_statuses : CollectAll(file_statuses_futures)) { + for (const auto& file_status : file_statuses) { + if (file_status->IsDir()) { + continue; + } + std::string path = file_status->GetPath(); + std::string file_name = PathUtil::GetName(path); + if (!SupportToClean(file_name)) { + continue; + } + if (file_status->GetModificationTime() < older_than_ms_ && + !used_file_names.count(file_name)) { + if (should_be_retained_ && should_be_retained_(file_name)) { + continue; + } + if (file_status->GetModificationTime() <= MIN_VALID_FILE_MODIFICATION_MS) { + return Status::Invalid( + fmt::format("file '{}' modification '{}' is not in millisecond", path, + file_status->GetModificationTime())); + } + need_to_deletes.insert(path); + futures.push_back(Via(executor_.get(), [this, path]() { + auto s = fs_->Delete(path, /*recursive=*/false); + (void)s; + })); Review Comment: `Clean()` returns a set described as 'paths of the cleaned files', but deletions are executed asynchronously and deletion statuses are ignored. This can cause `Clean()` to report files as 'cleaned' even when `fs_->Delete(...)` fails. Consider collecting per-delete statuses and only returning successfully deleted paths (or adjusting the API/returned set semantics to explicitly mean 'scheduled/attempted deletions'). ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -0,0 +1,337 @@ +/* + * 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. + */ + +#include "paimon/core/operation/expire_snapshots.h" + +#include <algorithm> +#include <cassert> +#include <cstddef> +#include <cstdint> +#include <functional> +#include <future> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +ExpireSnapshots::ExpireSnapshots(const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<ManifestList>& manifest_list, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<FileSystem>& fs, const ExpireConfig& config, + const std::shared_ptr<Executor>& executor) + : snapshot_manager_(snapshot_manager), + path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + fs_(fs), + config_(config), + executor_(executor), + logger_(Logger::GetLogger("ExpireSnapshots")) {} + +Result<int32_t> ExpireSnapshots::Expire() { + int32_t retain_min = config_.GetSnapshotRetainMin(); + if (retain_min < 1) { + return Status::Invalid( + fmt::format("Expire failed: snapshot retain minimum '{}' is less than 1", retain_min)); + } + + int32_t retain_max = config_.GetSnapshotRetainMax(); + if (retain_max < retain_min) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot retain maximum '{}' must be greater or equal than retain " + "minimum '{}'", + retain_max, retain_min)); + } + int32_t max_deletes = config_.GetSnapshotMaxDeletes(); + if (max_deletes < 0) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot max delete num '{}' must be greater than 0", max_deletes)); + } + if (snapshot_manager_ == nullptr) { + return Status::Invalid("Expire failed: snapshot manager is nullptr"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest_snapshot_id, + snapshot_manager_->LatestSnapshotId()); + if (latest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> earliest_snapshot_id, + snapshot_manager_->EarliestSnapshotId()); + if (earliest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + + // TODO(jinli.zjw): why not only use earliest snapshot id + int64_t min = + std::max(latest_snapshot_id.value() - retain_max + 1, earliest_snapshot_id.value()); + int64_t max = latest_snapshot_id.value() - retain_min + 1; + max = std::min(max, earliest_snapshot_id.value() + max_deletes); + // TODO(jinli.zjw): support consumer manager + int64_t older_than_ms = + DateTimeUtils::GetCurrentUTCTimeUs() / 1000 - config_.GetSnapshotTimeRetainMs(); + for (int64_t snapshot_id = min; snapshot_id < max; snapshot_id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(snapshot_id)); + if (exist) { + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(snapshot_id)); + if (older_than_ms <= snapshot.TimeMillis()) { + return ExpireUntil(earliest_snapshot_id.value(), snapshot_id); + } + } + } + return ExpireUntil(earliest_snapshot_id.value(), max); +} + +Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, + int64_t end_exclusive_id) { + if (end_exclusive_id <= earliest_snapshot_id) { + // TODO(jinli.zjw): write earliest hint + return 0; + } + int64_t begin_inclusive_id = earliest_snapshot_id; + for (int64_t id = end_exclusive_id - 1; id >= earliest_snapshot_id; id--) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id = id + 1; + break; + } + } + PAIMON_LOG_DEBUG(logger_, "Snapshot expire range is [%ld, %ld]", begin_inclusive_id, + end_exclusive_id); + + // Since the data file deletion information for each snapshot is recorded in the delta part of + // the next snapshot, it is necessary to check the next snapshot. Otherwise, its data files will + // not be deleted in this round. + for (int64_t id = begin_inclusive_id + 1; id <= end_exclusive_id; id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedDataFiles(snapshot.DeltaManifestList())); + } + // TODO(jinli.zjw): support delete changelog files + + // data files in bucket directories has been deleted + // then delete changed bucket directories if they are empty + PAIMON_RETURN_NOT_OK(CleanEmptyDirectories()); + + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(end_exclusive_id)); + if (!exist) { + return 0; + } + std::vector<Snapshot> retained_snapshots; + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(end_exclusive_id)); + retained_snapshots.push_back(snapshot); + std::set<std::string> skipping_sets; + PAIMON_RETURN_NOT_OK(GetManifestSkippingSet(retained_snapshots, &skipping_sets)); + for (int64_t id = begin_inclusive_id; id < end_exclusive_id; id++) { + PAIMON_LOG_DEBUG(logger_, "Ready to delete manifests in snapshot #%ld", id); + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.BaseManifestList(), skipping_sets)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.DeltaManifestList(), skipping_sets)); + auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id)); + // delete quietly will ignore any status error + (void)status; + } + PAIMON_RETURN_NOT_OK(snapshot_manager_->CommitEarliestHint(end_exclusive_id)); + return end_exclusive_id - begin_inclusive_id; +} + +/// Try to delete data directories that may be empty after data file deletion. +Status ExpireSnapshots::CleanEmptyDirectories() { + if (!config_.CleanEmptyDirectories() || deletion_buckets_.empty()) { + return Status::OK(); + } + + // All directory paths are deduplicated and sorted by hierarchy level + std::map<int32_t, std::set<std::string>> deduplicate; + for (const auto& [partition, buckets] : deletion_buckets_) { + std::vector<std::string> to_delete_empty_directories; + // try to delete bucket directories + for (const auto& bucket : buckets) { + PAIMON_ASSIGN_OR_RAISE(std::string bucket_path, + path_factory_->BucketPath(partition, bucket)); + to_delete_empty_directories.push_back(bucket_path); + } + std::vector<std::future<void>> futures; + for (const auto& empty_directory : to_delete_empty_directories) { + futures.push_back(Via(executor_.get(), [this, &empty_directory] { + auto ret = TryDeleteEmptyDirectory(empty_directory); + (void)ret; + })); + } + Wait(futures); + + PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> hierarchical_paths, + path_factory_->GetHierarchicalPartitionPath(partition)); + size_t hierarchies = hierarchical_paths.size(); + if (hierarchies == 0) { + continue; + } + + if (TryDeleteEmptyDirectory(hierarchical_paths[hierarchies - 1])) { + // deduplicate high level partition directories + for (size_t hierarchy = 0; hierarchy < hierarchies - 1; hierarchy++) { + deduplicate[hierarchy].insert(hierarchical_paths[hierarchy]); + } + } + } + + for (int32_t hierarchy = deduplicate.size() - 1; hierarchy >= 0; hierarchy--) { Review Comment: When `deduplicate` is empty, `deduplicate.size() - 1` underflows before being converted to `int32_t`, making the loop initialization unsafe/implementation-dependent. Add an explicit empty check before the loop, or iterate safely using reverse iterators (or a `for` loop with `size_t` and a decrement pattern that avoids underflow). ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -0,0 +1,337 @@ +/* + * 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. + */ + +#include "paimon/core/operation/expire_snapshots.h" + +#include <algorithm> +#include <cassert> +#include <cstddef> +#include <cstdint> +#include <functional> +#include <future> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +ExpireSnapshots::ExpireSnapshots(const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<ManifestList>& manifest_list, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<FileSystem>& fs, const ExpireConfig& config, + const std::shared_ptr<Executor>& executor) + : snapshot_manager_(snapshot_manager), + path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + fs_(fs), + config_(config), + executor_(executor), + logger_(Logger::GetLogger("ExpireSnapshots")) {} + +Result<int32_t> ExpireSnapshots::Expire() { + int32_t retain_min = config_.GetSnapshotRetainMin(); + if (retain_min < 1) { + return Status::Invalid( + fmt::format("Expire failed: snapshot retain minimum '{}' is less than 1", retain_min)); + } + + int32_t retain_max = config_.GetSnapshotRetainMax(); + if (retain_max < retain_min) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot retain maximum '{}' must be greater or equal than retain " + "minimum '{}'", + retain_max, retain_min)); + } + int32_t max_deletes = config_.GetSnapshotMaxDeletes(); + if (max_deletes < 0) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot max delete num '{}' must be greater than 0", max_deletes)); + } Review Comment: The validation allows `max_deletes == 0` (only rejects `< 0`), but the error message says 'must be greater than 0'. Update the message to 'must be greater than or equal to 0' to match the actual constraint. ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -0,0 +1,337 @@ +/* + * 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. + */ + +#include "paimon/core/operation/expire_snapshots.h" + +#include <algorithm> +#include <cassert> +#include <cstddef> +#include <cstdint> +#include <functional> +#include <future> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +ExpireSnapshots::ExpireSnapshots(const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<ManifestList>& manifest_list, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<FileSystem>& fs, const ExpireConfig& config, + const std::shared_ptr<Executor>& executor) + : snapshot_manager_(snapshot_manager), + path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + fs_(fs), + config_(config), + executor_(executor), + logger_(Logger::GetLogger("ExpireSnapshots")) {} + +Result<int32_t> ExpireSnapshots::Expire() { + int32_t retain_min = config_.GetSnapshotRetainMin(); + if (retain_min < 1) { + return Status::Invalid( + fmt::format("Expire failed: snapshot retain minimum '{}' is less than 1", retain_min)); + } + + int32_t retain_max = config_.GetSnapshotRetainMax(); + if (retain_max < retain_min) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot retain maximum '{}' must be greater or equal than retain " + "minimum '{}'", + retain_max, retain_min)); + } + int32_t max_deletes = config_.GetSnapshotMaxDeletes(); + if (max_deletes < 0) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot max delete num '{}' must be greater than 0", max_deletes)); + } + if (snapshot_manager_ == nullptr) { + return Status::Invalid("Expire failed: snapshot manager is nullptr"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest_snapshot_id, + snapshot_manager_->LatestSnapshotId()); + if (latest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> earliest_snapshot_id, + snapshot_manager_->EarliestSnapshotId()); + if (earliest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + + // TODO(jinli.zjw): why not only use earliest snapshot id + int64_t min = + std::max(latest_snapshot_id.value() - retain_max + 1, earliest_snapshot_id.value()); + int64_t max = latest_snapshot_id.value() - retain_min + 1; + max = std::min(max, earliest_snapshot_id.value() + max_deletes); + // TODO(jinli.zjw): support consumer manager + int64_t older_than_ms = + DateTimeUtils::GetCurrentUTCTimeUs() / 1000 - config_.GetSnapshotTimeRetainMs(); + for (int64_t snapshot_id = min; snapshot_id < max; snapshot_id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(snapshot_id)); + if (exist) { + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(snapshot_id)); + if (older_than_ms <= snapshot.TimeMillis()) { + return ExpireUntil(earliest_snapshot_id.value(), snapshot_id); + } + } + } + return ExpireUntil(earliest_snapshot_id.value(), max); +} + +Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, + int64_t end_exclusive_id) { + if (end_exclusive_id <= earliest_snapshot_id) { + // TODO(jinli.zjw): write earliest hint + return 0; + } + int64_t begin_inclusive_id = earliest_snapshot_id; + for (int64_t id = end_exclusive_id - 1; id >= earliest_snapshot_id; id--) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id = id + 1; + break; + } + } + PAIMON_LOG_DEBUG(logger_, "Snapshot expire range is [%ld, %ld]", begin_inclusive_id, + end_exclusive_id); + + // Since the data file deletion information for each snapshot is recorded in the delta part of + // the next snapshot, it is necessary to check the next snapshot. Otherwise, its data files will + // not be deleted in this round. + for (int64_t id = begin_inclusive_id + 1; id <= end_exclusive_id; id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedDataFiles(snapshot.DeltaManifestList())); + } + // TODO(jinli.zjw): support delete changelog files + + // data files in bucket directories has been deleted + // then delete changed bucket directories if they are empty + PAIMON_RETURN_NOT_OK(CleanEmptyDirectories()); + + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(end_exclusive_id)); + if (!exist) { + return 0; + } + std::vector<Snapshot> retained_snapshots; + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(end_exclusive_id)); + retained_snapshots.push_back(snapshot); + std::set<std::string> skipping_sets; + PAIMON_RETURN_NOT_OK(GetManifestSkippingSet(retained_snapshots, &skipping_sets)); + for (int64_t id = begin_inclusive_id; id < end_exclusive_id; id++) { + PAIMON_LOG_DEBUG(logger_, "Ready to delete manifests in snapshot #%ld", id); + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.BaseManifestList(), skipping_sets)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.DeltaManifestList(), skipping_sets)); + auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id)); + // delete quietly will ignore any status error + (void)status; + } + PAIMON_RETURN_NOT_OK(snapshot_manager_->CommitEarliestHint(end_exclusive_id)); + return end_exclusive_id - begin_inclusive_id; +} + +/// Try to delete data directories that may be empty after data file deletion. +Status ExpireSnapshots::CleanEmptyDirectories() { + if (!config_.CleanEmptyDirectories() || deletion_buckets_.empty()) { + return Status::OK(); + } + + // All directory paths are deduplicated and sorted by hierarchy level + std::map<int32_t, std::set<std::string>> deduplicate; + for (const auto& [partition, buckets] : deletion_buckets_) { + std::vector<std::string> to_delete_empty_directories; + // try to delete bucket directories + for (const auto& bucket : buckets) { + PAIMON_ASSIGN_OR_RAISE(std::string bucket_path, + path_factory_->BucketPath(partition, bucket)); + to_delete_empty_directories.push_back(bucket_path); + } + std::vector<std::future<void>> futures; + for (const auto& empty_directory : to_delete_empty_directories) { + futures.push_back(Via(executor_.get(), [this, &empty_directory] { + auto ret = TryDeleteEmptyDirectory(empty_directory); + (void)ret; + })); + } Review Comment: The async lambda captures `empty_directory` by reference, but `empty_directory` is a loop variable that changes each iteration. This can lead to use-after-change and deleting the wrong directory. Capture by value (e.g., copy the string into the lambda capture) to make the async work safe. ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -0,0 +1,337 @@ +/* + * 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. + */ + +#include "paimon/core/operation/expire_snapshots.h" + +#include <algorithm> +#include <cassert> +#include <cstddef> +#include <cstdint> +#include <functional> +#include <future> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +ExpireSnapshots::ExpireSnapshots(const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<ManifestList>& manifest_list, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<FileSystem>& fs, const ExpireConfig& config, + const std::shared_ptr<Executor>& executor) + : snapshot_manager_(snapshot_manager), + path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + fs_(fs), + config_(config), + executor_(executor), + logger_(Logger::GetLogger("ExpireSnapshots")) {} + +Result<int32_t> ExpireSnapshots::Expire() { + int32_t retain_min = config_.GetSnapshotRetainMin(); + if (retain_min < 1) { + return Status::Invalid( + fmt::format("Expire failed: snapshot retain minimum '{}' is less than 1", retain_min)); + } + + int32_t retain_max = config_.GetSnapshotRetainMax(); + if (retain_max < retain_min) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot retain maximum '{}' must be greater or equal than retain " + "minimum '{}'", + retain_max, retain_min)); + } + int32_t max_deletes = config_.GetSnapshotMaxDeletes(); + if (max_deletes < 0) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot max delete num '{}' must be greater than 0", max_deletes)); + } + if (snapshot_manager_ == nullptr) { + return Status::Invalid("Expire failed: snapshot manager is nullptr"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest_snapshot_id, + snapshot_manager_->LatestSnapshotId()); + if (latest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> earliest_snapshot_id, + snapshot_manager_->EarliestSnapshotId()); + if (earliest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + + // TODO(jinli.zjw): why not only use earliest snapshot id + int64_t min = + std::max(latest_snapshot_id.value() - retain_max + 1, earliest_snapshot_id.value()); + int64_t max = latest_snapshot_id.value() - retain_min + 1; + max = std::min(max, earliest_snapshot_id.value() + max_deletes); + // TODO(jinli.zjw): support consumer manager + int64_t older_than_ms = + DateTimeUtils::GetCurrentUTCTimeUs() / 1000 - config_.GetSnapshotTimeRetainMs(); + for (int64_t snapshot_id = min; snapshot_id < max; snapshot_id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(snapshot_id)); + if (exist) { + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(snapshot_id)); + if (older_than_ms <= snapshot.TimeMillis()) { + return ExpireUntil(earliest_snapshot_id.value(), snapshot_id); + } + } + } + return ExpireUntil(earliest_snapshot_id.value(), max); +} + +Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, + int64_t end_exclusive_id) { + if (end_exclusive_id <= earliest_snapshot_id) { + // TODO(jinli.zjw): write earliest hint + return 0; + } + int64_t begin_inclusive_id = earliest_snapshot_id; + for (int64_t id = end_exclusive_id - 1; id >= earliest_snapshot_id; id--) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id = id + 1; + break; + } + } + PAIMON_LOG_DEBUG(logger_, "Snapshot expire range is [%ld, %ld]", begin_inclusive_id, + end_exclusive_id); + + // Since the data file deletion information for each snapshot is recorded in the delta part of + // the next snapshot, it is necessary to check the next snapshot. Otherwise, its data files will + // not be deleted in this round. + for (int64_t id = begin_inclusive_id + 1; id <= end_exclusive_id; id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedDataFiles(snapshot.DeltaManifestList())); + } + // TODO(jinli.zjw): support delete changelog files + + // data files in bucket directories has been deleted + // then delete changed bucket directories if they are empty + PAIMON_RETURN_NOT_OK(CleanEmptyDirectories()); + + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(end_exclusive_id)); + if (!exist) { + return 0; + } + std::vector<Snapshot> retained_snapshots; + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(end_exclusive_id)); + retained_snapshots.push_back(snapshot); + std::set<std::string> skipping_sets; + PAIMON_RETURN_NOT_OK(GetManifestSkippingSet(retained_snapshots, &skipping_sets)); + for (int64_t id = begin_inclusive_id; id < end_exclusive_id; id++) { + PAIMON_LOG_DEBUG(logger_, "Ready to delete manifests in snapshot #%ld", id); + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.BaseManifestList(), skipping_sets)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.DeltaManifestList(), skipping_sets)); + auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id)); + // delete quietly will ignore any status error + (void)status; + } + PAIMON_RETURN_NOT_OK(snapshot_manager_->CommitEarliestHint(end_exclusive_id)); + return end_exclusive_id - begin_inclusive_id; +} + +/// Try to delete data directories that may be empty after data file deletion. +Status ExpireSnapshots::CleanEmptyDirectories() { + if (!config_.CleanEmptyDirectories() || deletion_buckets_.empty()) { + return Status::OK(); + } + + // All directory paths are deduplicated and sorted by hierarchy level + std::map<int32_t, std::set<std::string>> deduplicate; + for (const auto& [partition, buckets] : deletion_buckets_) { + std::vector<std::string> to_delete_empty_directories; + // try to delete bucket directories + for (const auto& bucket : buckets) { + PAIMON_ASSIGN_OR_RAISE(std::string bucket_path, + path_factory_->BucketPath(partition, bucket)); + to_delete_empty_directories.push_back(bucket_path); + } + std::vector<std::future<void>> futures; + for (const auto& empty_directory : to_delete_empty_directories) { + futures.push_back(Via(executor_.get(), [this, &empty_directory] { + auto ret = TryDeleteEmptyDirectory(empty_directory); + (void)ret; + })); + } + Wait(futures); + + PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> hierarchical_paths, + path_factory_->GetHierarchicalPartitionPath(partition)); + size_t hierarchies = hierarchical_paths.size(); + if (hierarchies == 0) { + continue; + } + + if (TryDeleteEmptyDirectory(hierarchical_paths[hierarchies - 1])) { + // deduplicate high level partition directories + for (size_t hierarchy = 0; hierarchy < hierarchies - 1; hierarchy++) { + deduplicate[hierarchy].insert(hierarchical_paths[hierarchy]); + } + } + } + + for (int32_t hierarchy = deduplicate.size() - 1; hierarchy >= 0; hierarchy--) { + auto iter = deduplicate.find(hierarchy); + if (iter == deduplicate.end()) { + continue; + } + for (const auto& path : iter->second) { + TryDeleteEmptyDirectory(path); + } + } + + deletion_buckets_.clear(); + return Status::OK(); +} + +bool ExpireSnapshots::TryDeleteEmptyDirectory(const std::string& path) const { + auto s = fs_->Delete(path, /*recursive=*/false); + if (s.ok()) { + return true; + } + return false; +} + +Status ExpireSnapshots::CleanUnusedManifests(const std::string& manifest_list_name, + const std::set<std::string>& skipping_sets) { + std::vector<ManifestFileMeta> manifest_file_metas; + auto status = manifest_list_->Read(manifest_list_name, nullptr, &manifest_file_metas); + if (status.ok()) { + std::vector<std::string> to_delete_manifests; + // TODO(jinli.zjw): optimize for async + for (const auto& manifest_file_meta : manifest_file_metas) { Review Comment: `to_delete_manifests` is declared but never used. Remove it to reduce noise, or use it to batch deletions if that's the intended future direction. ########## src/paimon/core/operation/expire_snapshots.cpp: ########## @@ -0,0 +1,337 @@ +/* + * 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. + */ + +#include "paimon/core/operation/expire_snapshots.h" + +#include <algorithm> +#include <cassert> +#include <cstddef> +#include <cstdint> +#include <functional> +#include <future> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/executor/future.h" +#include "paimon/common/utils/date_time_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +ExpireSnapshots::ExpireSnapshots(const std::shared_ptr<SnapshotManager>& snapshot_manager, + const std::shared_ptr<FileStorePathFactory>& path_factory, + const std::shared_ptr<ManifestList>& manifest_list, + const std::shared_ptr<ManifestFile>& manifest_file, + const std::shared_ptr<FileSystem>& fs, const ExpireConfig& config, + const std::shared_ptr<Executor>& executor) + : snapshot_manager_(snapshot_manager), + path_factory_(path_factory), + manifest_list_(manifest_list), + manifest_file_(manifest_file), + fs_(fs), + config_(config), + executor_(executor), + logger_(Logger::GetLogger("ExpireSnapshots")) {} + +Result<int32_t> ExpireSnapshots::Expire() { + int32_t retain_min = config_.GetSnapshotRetainMin(); + if (retain_min < 1) { + return Status::Invalid( + fmt::format("Expire failed: snapshot retain minimum '{}' is less than 1", retain_min)); + } + + int32_t retain_max = config_.GetSnapshotRetainMax(); + if (retain_max < retain_min) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot retain maximum '{}' must be greater or equal than retain " + "minimum '{}'", + retain_max, retain_min)); + } + int32_t max_deletes = config_.GetSnapshotMaxDeletes(); + if (max_deletes < 0) { + return Status::Invalid(fmt::format( + "Expire failed: snapshot max delete num '{}' must be greater than 0", max_deletes)); + } + if (snapshot_manager_ == nullptr) { + return Status::Invalid("Expire failed: snapshot manager is nullptr"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest_snapshot_id, + snapshot_manager_->LatestSnapshotId()); + if (latest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> earliest_snapshot_id, + snapshot_manager_->EarliestSnapshotId()); + if (earliest_snapshot_id == std::nullopt) { + // no snapshot, nothing to expire + return 0; + } + + // TODO(jinli.zjw): why not only use earliest snapshot id + int64_t min = + std::max(latest_snapshot_id.value() - retain_max + 1, earliest_snapshot_id.value()); + int64_t max = latest_snapshot_id.value() - retain_min + 1; + max = std::min(max, earliest_snapshot_id.value() + max_deletes); + // TODO(jinli.zjw): support consumer manager + int64_t older_than_ms = + DateTimeUtils::GetCurrentUTCTimeUs() / 1000 - config_.GetSnapshotTimeRetainMs(); + for (int64_t snapshot_id = min; snapshot_id < max; snapshot_id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(snapshot_id)); + if (exist) { + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(snapshot_id)); + if (older_than_ms <= snapshot.TimeMillis()) { + return ExpireUntil(earliest_snapshot_id.value(), snapshot_id); + } + } + } + return ExpireUntil(earliest_snapshot_id.value(), max); +} + +Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, + int64_t end_exclusive_id) { + if (end_exclusive_id <= earliest_snapshot_id) { + // TODO(jinli.zjw): write earliest hint + return 0; + } + int64_t begin_inclusive_id = earliest_snapshot_id; + for (int64_t id = end_exclusive_id - 1; id >= earliest_snapshot_id; id--) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id = id + 1; + break; + } + } + PAIMON_LOG_DEBUG(logger_, "Snapshot expire range is [%ld, %ld]", begin_inclusive_id, + end_exclusive_id); + + // Since the data file deletion information for each snapshot is recorded in the delta part of + // the next snapshot, it is necessary to check the next snapshot. Otherwise, its data files will + // not be deleted in this round. + for (int64_t id = begin_inclusive_id + 1; id <= end_exclusive_id; id++) { + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedDataFiles(snapshot.DeltaManifestList())); + } + // TODO(jinli.zjw): support delete changelog files + + // data files in bucket directories has been deleted + // then delete changed bucket directories if they are empty + PAIMON_RETURN_NOT_OK(CleanEmptyDirectories()); + + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(end_exclusive_id)); + if (!exist) { + return 0; + } + std::vector<Snapshot> retained_snapshots; + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(end_exclusive_id)); + retained_snapshots.push_back(snapshot); + std::set<std::string> skipping_sets; + PAIMON_RETURN_NOT_OK(GetManifestSkippingSet(retained_snapshots, &skipping_sets)); + for (int64_t id = begin_inclusive_id; id < end_exclusive_id; id++) { + PAIMON_LOG_DEBUG(logger_, "Ready to delete manifests in snapshot #%ld", id); + PAIMON_ASSIGN_OR_RAISE(bool exist, snapshot_manager_->SnapshotExists(id)); + if (!exist) { + begin_inclusive_id++; + continue; + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.BaseManifestList(), skipping_sets)); + PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.DeltaManifestList(), skipping_sets)); + auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id)); + // delete quietly will ignore any status error + (void)status; + } + PAIMON_RETURN_NOT_OK(snapshot_manager_->CommitEarliestHint(end_exclusive_id)); + return end_exclusive_id - begin_inclusive_id; +} + +/// Try to delete data directories that may be empty after data file deletion. +Status ExpireSnapshots::CleanEmptyDirectories() { + if (!config_.CleanEmptyDirectories() || deletion_buckets_.empty()) { + return Status::OK(); + } + + // All directory paths are deduplicated and sorted by hierarchy level + std::map<int32_t, std::set<std::string>> deduplicate; + for (const auto& [partition, buckets] : deletion_buckets_) { + std::vector<std::string> to_delete_empty_directories; + // try to delete bucket directories + for (const auto& bucket : buckets) { + PAIMON_ASSIGN_OR_RAISE(std::string bucket_path, + path_factory_->BucketPath(partition, bucket)); + to_delete_empty_directories.push_back(bucket_path); + } + std::vector<std::future<void>> futures; + for (const auto& empty_directory : to_delete_empty_directories) { + futures.push_back(Via(executor_.get(), [this, &empty_directory] { + auto ret = TryDeleteEmptyDirectory(empty_directory); + (void)ret; + })); + } + Wait(futures); + + PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> hierarchical_paths, + path_factory_->GetHierarchicalPartitionPath(partition)); + size_t hierarchies = hierarchical_paths.size(); + if (hierarchies == 0) { + continue; + } + + if (TryDeleteEmptyDirectory(hierarchical_paths[hierarchies - 1])) { + // deduplicate high level partition directories + for (size_t hierarchy = 0; hierarchy < hierarchies - 1; hierarchy++) { + deduplicate[hierarchy].insert(hierarchical_paths[hierarchy]); + } + } + } + + for (int32_t hierarchy = deduplicate.size() - 1; hierarchy >= 0; hierarchy--) { + auto iter = deduplicate.find(hierarchy); + if (iter == deduplicate.end()) { + continue; + } + for (const auto& path : iter->second) { + TryDeleteEmptyDirectory(path); + } + } + + deletion_buckets_.clear(); + return Status::OK(); +} + +bool ExpireSnapshots::TryDeleteEmptyDirectory(const std::string& path) const { + auto s = fs_->Delete(path, /*recursive=*/false); + if (s.ok()) { + return true; + } + return false; +} + +Status ExpireSnapshots::CleanUnusedManifests(const std::string& manifest_list_name, + const std::set<std::string>& skipping_sets) { + std::vector<ManifestFileMeta> manifest_file_metas; + auto status = manifest_list_->Read(manifest_list_name, nullptr, &manifest_file_metas); + if (status.ok()) { + std::vector<std::string> to_delete_manifests; + // TODO(jinli.zjw): optimize for async + for (const auto& manifest_file_meta : manifest_file_metas) { + if (skipping_sets.count(manifest_file_meta.FileName()) == 0) { + manifest_file_->DeleteQuietly(manifest_file_meta.FileName()); + } + } + if (skipping_sets.count(manifest_list_name) == 0) { + manifest_list_->DeleteQuietly(manifest_list_name); + } + } + return Status::OK(); +} + +Status ExpireSnapshots::CleanUnusedDataFiles(const std::string& manifest_list_name) { + std::vector<ManifestFileMeta> manifest_file_metas; + auto status = manifest_list_->Read(manifest_list_name, nullptr, &manifest_file_metas); + if (status.ok()) { + std::map<std::string, ManifestEntry> data_files_to_delete; + for (const auto& manifest_file_meta : manifest_file_metas) { + std::vector<ManifestEntry> manifest_entries; + auto status = + manifest_file_->Read(manifest_file_meta.FileName(), nullptr, &manifest_entries); + if (!status.ok()) { + // cancel deletion if any exception occurs + PAIMON_LOG_WARN(logger_, "Failed to read some manifest files. Cancel deletion. %s", + status.ToString().c_str()); + return Status::OK(); + } else { + PAIMON_RETURN_NOT_OK(GetDataFilesToDelete(manifest_entries, &data_files_to_delete)); + } + } + + std::vector<std::future<void>> futures; + ScopeGuard guard([&futures]() { Wait(futures); }); + for (const auto& [data_file_to_delete, entry] : data_files_to_delete) { + auto delete_file_path = data_file_to_delete; + futures.push_back(Via(executor_.get(), [this, delete_file_path]() { + auto status = fs_->Delete(delete_file_path); + // delete quietly will ignore any status error + (void)status; + })); + deletion_buckets_[entry.Partition()].insert(entry.Bucket()); + } + } + return Status::OK(); +} + +Status ExpireSnapshots::GetDataFilesToDelete( + const std::vector<ManifestEntry>& data_file_entries, + std::map<std::string, ManifestEntry>* data_files_to_delete) const { + for (const auto& entry : data_file_entries) { + PAIMON_ASSIGN_OR_RAISE(std::string bucket_path, + path_factory_->BucketPath(entry.Partition(), entry.Bucket())); + std::string data_file_path = PathUtil::JoinPath(bucket_path, entry.FileName()); + if (entry.Kind() == FileKind::Add()) { + data_files_to_delete->erase(data_file_path); + } else if (entry.Kind() == FileKind::Delete()) { + // TODO(jinli.zjw): do not support extra files + data_files_to_delete->insert({data_file_path, entry}); + } else { + return Status::Invalid( + fmt::format("Unknown value kind {}", entry.Kind().ToByteValue())); + } + } + return Status::OK(); +} + +Status ExpireSnapshots::GetManifestSkippingSet(const std::vector<Snapshot>& retained_snapshots, + std::set<std::string>* skipping_manifest_set) const { + for (const auto& snapshot : retained_snapshots) { + skipping_manifest_set->insert(snapshot.BaseManifestList()); + skipping_manifest_set->insert(snapshot.DeltaManifestList()); + std::vector<ManifestFileMeta> manifests; + PAIMON_RETURN_NOT_OK(manifest_list_->ReadDataManifests(snapshot, &manifests)); + for (const auto& manifest : manifests) { + skipping_manifest_set->insert(manifest.FileName()); + } + // TODO(jinli.zjw): skip index manifests + if (snapshot.IndexManifest() && snapshot.IndexManifest().value() != "") { + assert(false); + return Status::NotImplemented("do not support expire snapshot with index manifest"); + } Review Comment: This code will abort the process in debug builds whenever an index manifest is present due to `assert(false)`. Returning `NotImplemented` is sufficient to signal unsupported behavior; remove the assert (or gate it behind a non-fatal logging path) so the library does not crash in supported runtime configurations. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
