// 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 "cloud/cloud_tablet.h"

#include <bvar/bvar.h>

#include <bvar/latency_recorder.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <rapidjson/document.h>

#include <rapidjson/encodings.h>

#include <rapidjson/prettywriter.h>

#include <rapidjson/rapidjson.h>

#include <rapidjson/stringbuffer.h>

#include <atomic>

#include <chrono>

#include <cstdint>

#include <memory>

#include <ranges>

#include <ratio>

#include <shared_mutex>

#include <unordered_map>

#include <vector>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet_mgr.h"

#include "cloud/cloud_warm_up_manager.h"

#include "common/cast_set.h"

#include "common/config.h"

#include "common/logging.h"

#include "cpp/sync_point.h"

#include "io/cache/block_file_cache_downloader.h"

#include "io/cache/block_file_cache_factory.h"

#include "olap/base_tablet.h"

#include "olap/compaction.h"

#include "olap/cumulative_compaction_time_series_policy.h"

#include "olap/olap_define.h"

#include "olap/rowset/beta_rowset.h"

#include "olap/rowset/rowset.h"

#include "olap/rowset/rowset_factory.h"

#include "olap/rowset/rowset_fwd.h"

#include "olap/rowset/rowset_writer.h"

#include "olap/rowset/segment_v2/inverted_index_desc.h"

#include "olap/storage_policy.h"

#include "olap/tablet_schema.h"

#include "olap/txn_manager.h"

#include "util/debug_points.h"

#include "vec/common/schema_util.h"

namespace doris {

#include "common/compile_check_begin.h"

using namespace ErrorCode;

bvar::LatencyRecorder g_cu_compaction_get_delete_bitmap_lock_time_ms(

        "cu_compaction_get_delete_bitmap_lock_time_ms");

bvar::LatencyRecorder g_base_compaction_get_delete_bitmap_lock_time_ms(

        "base_compaction_get_delete_bitmap_lock_time_ms");

bvar::Adder<int64_t> g_unused_rowsets_count("unused_rowsets_count");

bvar::Adder<int64_t> g_unused_rowsets_bytes("unused_rowsets_bytes");

bvar::Adder<int64_t> g_capture_prefer_cache_count("capture_prefer_cache_count");

bvar::Adder<int64_t> g_capture_with_freshness_tolerance_count(

        "capture_with_freshness_tolerance_count");

bvar::Adder<int64_t> g_capture_with_freshness_tolerance_fallback_count(

        "capture_with_freshness_tolerance_fallback_count");

bvar::Window<bvar::Adder<int64_t>> g_capture_prefer_cache_count_window(

        "capture_prefer_cache_count_window", &g_capture_prefer_cache_count, 30);

bvar::Window<bvar::Adder<int64_t>> g_capture_with_freshness_tolerance_count_window(

        "capture_with_freshness_tolerance_count_window", &g_capture_with_freshness_tolerance_count,

        30);

bvar::Window<bvar::Adder<int64_t>> g_capture_with_freshness_tolerance_fallback_count_window(

        "capture_with_freshness_tolerance_fallback_count_window",

        &g_capture_with_freshness_tolerance_fallback_count, 30);

static constexpr int LOAD_INITIATOR_ID = -1;

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_submitted_segment_size(

        "file_cache_cloud_tablet_submitted_segment_size");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_submitted_segment_num(

        "file_cache_cloud_tablet_submitted_segment_num");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_submitted_index_size(

        "file_cache_cloud_tablet_submitted_index_size");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_submitted_index_num(

        "file_cache_cloud_tablet_submitted_index_num");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_finished_segment_size(

        "file_cache_cloud_tablet_finished_segment_size");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_finished_segment_num(

        "file_cache_cloud_tablet_finished_segment_num");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_finished_index_size(

        "file_cache_cloud_tablet_finished_index_size");

bvar::Adder<uint64_t> g_file_cache_cloud_tablet_finished_index_num(

        "file_cache_cloud_tablet_finished_index_num");

bvar::Adder<uint64_t> g_file_cache_recycle_cached_data_segment_num(

        "file_cache_recycle_cached_data_segment_num");

bvar::Adder<uint64_t> g_file_cache_recycle_cached_data_segment_size(

        "file_cache_recycle_cached_data_segment_size");

bvar::Adder<uint64_t> g_file_cache_recycle_cached_data_index_num(

        "file_cache_recycle_cached_data_index_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_segment_complete_num(

        "file_cache_warm_up_segment_complete_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_segment_failed_num(

        "file_cache_warm_up_segment_failed_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_inverted_idx_complete_num(

        "file_cache_warm_up_inverted_idx_complete_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_inverted_idx_failed_num(

        "file_cache_warm_up_inverted_idx_failed_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_complete_num(

        "file_cache_warm_up_rowset_complete_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_triggered_by_job_num(

        "file_cache_warm_up_rowset_triggered_by_job_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_triggered_by_sync_rowset_num(

        "file_cache_warm_up_rowset_triggered_by_sync_rowset_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_triggered_by_event_driven_num(

        "file_cache_warm_up_rowset_triggered_by_event_driven_num");

bvar::LatencyRecorder g_file_cache_warm_up_rowset_all_segments_latency(

        "file_cache_warm_up_rowset_all_segments_latency");

CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)

        : BaseTablet(std::move(tablet_meta)), _engine(engine) {}

CloudTablet::~CloudTablet() = default;

bool CloudTablet::exceed_version_limit(int32_t limit) {

    return _approximate_num_rowsets.load(std::memory_order_relaxed) > limit;

}

std::string CloudTablet::tablet_path() const {

    return "";

}

Status CloudTablet::capture_rs_readers(const Version& spec_version,

                                       std::vector<RowSetSplits>* rs_splits,

                                       const CaptureRowsetOps& opts) {

    DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", {

        LOG_WARNING("CloudTablet.capture_rs_readers.return e-230").tag("tablet_id", tablet_id());

        return Status::Error<false>(-230, "injected error");

    });

    std::shared_lock rlock(_meta_lock);

    *rs_splits = DORIS_TRY(capture_rs_readers_unlocked(

            spec_version, CaptureRowsetOps {.skip_missing_versions = opts.skip_missing_versions}));

    return Status::OK();

}

[[nodiscard]] Result<std::vector<Version>> CloudTablet::capture_consistent_versions_unlocked(

        const Version& version_range, const CaptureRowsetOps& options) const {

    if (options.query_freshness_tolerance_ms > 0) {

        return capture_versions_with_freshness_tolerance(version_range, options);

    } else if (options.enable_prefer_cached_rowset && !enable_unique_key_merge_on_write()) {

        return capture_versions_prefer_cache(version_range);

    }

    return BaseTablet::capture_consistent_versions_unlocked(version_range, options);

}

Result<std::vector<Version>> CloudTablet::capture_versions_prefer_cache(

        const Version& spec_version) const {

    g_capture_prefer_cache_count << 1;

    Versions version_path;

    std::shared_lock rlock(_meta_lock);

    auto st = _timestamped_version_tracker.capture_consistent_versions_prefer_cache(

            spec_version, version_path,

            [&](int64_t start, int64_t end) { return rowset_is_warmed_up_unlocked(start, end); });

    if (!st.ok()) {

        return ResultError(st);

    }

    int64_t path_max_version = version_path.back().second;

    VLOG_DEBUG << fmt::format(

            "[verbose] CloudTablet::capture_versions_prefer_cache, capture path: {}, "

            "tablet_id={}, spec_version={}, path_max_version={}",

            fmt::join(version_path | std::views::transform([](const auto& version) {

                          return fmt::format("{}", version.to_string());

                      }),

                      ", "),

            tablet_id(), spec_version.to_string(), path_max_version);

    return version_path;

}

bool CloudTablet::rowset_is_warmed_up_unlocked(int64_t start_version, int64_t end_version) const {

    if (start_version > end_version) {

        return false;

    }

    Version version {start_version, end_version};

    auto it = _rs_version_map.find(version);

    if (it == _rs_version_map.end()) {

        it = _stale_rs_version_map.find(version);

        if (it == _stale_rs_version_map.end()) {

            LOG_WARNING(

                    "fail to find Rowset in rs_version or stale_rs_version for version. "

                    "tablet={}, version={}",

                    tablet_id(), version.to_string());

            return false;

        }

    }

    const auto& rs = it->second;

    if (rs->visible_timestamp() < _engine.startup_timepoint()) {

        // We only care about rowsets that are created after startup time point. For other rowsets,

        // we assume they are warmed up.

        return true;

    }

    return is_rowset_warmed_up(rs->rowset_id());

};

Result<std::vector<Version>> CloudTablet::capture_versions_with_freshness_tolerance(

        const Version& spec_version, const CaptureRowsetOps& options) const {

    g_capture_with_freshness_tolerance_count << 1;

    using namespace std::chrono;

    auto query_freshness_tolerance_ms = options.query_freshness_tolerance_ms;

    auto freshness_limit_tp = system_clock::now() - milliseconds(query_freshness_tolerance_ms);

    // find a version path where every edge(rowset) has been warmuped

    Versions version_path;

    std::shared_lock rlock(_meta_lock);

    if (enable_unique_key_merge_on_write()) {

        // For merge-on-write table, newly generated delete bitmap marks will be on the rowsets which are in newest layout.

        // So we can ony capture rowsets which are in newest data layout. Otherwise there may be data correctness issue.

        RETURN_IF_ERROR_RESULT(

                _timestamped_version_tracker.capture_consistent_versions_with_validator_mow(

                        spec_version, version_path, [&](int64_t start, int64_t end) {

                            return rowset_is_warmed_up_unlocked(start, end);

                        }));

    } else {

        RETURN_IF_ERROR_RESULT(

                _timestamped_version_tracker.capture_consistent_versions_with_validator(

                        spec_version, version_path, [&](int64_t start, int64_t end) {

                            return rowset_is_warmed_up_unlocked(start, end);

                        }));

    }

    int64_t path_max_version = version_path.back().second;

    auto should_be_visible_but_not_warmed_up = [&](const auto& rs_meta) -> bool {

        if (rs_meta->version() == Version {0, 1}) {

            // skip rowset[0-1]

            return false;

        }

        bool ret = rs_meta->start_version() > path_max_version &&

                   rs_meta->visible_timestamp() < freshness_limit_tp;

        if (ret && config::read_cluster_cache_opt_verbose_log) {

            std::time_t t1 = system_clock::to_time_t(rs_meta->visible_timestamp());

            std::tm tm1 = *std::localtime(&t1);

            std::ostringstream oss1;

            oss1 << std::put_time(&tm1, "%Y-%m-%d %H:%M:%S");

            std::time_t t2 = system_clock::to_time_t(freshness_limit_tp);

            std::tm tm2 = *std::localtime(&t2);

            std::ostringstream oss2;

            oss2 << std::put_time(&tm2, "%Y-%m-%d %H:%M:%S");

            LOG_INFO(

                    "[verbose] CloudTablet::capture_rs_readers_with_freshness_tolerance, "

                    "find a rowset which should be visible but not warmed up, tablet_id={}, "

                    "path_max_version={}, rowset_id={}, version={}, visible_time={}, "

                    "freshness_limit={}, version_graph={}, rowset_warmup_digest={}",

                    tablet_id(), path_max_version, rs_meta->rowset_id().to_string(),

                    rs_meta->version().to_string(), oss1.str(), oss2.str(),

                    _timestamped_version_tracker.debug_string(), rowset_warmup_digest());

        }

        return ret;

    };

    // use std::views::concat after C++26

    bool should_fallback =

            std::ranges::any_of(std::views::values(_tablet_meta->all_rs_metas()),

                                should_be_visible_but_not_warmed_up) ||

            std::ranges::any_of(std::views::values(_tablet_meta->all_stale_rs_metas()),

                                should_be_visible_but_not_warmed_up);

    if (should_fallback) {

        rlock.unlock();

        g_capture_with_freshness_tolerance_fallback_count << 1;

        // if there exists a rowset which satisfies freshness tolerance and its start version is larger than the path max version

        // but has not been warmuped up yet, fallback to capture rowsets as usual

        return BaseTablet::capture_consistent_versions_unlocked(spec_version, options);

    }

    VLOG_DEBUG << fmt::format(

            "[verbose] CloudTablet::capture_versions_with_freshness_tolerance, capture path: {}, "

            "tablet_id={}, spec_version={}, path_max_version={}",

            fmt::join(version_path | std::views::transform([](const auto& version) {

                          return fmt::format("{}", version.to_string());

                      }),

                      ", "),

            tablet_id(), spec_version.to_string(), path_max_version);

    return version_path;

}

// There are only two tablet_states RUNNING and NOT_READY in cloud mode

// This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS.

Status CloudTablet::sync_rowsets(const SyncOptions& options, SyncRowsetStats* stats) {

    RETURN_IF_ERROR(sync_if_not_running(stats));

    if (options.query_version > 0) {

        std::shared_lock rlock(_meta_lock);

        if (_max_version >= options.query_version) {

            return Status::OK();

        }

    }

    // serially execute sync to reduce unnecessary network overhead

    std::unique_lock lock(_sync_meta_lock);

    if (options.query_version > 0) {

        std::shared_lock rlock(_meta_lock);

        if (_max_version >= options.query_version) {

            return Status::OK();

        }

    }

    auto st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, options, stats);

    if (st.is<ErrorCode::NOT_FOUND>()) {

        clear_cache();

    }

    return st;

}

// Sync tablet meta and all rowset meta if not running.

// This could happen when BE didn't finish schema change job and another BE committed this schema change job.

// It should be a quite rare situation.

Status CloudTablet::sync_if_not_running(SyncRowsetStats* stats) {

    if (tablet_state() == TABLET_RUNNING) {

        return Status::OK();

    }

    // Serially execute sync to reduce unnecessary network overhead

    std::unique_lock lock(_sync_meta_lock);

    {

        std::shared_lock rlock(_meta_lock);

        if (tablet_state() == TABLET_RUNNING) {

            return Status::OK();

        }

    }

    TabletMetaSharedPtr tablet_meta;

    auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);

    if (!st.ok()) {

        if (st.is<ErrorCode::NOT_FOUND>()) {

            clear_cache();

        }

        return st;

    }

    if (tablet_meta->tablet_state() != TABLET_RUNNING) [[unlikely]] {

        // MoW may go to here when load while schema change

        return Status::OK();

    }

    TimestampedVersionTracker empty_tracker;

    {

        std::lock_guard wlock(_meta_lock);

        RETURN_IF_ERROR(set_tablet_state(TABLET_RUNNING));

        _rs_version_map.clear();

        _stale_rs_version_map.clear();

        std::swap(_timestamped_version_tracker, empty_tracker);

        _tablet_meta->clear_rowsets();

        _tablet_meta->clear_stale_rowset();

        _max_version = -1;

    }

    st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, {}, stats);

    if (st.is<ErrorCode::NOT_FOUND>()) {

        clear_cache();

    }

    return st;

}

void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,

                              std::unique_lock<std::shared_mutex>& meta_lock,

                              bool warmup_delta_data) {

    if (to_add.empty()) {

        return;

    }

    auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) {

        for (auto& rs : rowsets) {

            if (warmup_delta_data) {

#ifndef BE_TEST

                bool warm_up_state_updated = false;

                // Warmup rowset data in background

                for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {

                    const auto& rowset_meta = rs->rowset_meta();

                    constexpr int64_t interval = 600; // 10 mins

                    // When BE restart and receive the `load_sync` rpc, it will sync all historical rowsets first time.

                    // So we need to filter out the old rowsets avoid to download the whole table.

                    if (warmup_delta_data &&

                        ::time(nullptr) - rowset_meta->newest_write_timestamp() >= interval) {

                        continue;

                    }

                    auto storage_resource = rowset_meta->remote_storage_resource();

                    if (!storage_resource) {

                        LOG(WARNING) << storage_resource.error();

                        continue;

                    }

                    int64_t expiration_time =

                            _tablet_meta->ttl_seconds() == 0 ||

                                            rowset_meta->newest_write_timestamp() <= 0

                                    ? 0

                                    : rowset_meta->newest_write_timestamp() +

                                              _tablet_meta->ttl_seconds();

                    g_file_cache_cloud_tablet_submitted_segment_num << 1;

                    if (rs->rowset_meta()->segment_file_size(seg_id) > 0) {

                        g_file_cache_cloud_tablet_submitted_segment_size

                                << rs->rowset_meta()->segment_file_size(seg_id);

                    }

                    if (!warm_up_state_updated) {

                        VLOG_DEBUG << "warm up rowset " << rs->version() << "(" << rs->rowset_id()

                                   << ") triggerd by sync rowset";

                        if (!add_rowset_warmup_state_unlocked(*(rs->rowset_meta()),

                                                              WarmUpTriggerSource::SYNC_ROWSET)) {

                            LOG(INFO) << "found duplicate warmup task for rowset "

                                      << rs->rowset_id() << ", skip it";

                            break;

                        }

                        warm_up_state_updated = true;

                    }

                    // clang-format off

                    auto self = std::dynamic_pointer_cast<CloudTablet>(shared_from_this());

                    if (!config::file_cache_enable_only_warm_up_idx) {

                        _engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta {

                                .path = storage_resource.value()->remote_segment_path(*rowset_meta, seg_id),

                                .file_size = rs->rowset_meta()->segment_file_size(seg_id),

                                .file_system = storage_resource.value()->fs,

                                .ctx =

                                        {

                                                .expiration_time = expiration_time,

                                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,

                                                .is_warmup = true

                                        },

                                .download_done {[=](Status st) {

                                    DBUG_EXECUTE_IF("CloudTablet::add_rowsets.download_data.callback.block_compaction_rowset", {

                                                if (rs->version().second > rs->version().first) {

                                                    auto sleep_time = dp->param<int>("sleep", 3);

                                                    LOG_INFO(

                                                            "[verbose] block download for rowset={}, "

                                                            "version={}, sleep={}",

                                                            rs->rowset_id().to_string(),

                                                            rs->version().to_string(), sleep_time);

                                                    std::this_thread::sleep_for(

                                                            std::chrono::seconds(sleep_time));

                                                }

                                    });

                                    self->complete_rowset_segment_warmup(WarmUpTriggerSource::SYNC_ROWSET, rowset_meta->rowset_id(), st, 1, 0);

                                    if (!st) {

                                        LOG_WARNING("add rowset warm up error ").error(st);

                                    }

                                }},

                        });

                    }

                    auto download_idx_file = [&, self](const io::Path& idx_path, int64_t idx_size) {

                        io::DownloadFileMeta meta {

                                .path = idx_path,

                                .file_size = idx_size,

                                .file_system = storage_resource.value()->fs,

                                .ctx =

                                        {

                                                .expiration_time = expiration_time,

                                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,

                                                .is_warmup = true

                                        },

                                .download_done {[=](Status st) {

                                    DBUG_EXECUTE_IF("CloudTablet::add_rowsets.download_idx.callback.block", {

                                                // clang-format on

                                                auto sleep_time = dp->param<int>("sleep", 3);

                                                LOG_INFO(

                                                        "[verbose] block download for "

                                                        "rowset={}, inverted_idx_file={}, "

                                                        "sleep={}",

                                                        rs->rowset_id().to_string(),

                                                        idx_path.string(), sleep_time);

                                                std::this_thread::sleep_for(

                                                        std::chrono::seconds(sleep_time));

                                                // clang-format off

                                    });

                                    self->complete_rowset_segment_warmup(WarmUpTriggerSource::SYNC_ROWSET, rowset_meta->rowset_id(), st, 0, 1);

                                    if (!st) {

                                        LOG_WARNING("add rowset warm up error ").error(st);

                                    }

                                }},

                        };

                        self->update_rowset_warmup_state_inverted_idx_num_unlocked(WarmUpTriggerSource::SYNC_ROWSET, rowset_meta->rowset_id(), 1);

                        _engine.file_cache_block_downloader().submit_download_task(std::move(meta));

                        g_file_cache_cloud_tablet_submitted_index_num << 1;

                        g_file_cache_cloud_tablet_submitted_index_size << idx_size;

                    };

                    // clang-format on

                    auto schema_ptr = rowset_meta->tablet_schema();

                    auto idx_version = schema_ptr->get_inverted_index_storage_format();

                    if (idx_version == InvertedIndexStorageFormatPB::V1) {

                        std::unordered_map<int64_t, int64_t> index_size_map;

                        auto&& inverted_index_info = rowset_meta->inverted_index_file_info(seg_id);

                        for (const auto& info : inverted_index_info.index_info()) {

                            if (info.index_file_size() != -1) {

                                index_size_map[info.index_id()] = info.index_file_size();

                            } else {

                                VLOG_DEBUG << "Invalid index_file_size for segment_id " << seg_id

                                           << ", index_id " << info.index_id();

                            }

                        }

                        for (const auto& index : schema_ptr->inverted_indexes()) {

                            auto idx_path = storage_resource.value()->remote_idx_v1_path(

                                    *rowset_meta, seg_id, index->index_id(),

                                    index->get_index_suffix());

                            download_idx_file(idx_path, index_size_map[index->index_id()]);

                        }

                    } else {

                        if (schema_ptr->has_inverted_index() || schema_ptr->has_ann_index()) {

                            auto&& inverted_index_info =

                                    rowset_meta->inverted_index_file_info(seg_id);

                            int64_t idx_size = 0;

                            if (inverted_index_info.has_index_size()) {

                                idx_size = inverted_index_info.index_size();

                            } else {

                                VLOG_DEBUG << "index_size is not set for segment " << seg_id;

                            }

                            auto idx_path = storage_resource.value()->remote_idx_v2_path(

                                    *rowset_meta, seg_id);

                            download_idx_file(idx_path, idx_size);

                        }

                    }

                }

#endif

            }

            _rs_version_map.emplace(rs->version(), rs);

            _timestamped_version_tracker.add_version(rs->version());

            _max_version = std::max(rs->end_version(), _max_version);

            update_base_size(*rs);

        }

        _tablet_meta->add_rowsets_unchecked(rowsets);

    };

    if (!version_overlap) {

        add_rowsets_directly(to_add);

        return;

    }

    // Filter out existed rowsets

    auto remove_it =

            std::remove_if(to_add.begin(), to_add.end(), [this](const RowsetSharedPtr& rs) {

                if (auto find_it = _rs_version_map.find(rs->version());

                    find_it == _rs_version_map.end()) {

                    return false;

                } else if (find_it->second->rowset_id() == rs->rowset_id()) {

                    return true; // Same rowset

                }

                // If version of rowset in `to_add` is equal to rowset in tablet but rowset_id is not equal,

                // replace existed rowset with `to_add` rowset. This may occur when:

                //  1. schema change converts rowsets which have been double written to new tablet

                //  2. cumu compaction picks single overlapping input rowset to perform compaction

                // add existed rowset to unused_rowsets to remove delete bitmap and recycle cached data

                std::vector<RowsetSharedPtr> unused_rowsets;

                if (auto find_it = _rs_version_map.find(rs->version());

                    find_it != _rs_version_map.end()) {

                    if (find_it->second->rowset_id() == rs->rowset_id()) {

                        LOG(WARNING) << "tablet_id=" << tablet_id()

                                     << ", rowset_id=" << rs->rowset_id().to_string()

                                     << ", existed rowset_id="

                                     << find_it->second->rowset_id().to_string();

                        DCHECK(find_it->second->rowset_id() != rs->rowset_id())

                                << "tablet_id=" << tablet_id()

                                << ", rowset_id=" << rs->rowset_id().to_string()

                                << ", existed rowset_id="

                                << find_it->second->rowset_id().to_string();

                    }

                    unused_rowsets.push_back(find_it->second);

                }

                add_unused_rowsets(unused_rowsets);

                _tablet_meta->delete_rs_meta_by_version(rs->version(), nullptr);

                _rs_version_map[rs->version()] = rs;

                _tablet_meta->add_rowsets_unchecked({rs});

                update_base_size(*rs);

                return true;

            });

    to_add.erase(remove_it, to_add.end());

    // delete rowsets with overlapped version

    std::vector<RowsetSharedPtr> to_add_directly;

    for (auto& to_add_rs : to_add) {

        // delete rowsets with overlapped version

        std::vector<RowsetSharedPtr> to_delete;

        Version to_add_v = to_add_rs->version();

        // if start_version  > max_version, we can skip checking overlap here.

        if (to_add_v.first > _max_version) {

            // if start_version  > max_version, we can skip checking overlap here.

            to_add_directly.push_back(to_add_rs);

        } else {

            to_add_directly.push_back(to_add_rs);

            for (auto& [v, rs] : _rs_version_map) {

                if (to_add_v.contains(v)) {

                    to_delete.push_back(rs);

                }

            }

            delete_rowsets(to_delete, meta_lock);

        }

    }

    add_rowsets_directly(to_add_directly);

}

void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,

                                 std::unique_lock<std::shared_mutex>&) {

    if (to_delete.empty()) {

        return;

    }

    std::vector<RowsetMetaSharedPtr> rs_metas;

    rs_metas.reserve(to_delete.size());

    int64_t now = ::time(nullptr);

    for (auto&& rs : to_delete) {

        rs->rowset_meta()->set_stale_at(now);

        rs_metas.push_back(rs->rowset_meta());

        _stale_rs_version_map[rs->version()] = rs;

    }

    _timestamped_version_tracker.add_stale_path_version(rs_metas);

    for (auto&& rs : to_delete) {

        _rs_version_map.erase(rs->version());

    }

    _tablet_meta->modify_rs_metas({}, rs_metas, false);

}

uint64_t CloudTablet::delete_expired_stale_rowsets() {

    if (config::enable_mow_verbose_log) {

        LOG_INFO("begin delete_expired_stale_rowset for tablet={}", tablet_id());

    }

    std::vector<RowsetSharedPtr> expired_rowsets;

    // ATTN: trick, Use stale_rowsets to temporarily increase the reference count of the rowset shared pointer in _stale_rs_version_map so that in the recycle_cached_data function, it checks if the reference count is 2.

    std::vector<std::pair<Version, std::vector<RowsetSharedPtr>>> deleted_stale_rowsets;

    int64_t expired_stale_sweep_endtime =

            ::time(nullptr) - config::tablet_rowset_stale_sweep_time_sec;

    {

        std::unique_lock wlock(_meta_lock);

        std::vector<int64_t> path_ids;

        // capture the path version to delete

        _timestamped_version_tracker.capture_expired_paths(expired_stale_sweep_endtime, &path_ids);

        if (path_ids.empty()) {

            return 0;

        }

        for (int64_t path_id : path_ids) {

            int64_t start_version = -1;

            int64_t end_version = -1;

            std::vector<RowsetSharedPtr> stale_rowsets;

            // delete stale versions in version graph

            auto version_path = _timestamped_version_tracker.fetch_and_delete_path_by_id(path_id);

            for (auto& v_ts : version_path->timestamped_versions()) {

                auto rs_it = _stale_rs_version_map.find(v_ts->version());

                if (rs_it != _stale_rs_version_map.end()) {

                    expired_rowsets.push_back(rs_it->second);

                    stale_rowsets.push_back(rs_it->second);

                    VLOG_DEBUG << "erase stale rowset, tablet_id=" << tablet_id()

                               << " rowset_id=" << rs_it->second->rowset_id().to_string()

                               << " version=" << rs_it->first.to_string();

                    _stale_rs_version_map.erase(rs_it);

                } else {

                    LOG(WARNING) << "cannot find stale rowset " << v_ts->version() << " in tablet "

                                 << tablet_id();

                    // clang-format off

                    DCHECK(false) << [this, &wlock]() { wlock.unlock(); std::string json; get_compaction_status(&json); return json; }();

                    // clang-format on

                }

                if (start_version < 0) {

                    start_version = v_ts->version().first;

                }

                end_version = v_ts->version().second;

                _tablet_meta->delete_stale_rs_meta_by_version(v_ts->version());

            }

            Version version(start_version, end_version);

            if (!stale_rowsets.empty()) {

                deleted_stale_rowsets.emplace_back(version, std::move(stale_rowsets));

            }

        }

        _reconstruct_version_tracker_if_necessary();

    }

    // if the rowset is not used by any query, we can recycle its cached data early.

    auto recycled_rowsets = recycle_cached_data(expired_rowsets);

    if (!recycled_rowsets.empty()) {

        auto& manager = ExecEnv::GetInstance()->storage_engine().to_cloud().cloud_warm_up_manager();

        manager.recycle_cache(tablet_id(), recycled_rowsets);

    }

    if (config::enable_mow_verbose_log) {

        LOG_INFO("finish delete_expired_stale_rowset for tablet={}", tablet_id());

    }

    add_unused_rowsets(expired_rowsets);

    if (config::enable_agg_and_remove_pre_rowsets_delete_bitmap && keys_type() == UNIQUE_KEYS &&

        enable_unique_key_merge_on_write() && !deleted_stale_rowsets.empty()) {

        // agg delete bitmap for pre rowsets; record unused delete bitmap key ranges

        OlapStopWatch watch;

        for (const auto& [version, unused_rowsets] : deleted_stale_rowsets) {

            // agg delete bitmap for pre rowset

            DeleteBitmapKeyRanges remove_delete_bitmap_key_ranges;

            agg_delete_bitmap_for_stale_rowsets(version, remove_delete_bitmap_key_ranges);

            // add remove delete bitmap

            if (!remove_delete_bitmap_key_ranges.empty()) {

                std::vector<RowsetId> rowset_ids;

                for (const auto& rs : unused_rowsets) {

                    rowset_ids.push_back(rs->rowset_id());

                }

                std::lock_guard<std::mutex> lock(_gc_mutex);

                _unused_delete_bitmap.push_back(

                        std::make_pair(rowset_ids, remove_delete_bitmap_key_ranges));

            }

        }

        LOG(INFO) << "agg pre rowsets delete bitmap. tablet_id=" << tablet_id()

                  << ", size=" << deleted_stale_rowsets.size()

                  << ", cost(us)=" << watch.get_elapse_time_us();

    }

    return expired_rowsets.size();

}

bool CloudTablet::need_remove_unused_rowsets() {

    std::lock_guard<std::mutex> lock(_gc_mutex);

    return !_unused_rowsets.empty() || !_unused_delete_bitmap.empty();

}

void CloudTablet::add_unused_rowsets(const std::vector<RowsetSharedPtr>& rowsets) {

    std::lock_guard<std::mutex> lock(_gc_mutex);

    for (const auto& rowset : rowsets) {

        _unused_rowsets[rowset->rowset_id()] = rowset;

        g_unused_rowsets_bytes << rowset->total_disk_size();

    }

    g_unused_rowsets_count << rowsets.size();

}

void CloudTablet::remove_unused_rowsets() {

    std::vector<std::shared_ptr<Rowset>> removed_rowsets;

    int64_t removed_delete_bitmap_num = 0;

    OlapStopWatch watch;

    {

        std::lock_guard<std::mutex> lock(_gc_mutex);

        // 1. remove unused rowsets's cache data and delete bitmap

        for (auto it = _unused_rowsets.begin(); it != _unused_rowsets.end();) {

            auto& rs = it->second;

            if (rs.use_count() > 1) {

                LOG(WARNING) << "tablet_id:" << tablet_id() << " rowset: " << rs->rowset_id()

                             << " has " << rs.use_count() << " references, it cannot be removed";

                ++it;

                continue;

            }

            tablet_meta()->remove_rowset_delete_bitmap(rs->rowset_id(), rs->version());

            _rowset_warm_up_states.erase(rs->rowset_id());

            rs->clear_cache();

            g_unused_rowsets_count << -1;

            g_unused_rowsets_bytes << -rs->total_disk_size();

            removed_rowsets.push_back(std::move(rs));

            it = _unused_rowsets.erase(it);

        }

    }

    {

        std::vector<RecycledRowsets> recycled_rowsets;

        for (auto& rs : removed_rowsets) {

            auto index_names = rs->get_index_file_names();

            recycled_rowsets.emplace_back(rs->rowset_id(), rs->num_segments(), index_names);

            int64_t segment_size_sum = 0;

            for (int32_t i = 0; i < rs->num_segments(); i++) {

                segment_size_sum += rs->rowset_meta()->segment_file_size(i);

            }

            g_file_cache_recycle_cached_data_segment_num << rs->num_segments();

            g_file_cache_recycle_cached_data_segment_size << segment_size_sum;

            g_file_cache_recycle_cached_data_index_num << index_names.size();

        }

        if (recycled_rowsets.size() > 0) {

            auto& manager =

                    ExecEnv::GetInstance()->storage_engine().to_cloud().cloud_warm_up_manager();

            manager.recycle_cache(tablet_id(), recycled_rowsets);

        }

    }

    {

        std::lock_guard<std::mutex> lock(_gc_mutex);

        // 2. remove delete bitmap of pre rowsets

        for (auto it = _unused_delete_bitmap.begin(); it != _unused_delete_bitmap.end();) {

            auto& rowset_ids = std::get<0>(*it);

            bool find_unused_rowset = false;

            for (const auto& rowset_id : rowset_ids) {

                if (_unused_rowsets.find(rowset_id) != _unused_rowsets.end()) {

                    LOG(INFO) << "can not remove pre rowset delete bitmap because rowset is in use"

                              << ", tablet_id=" << tablet_id() << ", rowset_id=" << rowset_id;

                    find_unused_rowset = true;

                    break;

                }

            }

            if (find_unused_rowset) {

                ++it;

                continue;

            }

            auto& key_ranges = std::get<1>(*it);

            tablet_meta()->delete_bitmap().remove(key_ranges);

            it = _unused_delete_bitmap.erase(it);

            removed_delete_bitmap_num++;

            // TODO(kaijie): recycle cache for unused delete bitmap

        }

    }

    LOG(INFO) << "tablet_id=" << tablet_id() << ", unused_rowset size=" << _unused_rowsets.size()

              << ", unused_delete_bitmap size=" << _unused_delete_bitmap.size()

              << ", removed_rowsets_num=" << removed_rowsets.size()

              << ", removed_delete_bitmap_num=" << removed_delete_bitmap_num

              << ", cost(us)=" << watch.get_elapse_time_us();

}

void CloudTablet::update_base_size(const Rowset& rs) {

    // Define base rowset as the rowset of version [2-x]

    if (rs.start_version() == 2) {

        _base_size = rs.total_disk_size();

    }

}

void CloudTablet::clear_cache() {

    auto recycled_rowsets = CloudTablet::recycle_cached_data(get_snapshot_rowset(true));

    if (!recycled_rowsets.empty()) {

        auto& manager = ExecEnv::GetInstance()->storage_engine().to_cloud().cloud_warm_up_manager();

        manager.recycle_cache(tablet_id(), recycled_rowsets);

    }

    _engine.tablet_mgr().erase_tablet(tablet_id());

}

std::vector<RecycledRowsets> CloudTablet::recycle_cached_data(

        const std::vector<RowsetSharedPtr>& rowsets) {

    std::vector<RecycledRowsets> recycled_rowsets;

    for (const auto& rs : rowsets) {

        // rowsets and tablet._rs_version_map each hold a rowset shared_ptr, so at this point, the reference count of the shared_ptr is at least 2.

        if (rs.use_count() > 2) {

            LOG(WARNING) << "Rowset " << rs->rowset_id().to_string() << " has " << rs.use_count()

                         << " references. File Cache won't be recycled when query is using it.";

            continue;

        }

        rs->clear_cache();

        auto index_names = rs->get_index_file_names();

        recycled_rowsets.emplace_back(rs->rowset_id(), rs->num_segments(), index_names);

        int64_t segment_size_sum = 0;

        for (int32_t i = 0; i < rs->num_segments(); i++) {

            segment_size_sum += rs->rowset_meta()->segment_file_size(i);

        }

        g_file_cache_recycle_cached_data_segment_num << rs->num_segments();

        g_file_cache_recycle_cached_data_segment_size << segment_size_sum;

        g_file_cache_recycle_cached_data_index_num << index_names.size();

    }

    return recycled_rowsets;

}

void CloudTablet::reset_approximate_stats(int64_t num_rowsets, int64_t num_segments,

                                          int64_t num_rows, int64_t data_size) {

    _approximate_num_segments.store(num_segments, std::memory_order_relaxed);

    _approximate_num_rows.store(num_rows, std::memory_order_relaxed);

    _approximate_data_size.store(data_size, std::memory_order_relaxed);

    int64_t cumu_num_deltas = 0;

    int64_t cumu_num_rowsets = 0;

    auto cp = _cumulative_point.load(std::memory_order_relaxed);

    for (auto& [v, r] : _rs_version_map) {

        if (v.second < cp) {

            continue;

        }

        cumu_num_deltas += r->is_segments_overlapping() ? r->num_segments() : 1;

        ++cumu_num_rowsets;

    }

    // num_rowsets may be less than the size of _rs_version_map when there are some hole rowsets

    // in the version map, so we use the max value to ensure that the approximate number

    // of rowsets is at least the size of _rs_version_map.

    // Note that this is not the exact number of rowsets, but an approximate number.

    int64_t approximate_num_rowsets =

            std::max(num_rowsets, static_cast<int64_t>(_rs_version_map.size()));

    _approximate_num_rowsets.store(approximate_num_rowsets, std::memory_order_relaxed);

    _approximate_cumu_num_rowsets.store(cumu_num_rowsets, std::memory_order_relaxed);

    _approximate_cumu_num_deltas.store(cumu_num_deltas, std::memory_order_relaxed);

}

Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer(

        RowsetWriterContext& context, bool vertical) {

    context.rowset_id = _engine.next_rowset_id();

    // FIXME(plat1ko): Seems `tablet_id` and `index_id` has been set repeatedly

    context.tablet_id = tablet_id();

    context.index_id = index_id();

    context.partition_id = partition_id();

    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();

    context.encrypt_algorithm = tablet_meta()->encryption_algorithm();

    return RowsetFactory::create_rowset_writer(_engine, context, vertical);

}

// create a rowset writer with rowset_id and seg_id

// after writer, merge this transient rowset with original rowset

Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_writer(

        const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,

        int64_t txn_expiration) {

    if (rowset.rowset_meta_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE &&

        rowset.rowset_meta_state() != RowsetStatePB::COMMITTED) [[unlikely]] {

        auto msg = fmt::format(

                "wrong rowset state when create_transient_rowset_writer, rowset state should be "

                "BEGIN_PARTIAL_UPDATE or COMMITTED, but found {}, rowset_id={}, tablet_id={}",

                RowsetStatePB_Name(rowset.rowset_meta_state()), rowset.rowset_id().to_string(),

                tablet_id());

        // see `CloudRowsetWriter::build` for detail.

        // if this is in a retry task, the rowset state may have been changed to RowsetStatePB::COMMITTED

        // in `RowsetMeta::merge_rowset_meta()` in previous trials.

        LOG(WARNING) << msg;

        DCHECK(false) << msg;

    }

    RowsetWriterContext context;

    context.rowset_state = PREPARED;

    context.segments_overlap = OVERLAPPING;

    // During a partial update, the extracted columns of a variant should not be included in the tablet schema.

    // This is because the partial update for a variant needs to ignore the extracted columns.

    // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update,

    // the complete variant is constructed by reading all the sub-columns of the variant.

    context.tablet_schema = rowset.tablet_schema()->copy_without_variant_extracted_columns();

    context.newest_write_timestamp = UnixSeconds();

    context.tablet_id = table_id();

    context.enable_segcompaction = false;

    context.write_type = DataWriteType::TYPE_DIRECT;

    context.partial_update_info = std::move(partial_update_info);

    context.is_transient_rowset_writer = true;

    context.rowset_id = rowset.rowset_id();

    context.tablet_id = tablet_id();

    context.index_id = index_id();

    context.partition_id = partition_id();

    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();

    context.txn_expiration = txn_expiration;

    context.encrypt_algorithm = tablet_meta()->encryption_algorithm();

    // TODO(liaoxin) enable packed file for transient rowset

    context.allow_packed_file = false;

    auto storage_resource = rowset.rowset_meta()->remote_storage_resource();

    if (!storage_resource) {

        return ResultError(std::move(storage_resource.error()));

    }

    context.storage_resource = *storage_resource.value();

    return RowsetFactory::create_rowset_writer(_engine, context, false)

            .transform([&](auto&& writer) {

                writer->set_segment_start_id(cast_set<int32_t>(rowset.num_segments()));

                return writer;

            });

}

int64_t CloudTablet::get_cloud_base_compaction_score() const {

    if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {

        bool has_delete = false;

        int64_t point = cumulative_layer_point();

        std::shared_lock<std::shared_mutex> rlock(_meta_lock);