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

#include <fmt/format.h>

#include <gen_cpp/AgentService_types.h>

#include <gen_cpp/BackendService_types.h>

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/MasterService_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <re2/re2.h>

#include <unistd.h>

#include <algorithm>

#include <list>

#include <mutex>

#include <ostream>

// IWYU pragma: no_include <opentelemetry/common/threadlocal.h>

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

#include "common/config.h"

#include "common/logging.h"

#include "gutil/integral_types.h"

#include "gutil/strings/strcat.h"

#include "gutil/strings/substitute.h"

#include "io/fs/local_file_system.h"

#include "olap/cumulative_compaction_time_series_policy.h"

#include "olap/data_dir.h"

#include "olap/olap_common.h"

#include "olap/olap_define.h"

#include "olap/olap_meta.h"

#include "olap/pb_helper.h"

#include "olap/rowset/beta_rowset.h"

#include "olap/rowset/rowset.h"

#include "olap/rowset/rowset_meta_manager.h"

#include "olap/storage_engine.h"

#include "olap/tablet.h"

#include "olap/tablet_meta.h"

#include "olap/tablet_meta_manager.h"

#include "olap/tablet_schema.h"

#include "olap/txn_manager.h"

#include "runtime/exec_env.h"

#include "runtime/memory/mem_tracker.h"

#include "runtime/thread_context.h"

#include "service/backend_options.h"

#include "util/defer_op.h"

#include "util/doris_metrics.h"

#include "util/histogram.h"

#include "util/metrics.h"

#include "util/path_util.h"

#include "util/scoped_cleanup.h"

#include "util/stopwatch.hpp"

#include "util/time.h"

#include "util/trace.h"

#include "util/uid_util.h"

namespace doris {

class CumulativeCompactionPolicy;

} // namespace doris

using std::map;

using std::set;

using std::string;

using std::vector;

namespace doris {

using namespace ErrorCode;

DEFINE_GAUGE_METRIC_PROTOTYPE_5ARG(tablet_meta_mem_consumption, MetricUnit::BYTES, "",

                                   mem_consumption, Labels({{"type", "tablet_meta"}}));

TabletManager::TabletManager(int32_t tablet_map_lock_shard_size)

        : _mem_tracker(std::make_shared<MemTracker>(

                  "TabletManager", ExecEnv::GetInstance()->experimental_mem_tracker())),

          _tablet_meta_mem_tracker(std::make_shared<MemTracker>(

                  "TabletMeta", ExecEnv::GetInstance()->experimental_mem_tracker())),

          _tablets_shards_size(tablet_map_lock_shard_size),

          _tablets_shards_mask(tablet_map_lock_shard_size - 1) {

    CHECK_GT(_tablets_shards_size, 0);

    CHECK_EQ(_tablets_shards_size & _tablets_shards_mask, 0);

    _tablets_shards.resize(_tablets_shards_size);

    REGISTER_HOOK_METRIC(tablet_meta_mem_consumption,

                         [this]() { return _mem_tracker->consumption(); });

}

TabletManager::~TabletManager() {

    DEREGISTER_HOOK_METRIC(tablet_meta_mem_consumption);

}

Status TabletManager::_add_tablet_unlocked(TTabletId tablet_id, const TabletSharedPtr& tablet,

                                           bool update_meta, bool force, RuntimeProfile* profile) {

    if (profile->get_counter("AddTablet") == nullptr) {

        ADD_TIMER(profile, "AddTablet");

    }

    Status res = Status::OK();

    VLOG_NOTICE << "begin to add tablet to TabletManager. "

                << "tablet_id=" << tablet_id << ", force=" << force;

    TabletSharedPtr existed_tablet = nullptr;

    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);

    const auto& iter = tablet_map.find(tablet_id);

    if (iter != tablet_map.end()) {

        existed_tablet = iter->second;

    }

    if (existed_tablet == nullptr) {

        return _add_tablet_to_map_unlocked(tablet_id, tablet, update_meta, false /*keep_files*/,

                                           false /*drop_old*/, profile);

    }

    // During restore process, the tablet is exist and snapshot loader will replace the tablet's rowsets

    // and then reload the tablet, the tablet's path will the same

    if (!force) {

        if (existed_tablet->tablet_path() == tablet->tablet_path()) {

            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(

                    "add the same tablet twice! tablet_id={}, tablet_path={}", tablet_id,

                    tablet->tablet_path());

        }

        if (existed_tablet->data_dir() == tablet->data_dir()) {

            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(

                    "add tablet with same data dir twice! tablet_id={}", tablet_id);

        }

    }

    MonotonicStopWatch watch;

    watch.start();

    // During storage migration, the tablet is moved to another disk, have to check

    // if the new tablet's rowset version is larger than the old one to prevent losting data during

    // migration

    int64_t old_time, new_time;

    int32_t old_version, new_version;

    {

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

        const RowsetSharedPtr old_rowset = existed_tablet->rowset_with_max_version();

        const RowsetSharedPtr new_rowset = tablet->rowset_with_max_version();

        // If new tablet is empty, it is a newly created schema change tablet.

        // the old tablet is dropped before add tablet. it should not exist old tablet

        if (new_rowset == nullptr) {

            // it seems useless to call unlock and return here.

            // it could prevent error when log level is changed in the future.

            return Status::Error<ENGINE_INSERT_EXISTS_TABLE>(

                    "new tablet is empty and old tablet exists. it should not happen. tablet_id={}",

                    tablet_id);

        }

        old_time = old_rowset == nullptr ? -1 : old_rowset->creation_time();

        new_time = new_rowset->creation_time();

        old_version = old_rowset == nullptr ? -1 : old_rowset->end_version();

        new_version = new_rowset->end_version();

    }

    COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "GetExistTabletVersion", "AddTablet"),

                   static_cast<int64_t>(watch.reset()));

    // In restore process, we replace all origin files in tablet dir with

    // the downloaded snapshot files. Then we try to reload tablet header.

    // force == true means we forcibly replace the Tablet in tablet_map

    // with the new one. But if we do so, the files in the tablet dir will be

    // dropped when the origin Tablet deconstruct.

    // So we set keep_files == true to not delete files when the

    // origin Tablet deconstruct.

    // During restore process, snapshot loader

    // replaced the old tablet's rowset with new rowsets, but the tablet path is reused, if drop files

    // here, the new rowset's file will also be dropped, so use keep files here

    bool keep_files = force ? true : false;

    if (force ||

        (new_version > old_version || (new_version == old_version && new_time >= old_time))) {

        // check if new tablet's meta is in store and add new tablet's meta to meta store

        res = _add_tablet_to_map_unlocked(tablet_id, tablet, update_meta, keep_files,

                                          true /*drop_old*/, profile);

    } else {

        tablet->set_tablet_state(TABLET_SHUTDOWN);

        tablet->save_meta();

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "SaveMeta", "AddTablet"),

                       static_cast<int64_t>(watch.reset()));

        {

            std::lock_guard<std::shared_mutex> shutdown_tablets_wrlock(_shutdown_tablets_lock);

            _shutdown_tablets.push_back(tablet);

        }

        res = Status::Error<ENGINE_INSERT_OLD_TABLET>(

                "set tablet to shutdown state. tablet_id={}, tablet_path={}", tablet->tablet_id(),

                tablet->tablet_path());

    }

    LOG(WARNING) << "add duplicated tablet. force=" << force << ", res=" << res

                 << ", tablet_id=" << tablet_id << ", old_version=" << old_version

                 << ", new_version=" << new_version << ", old_time=" << old_time

                 << ", new_time=" << new_time

                 << ", old_tablet_path=" << existed_tablet->tablet_path()

                 << ", new_tablet_path=" << tablet->tablet_path();

    return res;

}

Status TabletManager::_add_tablet_to_map_unlocked(TTabletId tablet_id,

                                                  const TabletSharedPtr& tablet, bool update_meta,

                                                  bool keep_files, bool drop_old,

                                                  RuntimeProfile* profile) {

    // check if new tablet's meta is in store and add new tablet's meta to meta store

    Status res = Status::OK();

    MonotonicStopWatch watch;

    watch.start();

    if (update_meta) {

        // call tablet save meta in order to valid the meta

        tablet->save_meta();

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "SaveMeta", "AddTablet"),

                       static_cast<int64_t>(watch.reset()));

    }

    if (drop_old) {

        // If the new tablet is fresher than the existing one, then replace

        // the existing tablet with the new one.

        // Use default replica_id to ignore whether replica_id is match when drop tablet.

        Status status = _drop_tablet_unlocked(tablet_id, /* replica_id */ 0, keep_files, false);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "DropOldTablet", "AddTablet"),

                       static_cast<int64_t>(watch.reset()));

        RETURN_NOT_OK_STATUS_WITH_WARN(

                status, strings::Substitute("failed to drop old tablet when add new tablet. "

                                            "tablet_id=$0",

                                            tablet_id));

    }

    // Register tablet into DataDir, so that we can manage tablet from

    // the perspective of root path.

    // Example: unregister all tables when a bad disk found.

    tablet->register_tablet_into_dir();

    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);

    tablet_map[tablet_id] = tablet;

    _add_tablet_to_partition(tablet);

    // TODO: remove multiply 2 of tablet meta mem size

    // Because table schema will copy in tablet, there will be double mem cost

    // so here multiply 2

    _tablet_meta_mem_tracker->consume(tablet->tablet_meta()->mem_size() * 2);

    COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RegisterTabletInfo", "AddTablet"),

                   static_cast<int64_t>(watch.reset()));

    VLOG_NOTICE << "add tablet to map successfully."

                << " tablet_id=" << tablet_id;

    return res;

}

bool TabletManager::check_tablet_id_exist(TTabletId tablet_id) {

    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));

    return _check_tablet_id_exist_unlocked(tablet_id);

}

bool TabletManager::_check_tablet_id_exist_unlocked(TTabletId tablet_id) {

    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);

    return tablet_map.find(tablet_id) != tablet_map.end();

}

Status TabletManager::create_tablet(const TCreateTabletReq& request, std::vector<DataDir*> stores,

                                    RuntimeProfile* profile) {

    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);

    DorisMetrics::instance()->create_tablet_requests_total->increment(1);

    int64_t tablet_id = request.tablet_id;

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

              << ", table_id=" << request.table_id << ", partition_id=" << request.partition_id

              << ", replica_id=" << request.replica_id;

    // when we create rollup tablet A(assume on shard-1) from tablet B(assume on shard-2)

    // we need use write lock on shard-1 and then use read lock on shard-2

    // if there have create rollup tablet C(assume on shard-2) from tablet D(assume on shard-1) at the same time, we will meet deadlock

    std::unique_lock two_tablet_lock(_two_tablet_mtx, std::defer_lock);

    bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;

    bool is_schema_change_or_atomic_restore =

            request.__isset.base_tablet_id && request.base_tablet_id > 0;

    bool need_two_lock =

            is_schema_change_or_atomic_restore &&

            ((_tablets_shards_mask & request.base_tablet_id) != (_tablets_shards_mask & tablet_id));

    if (need_two_lock) {

        SCOPED_TIMER(ADD_TIMER(profile, "GetTwoTableLock"));

        two_tablet_lock.lock();

    }

    MonotonicStopWatch shard_lock_watch;

    shard_lock_watch.start();

    std::lock_guard wrlock(_get_tablets_shard_lock(tablet_id));

    shard_lock_watch.stop();

    COUNTER_UPDATE(ADD_TIMER(profile, "GetShardLock"),

                   static_cast<int64_t>(shard_lock_watch.elapsed_time()));

    // Make create_tablet operation to be idempotent:

    // 1. Return true if tablet with same tablet_id and schema_hash exist;

    //           false if tablet with same tablet_id but different schema_hash exist.

    // 2. When this is an alter task, if the tablet(both tablet_id and schema_hash are

    // same) already exist, then just return true(an duplicate request). But if

    // tablet_id exist but with different schema_hash, return an error(report task will

    // eventually trigger its deletion).

    {

        SCOPED_TIMER(ADD_TIMER(profile, "GetTabletUnlocked"));

        if (_get_tablet_unlocked(tablet_id) != nullptr) {

            LOG(INFO) << "success to create tablet. tablet already exist. tablet_id=" << tablet_id;

            return Status::OK();

        }

    }

    TabletSharedPtr base_tablet = nullptr;

    // If the CreateTabletReq has base_tablet_id then it is a alter-tablet request

    if (is_schema_change_or_atomic_restore) {

        // if base_tablet_id's lock diffrent with new_tablet_id, we need lock it.

        if (need_two_lock) {

            SCOPED_TIMER(ADD_TIMER(profile, "GetBaseTablet"));

            base_tablet = get_tablet(request.base_tablet_id);

            two_tablet_lock.unlock();

        } else {

            SCOPED_TIMER(ADD_TIMER(profile, "GetBaseTabletUnlocked"));

            base_tablet = _get_tablet_unlocked(request.base_tablet_id);

        }

        if (base_tablet == nullptr) {

            DorisMetrics::instance()->create_tablet_requests_failed->increment(1);

            return Status::Error<TABLE_CREATE_META_ERROR>(

                    "fail to create tablet(change schema/atomic restore), base tablet does not "

                    "exist. new_tablet_id={}, base_tablet_id={}",

                    tablet_id, request.base_tablet_id);

        }

        // If we are doing schema-change or atomic-restore, we should use the same data dir

        // TODO(lingbin): A litter trick here, the directory should be determined before

        // entering this method

        //

        // ATTN: Since all restored replicas will be saved to HDD, so no storage_medium check here.

        if (in_restore_mode ||

            request.storage_medium == base_tablet->data_dir()->storage_medium()) {

            LOG(INFO) << "create tablet use the base tablet data dir. tablet_id=" << tablet_id

                      << ", base tablet_id=" << request.base_tablet_id

                      << ", data dir=" << base_tablet->data_dir()->path();

            stores.clear();

            stores.push_back(base_tablet->data_dir());

        }

    }

    // set alter type to schema-change. it is useless

    TabletSharedPtr tablet = _internal_create_tablet_unlocked(

            request, is_schema_change_or_atomic_restore, base_tablet.get(), stores, profile);

    if (tablet == nullptr) {

        DorisMetrics::instance()->create_tablet_requests_failed->increment(1);

        return Status::Error<CE_CMD_PARAMS_ERROR>("fail to create tablet. tablet_id={}",

                                                  request.tablet_id);

    }

    LOG(INFO) << "success to create tablet. tablet_id=" << tablet_id;

    return Status::OK();

}

TabletSharedPtr TabletManager::_internal_create_tablet_unlocked(

        const TCreateTabletReq& request, const bool is_schema_change, const Tablet* base_tablet,

        const std::vector<DataDir*>& data_dirs, RuntimeProfile* profile) {

    // If in schema-change state, base_tablet must also be provided.

    // i.e., is_schema_change and base_tablet are either assigned or not assigned

    DCHECK((is_schema_change && base_tablet) || (!is_schema_change && !base_tablet));

    // NOTE: The existence of tablet_id and schema_hash has already been checked,

    // no need check again here.

    const std::string parent_timer_name = "InternalCreateTablet";

    SCOPED_TIMER(ADD_TIMER(profile, parent_timer_name));

    MonotonicStopWatch watch;

    watch.start();

    auto create_meta_timer = ADD_CHILD_TIMER(profile, "CreateMeta", parent_timer_name);

    auto tablet = _create_tablet_meta_and_dir_unlocked(request, is_schema_change, base_tablet,

                                                       data_dirs, profile);

    COUNTER_UPDATE(create_meta_timer, static_cast<int64_t>(watch.reset()));

    if (tablet == nullptr) {

        return nullptr;

    }

    int64_t new_tablet_id = request.tablet_id;

    int32_t new_schema_hash = request.tablet_schema.schema_hash;

    // should remove the tablet's pending_id no matter create-tablet success or not

    DataDir* data_dir = tablet->data_dir();

    // TODO(yiguolei)

    // the following code is very difficult to understand because it mixed alter tablet v2

    // and alter tablet v1 should remove alter tablet v1 code after v0.12

    Status res = Status::OK();

    bool is_tablet_added = false;

    do {

        res = tablet->init();

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "TabletInit", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        if (!res.ok()) {

            LOG(WARNING) << "tablet init failed. tablet:" << tablet->full_name();

            break;

        }

        // Create init version if this is not a restore mode replica and request.version is set

        // bool in_restore_mode = request.__isset.in_restore_mode && request.in_restore_mode;

        // if (!in_restore_mode && request.__isset.version) {

        // create initial rowset before add it to storage engine could omit many locks

        res = tablet->create_initial_rowset(request.version);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "InitRowset", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        if (!res.ok()) {

            LOG(WARNING) << "fail to create initial version for tablet. res=" << res;

            break;

        }

        if (is_schema_change) {

            // if this is a new alter tablet, has to set its state to not ready

            // because schema change handler depends on it to check whether history data

            // convert finished

            tablet->set_tablet_state(TabletState::TABLET_NOTREADY);

        }

        // Add tablet to StorageEngine will make it visible to user

        // Will persist tablet meta

        auto add_tablet_timer = ADD_CHILD_TIMER(profile, "AddTablet", parent_timer_name);

        res = _add_tablet_unlocked(new_tablet_id, tablet, /*update_meta*/ true, false, profile);

        COUNTER_UPDATE(add_tablet_timer, static_cast<int64_t>(watch.reset()));

        if (!res.ok()) {

            LOG(WARNING) << "fail to add tablet to StorageEngine. res=" << res;

            break;

        }

        is_tablet_added = true;

        // TODO(lingbin): The following logic seems useless, can be removed?

        // Because if _add_tablet_unlocked() return OK, we must can get it from map.

        TabletSharedPtr tablet_ptr = _get_tablet_unlocked(new_tablet_id);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "GetTablet", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        if (tablet_ptr == nullptr) {

            res = Status::Error<TABLE_NOT_FOUND>("fail to get tablet. res={}", res);

            break;

        }

    } while (false);

    if (res.ok()) {

        return tablet;

    }

    // something is wrong, we need clear environment

    if (is_tablet_added) {

        Status status = _drop_tablet_unlocked(new_tablet_id, request.replica_id, false, false);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "DropTablet", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        if (!status.ok()) {

            LOG(WARNING) << "fail to drop tablet when create tablet failed. res=" << res;

        }

    } else {

        tablet->delete_all_files();

        TabletMetaManager::remove(data_dir, new_tablet_id, new_schema_hash);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RemoveTabletFiles", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

    }

    return nullptr;

}

static string _gen_tablet_dir(const string& dir, int16_t shard_id, int64_t tablet_id) {

    string path = dir;

    path = path_util::join_path_segments(path, DATA_PREFIX);

    path = path_util::join_path_segments(path, std::to_string(shard_id));

    path = path_util::join_path_segments(path, std::to_string(tablet_id));

    return path;

}

TabletSharedPtr TabletManager::_create_tablet_meta_and_dir_unlocked(

        const TCreateTabletReq& request, const bool is_schema_change, const Tablet* base_tablet,

        const std::vector<DataDir*>& data_dirs, RuntimeProfile* profile) {

    string pending_id = StrCat(TABLET_ID_PREFIX, request.tablet_id);

    // Many attempts are made here in the hope that even if a disk fails, it can still continue.

    std::string parent_timer_name = "CreateMeta";

    MonotonicStopWatch watch;

    watch.start();

    for (auto& data_dir : data_dirs) {

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "RemovePendingIds", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        TabletMetaSharedPtr tablet_meta;

        // if create meta failed, do not need to clean dir, because it is only in memory

        Status res = _create_tablet_meta_unlocked(request, data_dir, is_schema_change, base_tablet,

                                                  &tablet_meta);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "CreateMetaUnlock", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        if (!res.ok()) {

            LOG(WARNING) << "fail to create tablet meta. res=" << res

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

            continue;

        }

        string tablet_dir =

                _gen_tablet_dir(data_dir->path(), tablet_meta->shard_id(), request.tablet_id);

        string schema_hash_dir = path_util::join_path_segments(

                tablet_dir, std::to_string(request.tablet_schema.schema_hash));

        // Because the tablet is removed asynchronously, so that the dir may still exist when BE

        // receive create-tablet request again, For example retried schema-change request

        bool exists = true;

        res = io::global_local_filesystem()->exists(schema_hash_dir, &exists);

        if (!res.ok()) {

            continue;

        }

        if (exists) {

            LOG(WARNING) << "skip this dir because tablet path exist, path=" << schema_hash_dir;

            continue;

        } else {

            Status st = io::global_local_filesystem()->create_directory(schema_hash_dir);

            if (!st.ok()) {

                continue;

            }

        }

        if (tablet_meta->partition_id() <= 0) {

            LOG(WARNING) << "invalid partition id " << tablet_meta->partition_id() << ", tablet "

                         << tablet_meta->tablet_id();

        }

        TabletSharedPtr new_tablet = Tablet::create_tablet_from_meta(tablet_meta, data_dir);

        COUNTER_UPDATE(ADD_CHILD_TIMER(profile, "CreateTabletFromMeta", parent_timer_name),

                       static_cast<int64_t>(watch.reset()));

        DCHECK(new_tablet != nullptr);

        return new_tablet;

    }

    return nullptr;

}

Status TabletManager::drop_tablet(TTabletId tablet_id, TReplicaId replica_id,

                                  bool is_drop_table_or_partition) {

    auto& shard = _get_tablets_shard(tablet_id);

    std::lock_guard wrlock(shard.lock);

    return _drop_tablet_unlocked(tablet_id, replica_id, false, is_drop_table_or_partition);

}

// Drop specified tablet.

Status TabletManager::_drop_tablet_unlocked(TTabletId tablet_id, TReplicaId replica_id,

                                            bool keep_files, bool is_drop_table_or_partition) {

    LOG(INFO) << "begin drop tablet. tablet_id=" << tablet_id << ", replica_id=" << replica_id

              << ", is_drop_table_or_partition=" << is_drop_table_or_partition;

    DorisMetrics::instance()->drop_tablet_requests_total->increment(1);

    RETURN_IF_ERROR(register_transition_tablet(tablet_id, "drop tablet"));

    Defer defer {[&]() { unregister_transition_tablet(tablet_id, "drop tablet"); }};

    // Fetch tablet which need to be dropped

    TabletSharedPtr to_drop_tablet = _get_tablet_unlocked(tablet_id);

    if (to_drop_tablet == nullptr) {

        LOG(WARNING) << "fail to drop tablet because it does not exist. "

                     << "tablet_id=" << tablet_id;

        return Status::OK();

    }

    // We should compare replica id to avoid dropping new cloned tablet.

    // Iff request replica id is 0, FE may be an older release, then we drop this tablet as before.

    if (to_drop_tablet->replica_id() != replica_id && replica_id != 0) {

        return Status::Aborted("replica_id not match({} vs {})", to_drop_tablet->replica_id(),

                               replica_id);

    }

    _remove_tablet_from_partition(to_drop_tablet);

    tablet_map_t& tablet_map = _get_tablet_map(tablet_id);

    tablet_map.erase(tablet_id);

    to_drop_tablet->clear_cache();

    if (!keep_files) {

        // drop tablet will update tablet meta, should lock

        std::lock_guard<std::shared_mutex> wrlock(to_drop_tablet->get_header_lock());

        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);

        LOG(INFO) << "set tablet to shutdown state and remove it from memory. "

                  << "tablet_id=" << tablet_id << ", tablet_path=" << to_drop_tablet->tablet_path();

        // NOTE: has to update tablet here, but must not update tablet meta directly.

        // because other thread may hold the tablet object, they may save meta too.

        // If update meta directly here, other thread may override the meta

        // and the tablet will be loaded at restart time.

        // To avoid this exception, we first set the state of the tablet to `SHUTDOWN`.

        to_drop_tablet->set_tablet_state(TABLET_SHUTDOWN);

        // We must record unused remote rowsets path info to OlapMeta before tablet state is marked as TABLET_SHUTDOWN in OlapMeta,

        // otherwise if BE shutdown after saving tablet state, these remote rowsets path info will lost.

        if (is_drop_table_or_partition) {

            RETURN_IF_ERROR(to_drop_tablet->remove_all_remote_rowsets());

        }

        to_drop_tablet->save_meta();

        {

            std::lock_guard<std::shared_mutex> wrdlock(_shutdown_tablets_lock);

            _shutdown_tablets.push_back(to_drop_tablet);

        }

    }

    to_drop_tablet->deregister_tablet_from_dir();

    _tablet_meta_mem_tracker->release(to_drop_tablet->tablet_meta()->mem_size() * 2);

    return Status::OK();

}

TabletSharedPtr TabletManager::get_tablet(TTabletId tablet_id, bool include_deleted, string* err) {

    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));

    return _get_tablet_unlocked(tablet_id, include_deleted, err);

}

std::vector<TabletSharedPtr> TabletManager::get_all_tablet(std::function<bool(Tablet*)>&& filter) {

    std::vector<TabletSharedPtr> res;

    for_each_tablet([&](const TabletSharedPtr& tablet) { res.emplace_back(tablet); },

                    std::move(filter));

    return res;

}

void TabletManager::for_each_tablet(std::function<void(const TabletSharedPtr&)>&& handler,

                                    std::function<bool(Tablet*)>&& filter) {

    std::vector<TabletSharedPtr> tablets;

    for (const auto& tablets_shard : _tablets_shards) {

        tablets.clear();

        {

            std::shared_lock rdlock(tablets_shard.lock);

            for (const auto& [id, tablet] : tablets_shard.tablet_map) {

                if (filter(tablet.get())) {

                    tablets.emplace_back(tablet);

                }

            }

        }

        for (const auto& tablet : tablets) {

            handler(tablet);

        }

    }

}

std::pair<TabletSharedPtr, Status> TabletManager::get_tablet_and_status(TTabletId tablet_id,

                                                                        bool include_deleted) {

    std::string err;

    auto tablet = get_tablet(tablet_id, include_deleted, &err);

    if (tablet == nullptr) {

        return {tablet,

                Status::InternalError("failed to get tablet: {}, reason: {}", tablet_id, err)};

    }

    return {tablet, Status::OK()};

}

TabletSharedPtr TabletManager::_get_tablet_unlocked(TTabletId tablet_id, bool include_deleted,

                                                    string* err) {

    TabletSharedPtr tablet;

    tablet = _get_tablet_unlocked(tablet_id);

    if (tablet == nullptr && include_deleted) {

        std::shared_lock rdlock(_shutdown_tablets_lock);

        for (auto& deleted_tablet : _shutdown_tablets) {

            CHECK(deleted_tablet != nullptr) << "deleted tablet is nullptr";

            if (deleted_tablet->tablet_id() == tablet_id) {

                tablet = deleted_tablet;

                break;

            }

        }

    }

    if (tablet == nullptr) {

        if (err != nullptr) {

            *err = "tablet does not exist. " + BackendOptions::get_localhost();

        }

        return nullptr;

    }

    if (!tablet->is_used()) {

        LOG(WARNING) << "tablet cannot be used. tablet=" << tablet_id;

        if (err != nullptr) {

            *err = "tablet cannot be used. " + BackendOptions::get_localhost();

        }

        return nullptr;

    }

    return tablet;

}

TabletSharedPtr TabletManager::get_tablet(TTabletId tablet_id, TabletUid tablet_uid,

                                          bool include_deleted, string* err) {

    std::shared_lock rdlock(_get_tablets_shard_lock(tablet_id));

    TabletSharedPtr tablet = _get_tablet_unlocked(tablet_id, include_deleted, err);

    if (tablet != nullptr && tablet->tablet_uid() == tablet_uid) {

        return tablet;

    }

    return nullptr;

}

uint64_t TabletManager::get_rowset_nums() {

    uint64_t rowset_nums = 0;

    for_each_tablet([&](const TabletSharedPtr& tablet) { rowset_nums += tablet->version_count(); },

                    filter_all_tablets);

    return rowset_nums;

}

uint64_t TabletManager::get_segment_nums() {

    uint64_t segment_nums = 0;

    for_each_tablet([&](const TabletSharedPtr& tablet) { segment_nums += tablet->segment_count(); },

                    filter_all_tablets);

    return segment_nums;

}

bool TabletManager::get_tablet_id_and_schema_hash_from_path(const string& path,

                                                            TTabletId* tablet_id,

                                                            TSchemaHash* schema_hash) {

    // the path like: /data/14/10080/964828783/

    static re2::RE2 normal_re("/data/\\d+/(\\d+)/(\\d+)($|/)");

    // match tablet schema hash data path, for example, the path is /data/1/16791/29998

    // 1 is shard id , 16791 is tablet id, 29998 is schema hash

    if (RE2::PartialMatch(path, normal_re, tablet_id, schema_hash)) {

        return true;

    }

    // If we can't match normal path pattern, this may be a path which is a empty tablet

    // directory. Use this pattern to match empty tablet directory. In this case schema_hash

    // will be set to zero.

    static re2::RE2 empty_tablet_re("/data/\\d+/(\\d+)($|/$)");

    if (!RE2::PartialMatch(path, empty_tablet_re, tablet_id)) {

        return false;

    }

    *schema_hash = 0;

    return true;

}

bool TabletManager::get_rowset_id_from_path(const string& path, RowsetId* rowset_id) {

    // the path like: /data/14/10080/964828783/02000000000000969144d8725cb62765f9af6cd3125d5a91_0.dat

    static re2::RE2 re("/data/\\d+/\\d+/\\d+/([A-Fa-f0-9]+)_.*");

    string id_str;

    bool ret = RE2::PartialMatch(path, re, &id_str);

    if (ret) {

        rowset_id->init(id_str);

        return true;

    }

    return false;

}

void TabletManager::get_tablet_stat(TTabletStatResult* result) {

    std::shared_ptr<std::vector<TTabletStat>> local_cache;

    {

        std::lock_guard<std::mutex> guard(_tablet_stat_cache_mutex);

        local_cache = _tablet_stat_list_cache;

    }

    result->__set_tablet_stat_list(*local_cache);

}

TabletSharedPtr TabletManager::find_best_tablet_to_compaction(

        CompactionType compaction_type, DataDir* data_dir,

        const std::unordered_set<TTabletId>& tablet_submitted_compaction, uint32_t* score,

        const std::unordered_map<std::string_view, std::shared_ptr<CumulativeCompactionPolicy>>&

                all_cumulative_compaction_policies) {

    int64_t now_ms = UnixMillis();

    const string& compaction_type_str =

            compaction_type == CompactionType::BASE_COMPACTION ? "base" : "cumulative";

    uint32_t highest_score = 0;

    uint32_t compaction_score = 0;

    TabletSharedPtr best_tablet;

    auto handler = [&](const TabletSharedPtr& tablet_ptr) {

        if (tablet_ptr->tablet_meta()->tablet_schema()->disable_auto_compaction()) {

            LOG_EVERY_N(INFO, 500) << "Tablet " << tablet_ptr->tablet_id()

                                   << " will be ignored by automatic compaction tasks since it's "

                                   << "set to disabled automatic compaction.";

            return;

        }

        if (config::enable_skip_tablet_compaction &&

            tablet_ptr->should_skip_compaction(compaction_type, UnixSeconds())) {

            return;

        }

        if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) {

            return;

        }

        auto search = tablet_submitted_compaction.find(tablet_ptr->tablet_id());

        if (search != tablet_submitted_compaction.end()) {

            return;

        }

        int64_t last_failure_ms = tablet_ptr->last_cumu_compaction_failure_time();

        if (compaction_type == CompactionType::BASE_COMPACTION) {

            last_failure_ms = tablet_ptr->last_base_compaction_failure_time();

        }

        if (now_ms - last_failure_ms <= 5000) {

            VLOG_DEBUG << "Too often to check compaction, skip it. "

                       << "compaction_type=" << compaction_type_str

                       << ", last_failure_time_ms=" << last_failure_ms

                       << ", tablet_id=" << tablet_ptr->tablet_id();

            return;

        }

        if (compaction_type == CompactionType::BASE_COMPACTION) {

            std::unique_lock<std::mutex> lock(tablet_ptr->get_base_compaction_lock(),

                                              std::try_to_lock);

            if (!lock.owns_lock()) {

                LOG(INFO) << "can not get base lock: " << tablet_ptr->tablet_id();

                return;

            }

        } else {

            std::unique_lock<std::mutex> lock(tablet_ptr->get_cumulative_compaction_lock(),

                                              std::try_to_lock);

            if (!lock.owns_lock()) {

                LOG(INFO) << "can not get cumu lock: " << tablet_ptr->tablet_id();

                return;

            }

        }

        auto cumulative_compaction_policy = all_cumulative_compaction_policies.at(

                tablet_ptr->tablet_meta()->compaction_policy());

        uint32_t current_compaction_score =

                tablet_ptr->calc_compaction_score(compaction_type, cumulative_compaction_policy);

        if (current_compaction_score < 5) {

            tablet_ptr->set_skip_compaction(true, compaction_type, UnixSeconds());

        }

        if (current_compaction_score > highest_score) {

            highest_score = current_compaction_score;

            compaction_score = current_compaction_score;

            best_tablet = tablet_ptr;

        }

    };

    for_each_tablet(handler, filter_all_tablets);

    if (best_tablet != nullptr) {

        VLOG_CRITICAL << "Found the best tablet for compaction. "

                      << "compaction_type=" << compaction_type_str

                      << ", tablet_id=" << best_tablet->tablet_id() << ", path=" << data_dir->path()

                      << ", compaction_score=" << compaction_score

                      << ", highest_score=" << highest_score;

        *score = compaction_score;

    }

    return best_tablet;

}

Status TabletManager::load_tablet_from_meta(DataDir* data_dir, TTabletId tablet_id,

                                            TSchemaHash schema_hash, const string& meta_binary,

                                            bool update_meta, bool force, bool restore,

                                            bool check_path) {

    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);

    TabletMetaSharedPtr tablet_meta(new TabletMeta());

    Status status = tablet_meta->deserialize(meta_binary);

    if (!status.ok()) {

        return Status::Error<HEADER_PB_PARSE_FAILED>(

                "fail to load tablet because can not parse meta_binary string. tablet_id={}, "

                "schema_hash={}, path={}, status={}",

                tablet_id, schema_hash, data_dir->path(), status);

    }

    tablet_meta->init_rs_metas_fs(data_dir->fs());

    // check if tablet meta is valid

    if (tablet_meta->tablet_id() != tablet_id || tablet_meta->schema_hash() != schema_hash) {

        return Status::Error<HEADER_PB_PARSE_FAILED>(

                "fail to load tablet because meet invalid tablet meta. trying to load "

                "tablet(tablet_id={}, schema_hash={}), but meet tablet={}, path={}",

                tablet_id, schema_hash, tablet_meta->full_name(), data_dir->path());

    }

    if (tablet_meta->tablet_uid().hi == 0 && tablet_meta->tablet_uid().lo == 0) {

        return Status::Error<HEADER_PB_PARSE_FAILED>(

                "fail to load tablet because its uid == 0. tablet={}, path={}",

                tablet_meta->full_name(), data_dir->path());

    }

    if (restore) {

        // we're restoring tablet from trash, tablet state should be changed from shutdown back to running

        tablet_meta->set_tablet_state(TABLET_RUNNING);

    }

    if (tablet_meta->partition_id() == 0) {

        LOG(WARNING) << "tablet=" << tablet_id << " load from meta but partition id eq 0";

    }

    TabletSharedPtr tablet = Tablet::create_tablet_from_meta(tablet_meta, data_dir);

    if (tablet == nullptr) {

        return Status::Error<TABLE_CREATE_FROM_HEADER_ERROR>(

                "fail to load tablet. tablet_id={}, schema_hash={}", tablet_id, schema_hash);

    }

    // NOTE: method load_tablet_from_meta could be called by two cases as below

    // case 1: BE start;

    // case 2: Clone Task/Restore

    // For case 1 doesn't need path check because BE is just starting and not ready,

    // just check tablet meta status to judge whether tablet is delete is enough.

    // For case 2, If a tablet has just been copied to local BE,

    // it may be cleared by gc-thread(see perform_path_gc_by_tablet) because the tablet meta may not be loaded to memory.

    // So clone task should check path and then failed and retry in this case.

    if (check_path) {

        bool exists = true;

        RETURN_IF_ERROR(io::global_local_filesystem()->exists(tablet->tablet_path(), &exists));

        if (!exists) {

            return Status::Error<TABLE_ALREADY_DELETED_ERROR>(

                    "tablet path not exists, create tablet failed, path={}", tablet->tablet_path());

        }

    }

    if (tablet_meta->tablet_state() == TABLET_SHUTDOWN) {

        {

            std::lock_guard<std::shared_mutex> shutdown_tablets_wrlock(_shutdown_tablets_lock);

            _shutdown_tablets.push_back(tablet);

        }

        return Status::Error<TABLE_ALREADY_DELETED_ERROR>(

                "fail to load tablet because it is to be deleted. tablet_id={}, schema_hash={}, "

                "path={}",

                tablet_id, schema_hash, data_dir->path());

    }

    // NOTE: We do not check tablet's initial version here, because if BE restarts when

    // one tablet is doing schema-change, we may meet empty tablet.

    if (tablet->max_version().first == -1 && tablet->tablet_state() == TABLET_RUNNING) {

        // tablet state is invalid, drop tablet

        return Status::Error<TABLE_INDEX_VALIDATE_ERROR>(

                "fail to load tablet. it is in running state but without delta. tablet={}, path={}",

                tablet->full_name(), data_dir->path());

    }

    RETURN_NOT_OK_STATUS_WITH_WARN(

            tablet->init(),

            strings::Substitute("tablet init failed. tablet=$0", tablet->full_name()));

    RuntimeProfile profile("CreateTablet");

    std::lock_guard<std::shared_mutex> wrlock(_get_tablets_shard_lock(tablet_id));

    RETURN_NOT_OK_STATUS_WITH_WARN(

            _add_tablet_unlocked(tablet_id, tablet, update_meta, force, &profile),

            strings::Substitute("fail to add tablet. tablet=$0", tablet->full_name()));

    return Status::OK();

}

Status TabletManager::load_tablet_from_dir(DataDir* store, TTabletId tablet_id,

                                           SchemaHash schema_hash, const string& schema_hash_path,

                                           bool force, bool restore) {

    SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);

    LOG(INFO) << "begin to load tablet from dir. "

              << " tablet_id=" << tablet_id << " schema_hash=" << schema_hash

              << " path = " << schema_hash_path << " force = " << force << " restore = " << restore;

    // not add lock here, because load_tablet_from_meta already add lock

    std::string header_path = TabletMeta::construct_header_file_path(schema_hash_path, tablet_id);

    // should change shard id before load tablet

    std::string shard_path =

            path_util::dir_name(path_util::dir_name(path_util::dir_name(header_path)));

    std::string shard_str = shard_path.substr(shard_path.find_last_of('/') + 1);

    int32_t shard = stol(shard_str);

    bool exists = false;

    RETURN_IF_ERROR(io::global_local_filesystem()->exists(header_path, &exists));

    if (!exists) {

        return Status::Error<FILE_NOT_EXIST>("fail to find header file. [header_path={}]",

                                             header_path);

    }

    TabletMetaSharedPtr tablet_meta(new TabletMeta());

    if (!tablet_meta->create_from_file(header_path).ok()) {

        return Status::Error<ENGINE_LOAD_INDEX_TABLE_ERROR>(

                "fail to load tablet_meta. file_path={}", header_path);

    }

    TabletUid tablet_uid = TabletUid::gen_uid();

    // remove rowset binlog metas

    auto binlog_metas_file = fmt::format("{}/rowset_binlog_metas.pb", schema_hash_path);

    bool binlog_metas_file_exists = false;

    auto file_exists_status =

            io::global_local_filesystem()->exists(binlog_metas_file, &binlog_metas_file_exists);

    if (!file_exists_status.ok()) {

        return file_exists_status;

    }

    bool contain_binlog = false;

    RowsetBinlogMetasPB rowset_binlog_metas_pb;

    if (binlog_metas_file_exists) {

        auto binlog_meta_filesize = std::filesystem::file_size(binlog_metas_file);

        if (binlog_meta_filesize > 0) {

            contain_binlog = true;

            RETURN_IF_ERROR(read_pb(binlog_metas_file, &rowset_binlog_metas_pb));

            VLOG_DEBUG << "load rowset binlog metas from file. file_path=" << binlog_metas_file;

        }

        RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(binlog_metas_file));

    }

    if (contain_binlog) {

        auto binlog_dir = fmt::format("{}/_binlog", schema_hash_path);

        RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(binlog_dir));

        std::vector<io::FileInfo> files;

        RETURN_IF_ERROR(

                io::global_local_filesystem()->list(schema_hash_path, true, &files, &exists));

        for (auto& file : files) {

            auto& filename = file.file_name;

            std::string new_suffix;

            std::string old_suffix;

            if (filename.ends_with(".binlog")) {

                old_suffix = ".binlog";

                new_suffix = ".dat";

            } else if (filename.ends_with(".binlog-index")) {

                old_suffix = ".binlog-index";

                new_suffix = ".idx";

            } else {

                continue;

            }

            std::string new_filename = filename;

            new_filename.replace(filename.size() - old_suffix.size(), old_suffix.size(),