// 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 "olap/tablet.h"

#include <butil/logging.h>

#include <bvar/reducer.h>

#include <bvar/window.h>

#include <fmt/format.h>

#include <gen_cpp/FrontendService_types.h>

#include <gen_cpp/MasterService_types.h>

#include <gen_cpp/Metrics_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <gen_cpp/types.pb.h>

#include <glog/logging.h>

#include <rapidjson/document.h>

#include <rapidjson/encodings.h>

#include <rapidjson/prettywriter.h>

#include <rapidjson/rapidjson.h>

#include <rapidjson/stringbuffer.h>

#include <algorithm>

#include <atomic>

#include <boost/container/detail/std_fwd.hpp>

#include <cstdint>

#include <roaring/roaring.hh>

#include "common/compiler_util.h" // IWYU pragma: keep

// IWYU pragma: no_include <bits/chrono.h>

#include <chrono> // IWYU pragma: keep

#include <filesystem>

#include <iterator>

#include <limits>

#include <map>

#include <memory>

#include <mutex>

#include <set>

#include <shared_mutex>

#include <string>

#include <tuple>

#include <type_traits>

#include <unordered_map>

#include <unordered_set>

#include "agent/utils.h"

#include "common/config.h"

#include "common/consts.h"

#include "common/logging.h"

#include "common/signal_handler.h"

#include "common/status.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_reader_writer_fwd.h"

#include "io/fs/file_system.h"

#include "io/fs/file_writer.h"

#include "io/fs/path.h"

#include "io/fs/remote_file_system.h"

#include "io/io_common.h"

#include "olap/base_compaction.h"

#include "olap/base_tablet.h"

#include "olap/binlog.h"

#include "olap/cumulative_compaction.h"

#include "olap/cumulative_compaction_policy.h"

#include "olap/cumulative_compaction_time_series_policy.h"

#include "olap/delete_bitmap_calculator.h"

#include "olap/full_compaction.h"

#include "olap/memtable.h"

#include "olap/olap_common.h"

#include "olap/olap_define.h"

#include "olap/olap_meta.h"

#include "olap/primary_key_index.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_meta.h"

#include "olap/rowset/rowset_meta_manager.h"

#include "olap/rowset/rowset_writer.h"

#include "olap/rowset/rowset_writer_context.h"

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

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

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

#include "olap/rowset/vertical_beta_rowset_writer.h"

#include "olap/schema_change.h"

#include "olap/single_replica_compaction.h"

#include "olap/storage_engine.h"

#include "olap/storage_policy.h"

#include "olap/tablet_manager.h"

#include "olap/tablet_meta.h"

#include "olap/tablet_meta_manager.h"

#include "olap/tablet_schema.h"

#include "olap/txn_manager.h"

#include "olap/types.h"

#include "olap/utils.h"

#include "segment_loader.h"

#include "service/point_query_executor.h"

#include "tablet.h"

#include "util/bvar_helper.h"

#include "util/debug_points.h"

#include "util/defer_op.h"

#include "util/doris_metrics.h"

#include "util/pretty_printer.h"

#include "util/stopwatch.hpp"

#include "util/threadpool.h"

#include "util/time.h"

#include "util/trace.h"

#include "util/uid_util.h"

#include "util/work_thread_pool.hpp"

#include "vec/columns/column.h"

#include "vec/columns/column_string.h"

#include "vec/common/schema_util.h"

#include "vec/common/string_ref.h"

#include "vec/data_types/data_type.h"

#include "vec/data_types/data_type_factory.hpp"

#include "vec/data_types/serde/data_type_serde.h"

#include "vec/jsonb/serialize.h"

namespace doris {

#include "common/compile_check_begin.h"

class TupleDescriptor;

namespace vectorized {

class Block;

} // namespace vectorized

using namespace ErrorCode;

using namespace std::chrono_literals;

using std::pair;

using std::string;

using std::vector;

using io::FileSystemSPtr;

namespace {

bvar::Adder<uint64_t> exceed_version_limit_counter;

bvar::Window<bvar::Adder<uint64_t>> exceed_version_limit_counter_minute(

        &exceed_version_limit_counter, 60);

bvar::Adder<uint64_t> cooldown_pending_task("cooldown_pending_task");

bvar::Adder<uint64_t> cooldown_processing_task("cooldown_processing_task");

void set_last_failure_time(Tablet* tablet, const Compaction& compaction, int64_t ms) {

    switch (compaction.compaction_type()) {

    case ReaderType::READER_CUMULATIVE_COMPACTION:

        tablet->set_last_cumu_compaction_failure_time(ms);

        return;

    case ReaderType::READER_BASE_COMPACTION:

        tablet->set_last_base_compaction_failure_time(ms);

        return;

    case ReaderType::READER_FULL_COMPACTION:

        tablet->set_last_full_compaction_failure_time(ms);

        return;

    default:

        LOG(FATAL) << "invalid compaction type " << compaction.compaction_name()

                   << " tablet_id: " << tablet->tablet_id();

    }

};

} // namespace

bvar::Adder<uint64_t> unused_remote_rowset_num("unused_remote_rowset_num");

WriteCooldownMetaExecutors::WriteCooldownMetaExecutors(size_t executor_nums)

        : _executor_nums(executor_nums) {

    for (size_t i = 0; i < _executor_nums; i++) {

        std::unique_ptr<PriorityThreadPool> pool;

        static_cast<void>(ThreadPoolBuilder("WriteCooldownMetaExecutor")

                                  .set_min_threads(1)

                                  .set_max_threads(1)

                                  .set_max_queue_size(std::numeric_limits<int>::max())

                                  .build(&pool));

        _executors.emplace_back(std::move(pool));

    }

}

void WriteCooldownMetaExecutors::stop() {

    for (auto& pool_ptr : _executors) {

        if (pool_ptr) {

            pool_ptr->shutdown();

        }

    }

}

void WriteCooldownMetaExecutors::WriteCooldownMetaExecutors::submit(TabletSharedPtr tablet) {

    auto tablet_id = tablet->tablet_id();

    {

        std::shared_lock rdlock(tablet->get_header_lock());

        if (!tablet->tablet_meta()->cooldown_meta_id().initialized()) {

            VLOG_NOTICE << "tablet " << tablet_id << " is not cooldown replica";

            return;

        }

        if (tablet->tablet_state() == TABLET_SHUTDOWN) [[unlikely]] {

            LOG_INFO("tablet {} has been dropped, don't do cooldown", tablet_id);

            return;

        }

    }

    {

        // one tablet could at most have one cooldown task to be done

        std::unique_lock<std::mutex> lck {_latch};

        if (_pending_tablets.count(tablet_id) > 0) {

            return;

        }

        _pending_tablets.insert(tablet_id);

    }

    auto async_write_task = [this, t = std::move(tablet)]() {

        {

            std::unique_lock<std::mutex> lck {_latch};

            _pending_tablets.erase(t->tablet_id());

        }

        SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->orphan_mem_tracker());

        auto s = t->write_cooldown_meta();

        if (s.ok()) {

            return;

        }

        if (!s.is<ABORTED>()) {

            LOG_EVERY_SECOND(WARNING)

                    << "write tablet " << t->tablet_id() << " cooldown meta failed because: " << s;

            submit(t);

            return;

        }

        VLOG_DEBUG << "tablet " << t->tablet_id() << " is not cooldown replica";

    };

    cooldown_pending_task << 1;

    _executors[_get_executor_pos(tablet_id)]->offer([task = std::move(async_write_task)]() {

        cooldown_pending_task << -1;

        cooldown_processing_task << 1;

        task();

        cooldown_processing_task << -1;

    });

}

Tablet::Tablet(StorageEngine& engine, TabletMetaSharedPtr tablet_meta, DataDir* data_dir,

               const std::string_view& cumulative_compaction_type)

        : BaseTablet(std::move(tablet_meta)),

          _engine(engine),

          _data_dir(data_dir),

          _is_bad(false),

          _last_cumu_compaction_failure_millis(0),

          _last_base_compaction_failure_millis(0),

          _last_full_compaction_failure_millis(0),

          _last_cumu_compaction_success_millis(0),

          _last_base_compaction_success_millis(0),

          _last_full_compaction_success_millis(0),

          _cumulative_point(K_INVALID_CUMULATIVE_POINT),

          _newly_created_rowset_num(0),

          _last_checkpoint_time(0),

          _cumulative_compaction_type(cumulative_compaction_type),

          _is_tablet_path_exists(true),

          _last_missed_version(-1),

          _last_missed_time_s(0) {

    if (_data_dir != nullptr) {

        _tablet_path = fmt::format("{}/{}/{}/{}/{}", _data_dir->path(), DATA_PREFIX,

                                   _tablet_meta->shard_id(), tablet_id(), schema_hash());

    }

}

bool Tablet::set_tablet_schema_into_rowset_meta() {

    bool flag = false;

    for (auto&& rowset_meta : _tablet_meta->all_mutable_rs_metas()) {

        if (!rowset_meta->tablet_schema()) {

            rowset_meta->set_tablet_schema(_tablet_meta->tablet_schema());

            flag = true;

        }

    }

    return flag;

}

Status Tablet::_init_once_action() {

    Status res = Status::OK();

    VLOG_NOTICE << "begin to load tablet. tablet=" << tablet_id()

                << ", version_size=" << _tablet_meta->version_count();

#ifdef BE_TEST

    // init cumulative compaction policy by type

    _cumulative_compaction_policy =

            CumulativeCompactionPolicyFactory::create_cumulative_compaction_policy(

                    _tablet_meta->compaction_policy());

#endif

    for (const auto& rs_meta : _tablet_meta->all_rs_metas()) {

        Version version = rs_meta->version();

        RowsetSharedPtr rowset;

        res = create_rowset(rs_meta, &rowset);

        if (!res.ok()) {

            LOG(WARNING) << "fail to init rowset. tablet_id=" << tablet_id()

                         << ", schema_hash=" << schema_hash() << ", version=" << version

                         << ", res=" << res;

            return res;

        }

        _rs_version_map[version] = std::move(rowset);

    }

    // init stale rowset

    for (const auto& stale_rs_meta : _tablet_meta->all_stale_rs_metas()) {

        Version version = stale_rs_meta->version();

        RowsetSharedPtr rowset;

        res = create_rowset(stale_rs_meta, &rowset);

        if (!res.ok()) {

            LOG(WARNING) << "fail to init stale rowset. tablet_id:" << tablet_id()

                         << ", schema_hash:" << schema_hash() << ", version=" << version

                         << ", res:" << res;

            return res;

        }

        _stale_rs_version_map[version] = std::move(rowset);

    }

    return res;

}

Status Tablet::init() {

    return _init_once.call([this] { return _init_once_action(); });

}

// should save tablet meta to remote meta store

// if it's a primary replica

void Tablet::save_meta() {

    check_table_size_correctness();

    auto res = _tablet_meta->save_meta(_data_dir);

    CHECK_EQ(res, Status::OK()) << "fail to save tablet_meta. res=" << res

                                << ", root=" << _data_dir->path();

}

// Caller should hold _meta_lock.

Status Tablet::revise_tablet_meta(const std::vector<RowsetSharedPtr>& to_add,

                                  const std::vector<RowsetSharedPtr>& to_delete,

                                  bool is_incremental_clone) {

    LOG(INFO) << "begin to revise tablet. tablet_id=" << tablet_id();

    // 1. for incremental clone, we have to add the rowsets first to make it easy to compute

    //    all the delete bitmaps, and it's easy to delete them if we end up with a failure

    // 2. for full clone, we can calculate delete bitmaps on the cloned rowsets directly.

    if (is_incremental_clone) {

        CHECK(to_delete.empty()); // don't need to delete rowsets

        add_rowsets(to_add);

        // reconstruct from tablet meta

        _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas());

    }

    Status calc_bm_status;

    std::vector<RowsetSharedPtr> base_rowsets_for_full_clone = to_add; // copy vector

    while (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {

        std::vector<RowsetSharedPtr> calc_delete_bitmap_rowsets;

        int64_t to_add_min_version = INT64_MAX;

        int64_t to_add_max_version = INT64_MIN;

        for (auto& rs : to_add) {

            if (to_add_min_version > rs->start_version()) {

                to_add_min_version = rs->start_version();

            }

            if (to_add_max_version < rs->end_version()) {

                to_add_max_version = rs->end_version();

            }

        }

        Version calc_delete_bitmap_ver;

        if (is_incremental_clone) {

            // From the rowset of to_add with smallest version, all other rowsets

            // need to recalculate the delete bitmap

            // For example:

            // local tablet: [0-1] [2-5] [6-6] [9-10]

            // clone tablet: [7-7] [8-8]

            // new tablet:   [0-1] [2-5] [6-6] [7-7] [8-8] [9-10]

            // [7-7] [8-8] [9-10] need to recalculate delete bitmap

            calc_delete_bitmap_ver = Version(to_add_min_version, max_version_unlocked());

        } else {

            // the delete bitmap of to_add's rowsets has clone from remote when full clone.

            // only other rowsets in local need to recalculate the delete bitmap.

            // For example:

            // local tablet: [0-1]x [2-5]x [6-6]x [7-7]x [9-10]

            // clone tablet: [0-1]  [2-4]  [5-6]  [7-8]

            // new tablet:   [0-1]  [2-4]  [5-6]  [7-8] [9-10]

            // only [9-10] need to recalculate delete bitmap

            CHECK_EQ(to_add_min_version, 0) << "to_add_min_version is: " << to_add_min_version;

            calc_delete_bitmap_ver = Version(to_add_max_version + 1, max_version_unlocked());

        }

        if (calc_delete_bitmap_ver.first <= calc_delete_bitmap_ver.second) {

            calc_bm_status = capture_consistent_rowsets_unlocked(calc_delete_bitmap_ver,

                                                                 &calc_delete_bitmap_rowsets);

            if (!calc_bm_status.ok()) {

                LOG(WARNING) << "fail to capture_consistent_rowsets, res: " << calc_bm_status;

                break;

            }

            // FIXME(plat1ko): Use `const TabletSharedPtr&` as parameter

            auto self = _engine.tablet_manager()->get_tablet(tablet_id());

            CHECK(self);

            for (auto rs : calc_delete_bitmap_rowsets) {

                if (is_incremental_clone) {

                    calc_bm_status = update_delete_bitmap_without_lock(self, rs);

                } else {

                    calc_bm_status = update_delete_bitmap_without_lock(

                            self, rs, &base_rowsets_for_full_clone);

                    base_rowsets_for_full_clone.push_back(rs);

                }

                if (!calc_bm_status.ok()) {

                    LOG(WARNING) << "fail to update_delete_bitmap_without_lock, res: "

                                 << calc_bm_status;

                    break;

                }

            }

        }

        break; // while (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write())

    }

    DBUG_EXECUTE_IF("Tablet.revise_tablet_meta_fail", {

        auto ptablet_id = dp->param("tablet_id", 0);

        if (tablet_id() == ptablet_id) {

            LOG(INFO) << "injected revies_tablet_meta failure for tabelt: " << ptablet_id;

            calc_bm_status = Status::InternalError("fault injection error");

        }

    });

    // error handling

    if (!calc_bm_status.ok()) {

        if (is_incremental_clone) {

            RETURN_IF_ERROR(delete_rowsets(to_add, false));

            LOG(WARNING) << "incremental clone on tablet: " << tablet_id() << " failed due to "

                         << calc_bm_status.msg() << ", revert " << to_add.size()

                         << " rowsets added before.";

        } else {

            LOG(WARNING) << "full clone on tablet: " << tablet_id() << " failed due to "

                         << calc_bm_status.msg() << ", will not update tablet meta.";

        }

        return calc_bm_status;

    }

    // full clone, calculate delete bitmap succeeded, update rowset

    if (!is_incremental_clone) {

        RETURN_IF_ERROR(delete_rowsets(to_delete, false));

        add_rowsets(to_add);

        // reconstruct from tablet meta

        _timestamped_version_tracker.construct_versioned_tracker(_tablet_meta->all_rs_metas());

        // check the rowsets used for delete bitmap calculation is equal to the rowsets

        // that we can capture by version

        if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {

            Version full_version = Version(0, max_version_unlocked());

            std::vector<RowsetSharedPtr> expected_rowsets;

            auto st = capture_consistent_rowsets_unlocked(full_version, &expected_rowsets);

            DCHECK(st.ok()) << st;

            DCHECK_EQ(base_rowsets_for_full_clone.size(), expected_rowsets.size());

            if (st.ok() && base_rowsets_for_full_clone.size() != expected_rowsets.size())

                    [[unlikely]] {

                LOG(WARNING) << "full clone succeeded, but the count("

                             << base_rowsets_for_full_clone.size()

                             << ") of base rowsets used for delete bitmap calculation is not match "

                                "expect count("

                             << expected_rowsets.size() << ") we capture from tablet meta";

            }

        }

    }

    // clear stale rowset

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

        _engine.add_unused_rowset(rs);

    }

    _stale_rs_version_map.clear();

    _tablet_meta->clear_stale_rowset();

    save_meta();

    LOG(INFO) << "finish to revise tablet. tablet_id=" << tablet_id();

    return Status::OK();

}

Status Tablet::add_rowset(RowsetSharedPtr rowset) {

    DCHECK(rowset != nullptr);

    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);

    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);

    // If the rowset already exist, just return directly.  The rowset_id is an unique-id,

    // we can use it to check this situation.

    if (_contains_rowset(rowset->rowset_id())) {

        return Status::OK();

    }

    // Otherwise, the version should be not contained in any existing rowset.

    RETURN_IF_ERROR(_contains_version(rowset->version()));

    RETURN_IF_ERROR(_tablet_meta->add_rs_meta(rowset->rowset_meta()));

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

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

    add_compaction_score(rowset->rowset_meta()->get_compaction_score());

    std::vector<RowsetSharedPtr> rowsets_to_delete;

    // yiguolei: temp code, should remove the rowset contains by this rowset

    // but it should be removed in multi path version

    for (auto& it : _rs_version_map) {

        if (rowset->version().contains(it.first) && rowset->version() != it.first) {

            CHECK(it.second != nullptr)

                    << "there exist a version=" << it.first

                    << " contains the input rs with version=" << rowset->version()

                    << ", but the related rs is null";

            rowsets_to_delete.push_back(it.second);

        }

    }

    std::vector<RowsetSharedPtr> empty_vec;

    RETURN_IF_ERROR(modify_rowsets(empty_vec, rowsets_to_delete));

    ++_newly_created_rowset_num;

    return Status::OK();

}

bool Tablet::rowset_exists_unlocked(const RowsetSharedPtr& rowset) {

    if (auto it = _rs_version_map.find(rowset->version()); it == _rs_version_map.end()) {

        return false;

    } else if (rowset->rowset_id() != it->second->rowset_id()) {

        return false;

    }

    return true;

}

Status Tablet::modify_rowsets(std::vector<RowsetSharedPtr>& to_add,

                              std::vector<RowsetSharedPtr>& to_delete, bool check_delete) {

    // the compaction process allow to compact the single version, eg: version[4-4].

    // this kind of "single version compaction" has same "input version" and "output version".

    // which means "to_add->version()" equals to "to_delete->version()".

    // So we should delete the "to_delete" before adding the "to_add",

    // otherwise, the "to_add" will be deleted from _rs_version_map, eventually.

    //

    // And if the version of "to_add" and "to_delete" are exactly same. eg:

    // to_add:      [7-7]

    // to_delete:   [7-7]

    // In this case, we no longer need to add the rowset in "to_delete" to

    // _stale_rs_version_map, but can delete it directly.

    if (to_add.empty() && to_delete.empty()) {

        return Status::OK();

    }

    if (check_delete) {

        for (auto&& rs : to_delete) {

            if (auto it = _rs_version_map.find(rs->version()); it == _rs_version_map.end()) {

                return Status::Error<DELETE_VERSION_ERROR>(

                        "try to delete not exist version {} from {}", rs->version().to_string(),

                        tablet_id());

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

                return Status::Error<DELETE_VERSION_ERROR>(

                        "try to delete version {} from {}, but rowset id changed, delete rowset id "

                        "is {}, exists rowsetid is {}",

                        rs->version().to_string(), tablet_id(), rs->rowset_id().to_string(),

                        it->second->rowset_id().to_string());

            }

        }

    }

    bool same_version = true;

    std::sort(to_add.begin(), to_add.end(), Rowset::comparator);

    std::sort(to_delete.begin(), to_delete.end(), Rowset::comparator);

    if (to_add.size() == to_delete.size()) {

        for (int i = 0; i < to_add.size(); ++i) {

            if (to_add[i]->version() != to_delete[i]->version()) {

                same_version = false;

                break;

            }

        }

    } else {

        same_version = false;

    }

    std::vector<RowsetMetaSharedPtr> rs_metas_to_delete;

    for (auto& rs : to_delete) {

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

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

        if (!same_version) {

            // put compaction rowsets in _stale_rs_version_map.

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

        }

    }

    std::vector<RowsetMetaSharedPtr> rs_metas_to_add;

    for (auto& rs : to_add) {

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

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

        if (!same_version) {

            // If version are same, then _timestamped_version_tracker

            // already has this version, no need to add again.

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

        }

        ++_newly_created_rowset_num;

    }

    _tablet_meta->modify_rs_metas(rs_metas_to_add, rs_metas_to_delete, same_version);

    if (!same_version) {

        // add rs_metas_to_delete to tracker

        _timestamped_version_tracker.add_stale_path_version(rs_metas_to_delete);

    } else {

        // delete rowset in "to_delete" directly

        for (auto& rs : to_delete) {

            LOG(INFO) << "add unused rowset " << rs->rowset_id() << " because of same version";

            if (rs->is_local()) {

                _engine.add_unused_rowset(rs);

            }

        }

    }

    int32_t add_score = 0;

    for (auto rs : to_add) {

        add_score += rs->rowset_meta()->get_compaction_score();

    }

    int32_t sub_score = 0;

    for (auto rs : to_delete) {

        sub_score += rs->rowset_meta()->get_compaction_score();

    }

    add_compaction_score(add_score - sub_score);

    return Status::OK();

}

void Tablet::add_rowsets(const std::vector<RowsetSharedPtr>& to_add) {

    if (to_add.empty()) {

        return;

    }

    std::vector<RowsetMetaSharedPtr> rs_metas;

    rs_metas.reserve(to_add.size());

    for (auto& rs : to_add) {

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

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

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

    }

    _tablet_meta->modify_rs_metas(rs_metas, {});

}

Status Tablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete, bool move_to_stale) {

    if (to_delete.empty()) {

        return Status::OK();

    }

    std::vector<RowsetMetaSharedPtr> rs_metas;

    rs_metas.reserve(to_delete.size());

    for (const auto& rs : to_delete) {

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

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

    }

    _tablet_meta->modify_rs_metas({}, rs_metas, !move_to_stale);

    if (move_to_stale) {

        for (const auto& rs : to_delete) {

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

        }

        _timestamped_version_tracker.add_stale_path_version(rs_metas);

    } else {

        for (const auto& rs : to_delete) {

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

            if (rs->is_local()) {

                _engine.add_unused_rowset(rs);

                RETURN_IF_ERROR(RowsetMetaManager::remove(_data_dir->get_meta(), tablet_uid(),

                                                          rs->rowset_meta()->rowset_id()));

            }

        }

    }

    return Status::OK();

}

RowsetSharedPtr Tablet::_rowset_with_largest_size() {

    RowsetSharedPtr largest_rowset = nullptr;

    for (auto& it : _rs_version_map) {

        if (it.second->empty() || it.second->zero_num_rows()) {

            continue;

        }

        if (largest_rowset == nullptr || it.second->rowset_meta()->index_disk_size() >

                                                 largest_rowset->rowset_meta()->index_disk_size()) {

            largest_rowset = it.second;

        }

    }

    return largest_rowset;

}

// add inc rowset should not persist tablet meta, because it will be persisted when publish txn.

Status Tablet::add_inc_rowset(const RowsetSharedPtr& rowset) {

    DCHECK(rowset != nullptr);

    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);

    SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);

    if (_contains_rowset(rowset->rowset_id())) {

        return Status::OK();

    }

    RETURN_IF_ERROR(_contains_version(rowset->version()));

    RETURN_IF_ERROR(_tablet_meta->add_rs_meta(rowset->rowset_meta()));

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

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

    ++_newly_created_rowset_num;

    add_compaction_score(rowset->rowset_meta()->get_compaction_score());

    return Status::OK();

}

void Tablet::_delete_stale_rowset_by_version(const Version& version) {

    RowsetMetaSharedPtr rowset_meta = _tablet_meta->acquire_stale_rs_meta_by_version(version);

    if (rowset_meta == nullptr) {

        return;

    }

    _tablet_meta->delete_stale_rs_meta_by_version(version);

    VLOG_NOTICE << "delete stale rowset. tablet=" << tablet_id() << ", version=" << version;

}

void Tablet::delete_expired_stale_rowset() {

    if (config::enable_mow_verbose_log) {

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

    }

    int64_t now = UnixSeconds();

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

    // hold write lock while processing stable rowset

    {

        std::lock_guard<std::shared_mutex> wrlock(_meta_lock);

        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);

        // Compute the end time to delete rowsets, when a expired rowset createtime less then this time, it will be deleted.

        int64_t expired_stale_sweep_endtime =

                static_cast<int64_t>(::difftime(now, config::tablet_rowset_stale_sweep_time_sec));

        if (config::tablet_rowset_stale_sweep_by_size) {

            expired_stale_sweep_endtime = now;

        }

        std::vector<int64_t> path_id_vec;

        // capture the path version to delete

        _timestamped_version_tracker.capture_expired_paths(

                static_cast<int64_t>(expired_stale_sweep_endtime), &path_id_vec);

        if (path_id_vec.empty()) {

            return;

        }

        const RowsetSharedPtr lastest_delta = get_rowset_with_max_version();

        if (lastest_delta == nullptr) {

            LOG(WARNING) << "lastest_delta is null " << tablet_id();

            return;

        }

        // fetch missing version before delete

        Versions missed_versions = get_missed_versions_unlocked(lastest_delta->end_version());

        if (!missed_versions.empty()) {

            LOG(WARNING) << "tablet:" << tablet_id()

                         << ", missed version for version:" << lastest_delta->end_version();

            _print_missed_versions(missed_versions);

            return;

        }

        // do check consistent operation

        auto path_id_iter = path_id_vec.begin();

        std::map<int64_t, PathVersionListSharedPtr> stale_version_path_map;

        while (path_id_iter != path_id_vec.end()) {

            PathVersionListSharedPtr version_path =

                    _timestamped_version_tracker.fetch_and_delete_path_by_id(*path_id_iter);

            Version test_version = Version(0, lastest_delta->end_version());

            stale_version_path_map[*path_id_iter] = version_path;

            Status status =

                    capture_consistent_versions_unlocked(test_version, nullptr, false, false);

            // 1. When there is no consistent versions, we must reconstruct the tracker.

            if (!status.ok()) {

                // 2. fetch missing version after delete

                Versions after_missed_versions =

                        get_missed_versions_unlocked(lastest_delta->end_version());

                // 2.1 check whether missed_versions and after_missed_versions are the same.

                // when they are the same, it means we can delete the path securely.

                bool is_missing = missed_versions.size() != after_missed_versions.size();

                if (!is_missing) {

                    for (int ver_index = 0; ver_index < missed_versions.size(); ver_index++) {

                        if (missed_versions[ver_index] != after_missed_versions[ver_index]) {

                            is_missing = true;

                            break;

                        }

                    }

                }

                if (is_missing) {

                    LOG(WARNING) << "The consistent version check fails, there are bugs. "

                                 << "Reconstruct the tracker to recover versions in tablet="

                                 << tablet_id();

                    // 3. try to recover

                    _timestamped_version_tracker.recover_versioned_tracker(stale_version_path_map);

                    // 4. double check the consistent versions

                    // fetch missing version after recover

                    Versions recover_missed_versions =

                            get_missed_versions_unlocked(lastest_delta->end_version());

                    // 4.1 check whether missed_versions and recover_missed_versions are the same.

                    // when they are the same, it means we recover successfully.

                    bool is_recover_missing =

                            missed_versions.size() != recover_missed_versions.size();

                    if (!is_recover_missing) {

                        for (int ver_index = 0; ver_index < missed_versions.size(); ver_index++) {

                            if (missed_versions[ver_index] != recover_missed_versions[ver_index]) {

                                is_recover_missing = true;

                                break;

                            }

                        }

                    }

                    // 5. check recover fail, version is mission

                    if (is_recover_missing) {

                        if (!config::ignore_rowset_stale_unconsistent_delete) {

                            LOG(FATAL)

                                    << "rowset stale unconsistent delete. tablet= " << tablet_id();

                        } else {

                            LOG(WARNING)

                                    << "rowset stale unconsistent delete. tablet= " << tablet_id();

                        }

                    }

                }

                return;

            }

            path_id_iter++;

        }

        auto old_size = _stale_rs_version_map.size();

        auto old_meta_size = _tablet_meta->all_stale_rs_metas().size();

        // do delete operation

        auto to_delete_iter = stale_version_path_map.begin();

        while (to_delete_iter != stale_version_path_map.end()) {

            std::vector<TimestampedVersionSharedPtr>& to_delete_version =

                    to_delete_iter->second->timestamped_versions();

            int64_t start_version = -1;

            int64_t end_version = -1;

            std::vector<RowsetId> remove_rowset_ids;

            for (auto& timestampedVersion : to_delete_version) {

                auto it = _stale_rs_version_map.find(timestampedVersion->version());

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

                    it->second->clear_cache();

                    // delete rowset

                    if (it->second->is_local()) {

                        _engine.add_unused_rowset(it->second);

                        if (keys_type() == UNIQUE_KEYS && enable_unique_key_merge_on_write()) {

                            // mow does not support cold data in object storage

                            remove_rowset_ids.emplace_back(it->second->rowset_id());

                        }

                    }

                    _stale_rs_version_map.erase(it);

                    VLOG_NOTICE << "delete stale rowset tablet=" << tablet_id() << " version["

                                << timestampedVersion->version().first << ","

                                << timestampedVersion->version().second

                                << "] move to unused_rowset success " << std::fixed

                                << expired_stale_sweep_endtime;

                } else {

                    LOG(WARNING) << "delete stale rowset tablet=" << tablet_id() << " version["

                                 << timestampedVersion->version().first << ","

                                 << timestampedVersion->version().second

                                 << "] not find in stale rs version map";

                }

                if (start_version < 0) {

                    start_version = timestampedVersion->version().first;

                }

                end_version = timestampedVersion->version().second;

                _delete_stale_rowset_by_version(timestampedVersion->version());

            }

            Version version(start_version, end_version);

            to_delete_iter++;

            if (!remove_rowset_ids.empty()) {

                deleted_stale_rowsets.emplace_back(version, remove_rowset_ids);

            }

        }

        bool reconstructed = _reconstruct_version_tracker_if_necessary();

        VLOG_NOTICE << "delete stale rowset _stale_rs_version_map tablet=" << tablet_id()

                    << " current_size=" << _stale_rs_version_map.size() << " old_size=" << old_size

                    << " current_meta_size=" << _tablet_meta->all_stale_rs_metas().size()

                    << " old_meta_size=" << old_meta_size << " sweep endtime " << std::fixed

                    << expired_stale_sweep_endtime << ", reconstructed=" << reconstructed;

    }

    if (config::enable_agg_and_remove_pre_rowsets_delete_bitmap && !deleted_stale_rowsets.empty()) {

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

        OlapStopWatch watch;

        for (const auto& [version, remove_rowset_ids] : 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()) {

                _engine.add_unused_delete_bitmap_key_ranges(tablet_id(), remove_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();

    }

#ifndef BE_TEST

    {

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

        save_meta();

    }

#endif

    if (config::enable_mow_verbose_log) {

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

    }

    DBUG_EXECUTE_IF("Tablet.delete_expired_stale_rowset.start_delete_unused_rowset",

                    { _engine.start_delete_unused_rowset(); });

}

Status Tablet::capture_consistent_versions_unlocked(const Version& spec_version,

                                                    Versions* version_path,

                                                    bool skip_missing_version, bool quiet) const {

    Status status =

            _timestamped_version_tracker.capture_consistent_versions(spec_version, version_path);

    if (!status.ok() && !quiet) {

        Versions missed_versions = get_missed_versions_unlocked(spec_version.second);

        if (missed_versions.empty()) {

            // if version_path is null, it may be a compaction check logic.

            // so to avoid print too many logs.

            if (version_path != nullptr) {

                LOG(WARNING) << "tablet:" << tablet_id()

                             << ", version already has been merged. spec_version: " << spec_version

                             << ", max_version: " << max_version_unlocked();

            }

            status = Status::Error<VERSION_ALREADY_MERGED, false>(

                    "versions are already compacted, spec_version "

                    "{}, max_version {}, tablet_id {}",

                    spec_version.second, max_version_unlocked(), tablet_id());

        } else {

            if (version_path != nullptr) {

                LOG(WARNING) << "status:" << status << ", tablet:" << tablet_id()

                             << ", missed version for version:" << spec_version;

                _print_missed_versions(missed_versions);

                if (skip_missing_version) {

                    LOG(WARNING) << "force skipping missing version for tablet:" << tablet_id();

                    return Status::OK();

                }

            }

        }

    }

    DBUG_EXECUTE_IF("TTablet::capture_consistent_versions.inject_failure", {

        auto tablet_id = dp->param<int64_t>("tablet_id", -1);

        if (tablet_id != -1 && tablet_id == _tablet_meta->tablet_id()) {

            status = Status::Error<VERSION_ALREADY_MERGED>("version already merged");

        }

    });

    return status;

}

Status Tablet::check_version_integrity(const Version& version, bool quiet) {

    std::shared_lock rdlock(_meta_lock);

    return capture_consistent_versions_unlocked(version, nullptr, false, quiet);

}

bool Tablet::exceed_version_limit(int32_t limit) {

    if (_tablet_meta->version_count() > limit) {

        exceed_version_limit_counter << 1;

        return true;

    }

    return false;

}

// If any rowset contains the specific version, it means the version already exist

bool Tablet::check_version_exist(const Version& version) const {

    std::shared_lock rdlock(_meta_lock);

    for (auto& it : _rs_version_map) {

        if (it.first.contains(version)) {