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

#include <fmt/core.h>

#include <fmt/format.h>

#include <gen_cpp/FrontendService_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <atomic>

#include <cstdio>

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

#include <chrono> // IWYU pragma: keep

#include <cstddef>

#include <filesystem>

#include <memory>

#include <new>

#include <roaring/roaring.hh>

#include <set>

#include <sstream>

#include <string>

#include <thread>

#include <utility>

#include "common/cast_set.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_writer.h"

#include "io/fs/local_file_system.h"

#include "io/fs/path.h"

#include "service/backend_options.h"

#include "storage/delete/delete_handler.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/olap_meta.h"

#include "storage/rowset/beta_rowset.h"

#include "storage/rowset/pending_rowset_helper.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_id_generator.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/rowset/rowset_meta_manager.h"

#include "storage/storage_engine.h"

#include "storage/storage_policy.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_manager.h"

#include "storage/tablet/tablet_meta_manager.h"

#include "storage/txn/txn_manager.h"

#include "storage/utils.h" // for check_dir_existed

#include "util/string_util.h"

#include "util/uid_util.h"

namespace doris {

#include "common/compile_check_begin.h"

using namespace ErrorCode;

namespace {

Status read_cluster_id(const std::string& cluster_id_path, int32_t* cluster_id) {

    bool exists = false;

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

    *cluster_id = -1;

    if (exists) {

        io::FileReaderSPtr reader;

        RETURN_IF_ERROR(io::global_local_filesystem()->open_file(cluster_id_path, &reader));

        size_t fsize = reader->size();

        if (fsize > 0) {

            std::string content;

            content.resize(fsize, '\0');

            size_t bytes_read = 0;

            RETURN_IF_ERROR(reader->read_at(0, {content.data(), fsize}, &bytes_read));

            DCHECK_EQ(fsize, bytes_read);

            *cluster_id = std::stoi(content);

        }

    }

    return Status::OK();

}

Status _write_cluster_id_to_path(const std::string& path, int32_t cluster_id) {

    bool exists = false;

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

    if (!exists) {

        io::FileWriterPtr file_writer;

        RETURN_IF_ERROR(io::global_local_filesystem()->create_file(path, &file_writer));

        RETURN_IF_ERROR(file_writer->append(std::to_string(cluster_id)));

        RETURN_IF_ERROR(file_writer->close());

    }

    return Status::OK();

}

} // namespace

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_total_capacity, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_avail_capacity, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_local_used_capacity, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_remote_used_capacity, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_trash_used_capacity, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_state, MetricUnit::BYTES);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_compaction_score, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(disks_compaction_num, MetricUnit::NOUNIT);

DataDir::DataDir(StorageEngine& engine, const std::string& path, int64_t capacity_bytes,

                 TStorageMedium::type storage_medium)

        : _engine(engine),

          _path(path),

          _available_bytes(0),

          _disk_capacity_bytes(0),

          _trash_used_bytes(0),

          _storage_medium(storage_medium),

          _is_used(false),

          _cluster_id(-1),

          _to_be_deleted(false) {

    _data_dir_metric_entity = DorisMetrics::instance()->metric_registry()->register_entity(

            std::string("data_dir.") + path, {{"path", path}});

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_total_capacity);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_avail_capacity);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_local_used_capacity);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_remote_used_capacity);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_trash_used_capacity);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_state);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_compaction_score);

    INT_GAUGE_METRIC_REGISTER(_data_dir_metric_entity, disks_compaction_num);

}

DataDir::~DataDir() {

    DorisMetrics::instance()->metric_registry()->deregister_entity(_data_dir_metric_entity);

    delete _meta;

}

Status DataDir::init(bool init_meta) {

    bool exists = false;

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

    if (!exists) {

        RETURN_NOT_OK_STATUS_WITH_WARN(Status::IOError("opendir failed, path={}", _path),

                                       "check file exist failed");

    }

    RETURN_NOT_OK_STATUS_WITH_WARN(update_capacity(), "update_capacity failed");

    RETURN_NOT_OK_STATUS_WITH_WARN(_init_cluster_id(), "_init_cluster_id failed");

    RETURN_NOT_OK_STATUS_WITH_WARN(_init_capacity_and_create_shards(),

                                   "_init_capacity_and_create_shards failed");

    if (init_meta) {

        RETURN_NOT_OK_STATUS_WITH_WARN(_init_meta(), "_init_meta failed");

    }

    _is_used = true;

    return Status::OK();

}

void DataDir::stop_bg_worker() {

    _stop_bg_worker = true;

}

Status DataDir::_init_cluster_id() {

    auto cluster_id_path = fmt::format("{}/{}", _path, CLUSTER_ID_PREFIX);

    RETURN_IF_ERROR(read_cluster_id(cluster_id_path, &_cluster_id));

    if (_cluster_id == -1) {

        _cluster_id_incomplete = true;

    }

    return Status::OK();

}

Status DataDir::_init_capacity_and_create_shards() {

    RETURN_IF_ERROR(io::global_local_filesystem()->get_space_info(_path, &_disk_capacity_bytes,

                                                                  &_available_bytes));

    auto data_path = fmt::format("{}/{}", _path, DATA_PREFIX);

    bool exists = false;

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

    if (!exists) {

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

    }

    for (int i = 0; i < MAX_SHARD_NUM; ++i) {

        auto shard_path = fmt::format("{}/{}", data_path, i);

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

        if (!exists) {

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

        }

    }

    return Status::OK();

}

Status DataDir::_init_meta() {

    // init path hash

    _path_hash = hash_of_path(BackendOptions::get_localhost(), _path);

    LOG(INFO) << "path: " << _path << ", hash: " << _path_hash;

    // init meta

    _meta = new (std::nothrow) OlapMeta(_path);

    if (_meta == nullptr) {

        RETURN_NOT_OK_STATUS_WITH_WARN(

                Status::MemoryAllocFailed("allocate memory for OlapMeta failed"),

                "new OlapMeta failed");

    }

    Status res = _meta->init();

    if (!res.ok()) {

        RETURN_NOT_OK_STATUS_WITH_WARN(Status::IOError("open rocksdb failed, path={}", _path),

                                       "init OlapMeta failed");

    }

    return Status::OK();

}

Status DataDir::set_cluster_id(int32_t cluster_id) {

    if (_cluster_id != -1 && _cluster_id != cluster_id) {

        LOG(ERROR) << "going to set cluster id to already assigned store, cluster_id="

                   << _cluster_id << ", new_cluster_id=" << cluster_id;

        return Status::InternalError("going to set cluster id to already assigned store");

    }

    if (!_cluster_id_incomplete) {

        return Status::OK();

    }

    auto cluster_id_path = fmt::format("{}/{}", _path, CLUSTER_ID_PREFIX);

    return _write_cluster_id_to_path(cluster_id_path, cluster_id);

}

void DataDir::health_check() {

    // check disk

    if (_is_used) {

        Status res = _read_and_write_test_file();

        if (!res && res.is<IO_ERROR>()) {

            LOG(WARNING) << "store read/write test file occur IO Error. path=" << _path

                         << ", err: " << res;

            _engine.add_broken_path(_path);

            _is_used = !res.is<IO_ERROR>();

        }

    }

    disks_state->set_value(_is_used ? 1 : 0);

}

Status DataDir::_read_and_write_test_file() {

    auto test_file = fmt::format("{}/{}", _path, kTestFilePath);

    return read_write_test_file(test_file);

}

void DataDir::register_tablet(Tablet* tablet) {

    TabletInfo tablet_info(tablet->tablet_id(), tablet->tablet_uid());

    std::lock_guard<std::mutex> l(_mutex);

    _tablet_set.emplace(std::move(tablet_info));

}

void DataDir::deregister_tablet(Tablet* tablet) {

    TabletInfo tablet_info(tablet->tablet_id(), tablet->tablet_uid());

    std::lock_guard<std::mutex> l(_mutex);

    _tablet_set.erase(tablet_info);

}

void DataDir::clear_tablets(std::vector<TabletInfo>* tablet_infos) {

    std::lock_guard<std::mutex> l(_mutex);

    tablet_infos->insert(tablet_infos->end(), _tablet_set.begin(), _tablet_set.end());

    _tablet_set.clear();

}

std::string DataDir::get_absolute_shard_path(int64_t shard_id) {

    return fmt::format("{}/{}/{}", _path, DATA_PREFIX, shard_id);

}

std::string DataDir::get_absolute_tablet_path(int64_t shard_id, int64_t tablet_id,

                                              int32_t schema_hash) {

    return fmt::format("{}/{}/{}", get_absolute_shard_path(shard_id), tablet_id, schema_hash);

}

void DataDir::find_tablet_in_trash(int64_t tablet_id, std::vector<std::string>* paths) {

    // path: /root_path/trash/time_label/tablet_id/schema_hash

    auto trash_path = fmt::format("{}/{}", _path, TRASH_PREFIX);

    bool exists = true;

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

    Status st = io::global_local_filesystem()->list(trash_path, false, &sub_dirs, &exists);

    if (!st) {

        return;

    }

    for (auto& sub_dir : sub_dirs) {

        // sub dir is time_label

        if (sub_dir.is_file) {

            continue;

        }

        auto sub_path = fmt::format("{}/{}", trash_path, sub_dir.file_name);

        auto tablet_path = fmt::format("{}/{}", sub_path, tablet_id);

        st = io::global_local_filesystem()->exists(tablet_path, &exists);

        if (st && exists) {

            paths->emplace_back(std::move(tablet_path));

        }

    }

}

std::string DataDir::get_root_path_from_schema_hash_path_in_trash(

        const std::string& schema_hash_dir_in_trash) {

    return io::Path(schema_hash_dir_in_trash)

            .parent_path()

            .parent_path()

            .parent_path()

            .parent_path()

            .string();

}

Status DataDir::_check_incompatible_old_format_tablet() {

    auto check_incompatible_old_func = [](int64_t tablet_id, int32_t schema_hash,

                                          std::string_view value) -> bool {

        // if strict check incompatible old format, then log fatal

        if (config::storage_strict_check_incompatible_old_format) {

            throw Exception(Status::FatalError(

                    "There are incompatible old format metas, current version does not support and "

                    "it may lead to data missing!!! tablet_id = {} schema_hash = {}",

                    tablet_id, schema_hash));

        } else {

            LOG(WARNING)

                    << "There are incompatible old format metas, current version does not support "

                    << "and it may lead to data missing!!! "

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

        }

        return false;

    };

    // seek old header prefix. when check_incompatible_old_func is called, it has old format in olap_meta

    Status check_incompatible_old_status = TabletMetaManager::traverse_headers(

            _meta, check_incompatible_old_func, OLD_HEADER_PREFIX);

    if (!check_incompatible_old_status) {

        LOG(WARNING) << "check incompatible old format meta fails, it may lead to data missing!!! "

                     << _path;

    } else {

        LOG(INFO) << "successfully check incompatible old format meta " << _path;

    }

    return check_incompatible_old_status;

}

// TODO(ygl): deal with rowsets and tablets when load failed

Status DataDir::load() {

    LOG(INFO) << "start to load tablets from " << _path;

    // load rowset meta from meta env and create rowset

    // COMMITTED: add to txn manager

    // VISIBLE: add to tablet

    // if one rowset load failed, then the total data dir will not be loaded

    // necessarily check incompatible old format. when there are old metas, it may load to data missing

    RETURN_IF_ERROR(_check_incompatible_old_format_tablet());

    std::vector<RowsetMetaSharedPtr> dir_rowset_metas;

    LOG(INFO) << "begin loading rowset from meta";

    auto load_rowset_func = [&dir_rowset_metas, this](TabletUid tablet_uid, RowsetId rowset_id,

                                                      std::string_view meta_str) -> bool {

        RowsetMetaSharedPtr rowset_meta(new RowsetMeta());

        bool parsed = rowset_meta->init(meta_str);

        if (!parsed) {

            LOG(WARNING) << "parse rowset meta string failed for rowset_id:" << rowset_id;

            // return false will break meta iterator, return true to skip this error

            return true;

        }

        if (rowset_meta->has_delete_predicate()) {

            // copy the delete sub pred v1 to check then

            auto orig_delete_sub_pred = rowset_meta->delete_predicate().sub_predicates();

            auto* delete_pred = rowset_meta->mutable_delete_pred_pb();

            if ((!delete_pred->sub_predicates().empty() &&

                 delete_pred->sub_predicates_v2().empty()) ||

                (!delete_pred->in_predicates().empty() &&

                 delete_pred->in_predicates()[0].has_column_unique_id())) {

                // convert pred and write only when delete sub pred v2 is not set or there is in list pred to be set column uid

                RETURN_IF_ERROR(DeleteHandler::convert_to_sub_pred_v2(

                        delete_pred, rowset_meta->tablet_schema()));

                LOG(INFO) << fmt::format(

                        "convert rowset with old delete pred: rowset_id={}, tablet_id={}",

                        rowset_id.to_string(), tablet_uid.to_string());

                CHECK_EQ(orig_delete_sub_pred.size(), delete_pred->sub_predicates().size())

                        << "inconsistent sub predicate v1 after conversion";

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

                    CHECK_STREQ(orig_delete_sub_pred.Get(i).c_str(),

                                delete_pred->sub_predicates().Get(i).c_str())

                            << "inconsistent sub predicate v1 after conversion";

                }

                std::string result;

                rowset_meta->serialize(&result);

                std::string key =

                        ROWSET_PREFIX + tablet_uid.to_string() + "_" + rowset_id.to_string();

                RETURN_IF_ERROR(_meta->put(META_COLUMN_FAMILY_INDEX, key, result));

            }

        }

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

            LOG(WARNING) << "rs tablet=" << rowset_meta->tablet_id() << " rowset_id=" << rowset_id

                         << " load from meta but partition id eq 0";

        }

        dir_rowset_metas.push_back(rowset_meta);

        return true;

    };

    MonotonicStopWatch rs_timer;

    rs_timer.start();

    Status load_rowset_status = RowsetMetaManager::traverse_rowset_metas(_meta, load_rowset_func);

    rs_timer.stop();

    if (!load_rowset_status) {

        LOG(WARNING) << "errors when load rowset meta from meta env, skip this data dir:" << _path;

    } else {

        LOG(INFO) << "load rowset from meta finished, cost: "

                  << rs_timer.elapsed_time_milliseconds() << " ms, data dir: " << _path;

    }

    // load tablet

    // create tablet from tablet meta and add it to tablet mgr

    LOG(INFO) << "begin loading tablet from meta";

    std::set<int64_t> tablet_ids;

    std::set<int64_t> failed_tablet_ids;

    auto load_tablet_func = [this, &tablet_ids, &failed_tablet_ids](

                                    int64_t tablet_id, int32_t schema_hash,

                                    std::string_view value) -> bool {

        Status status = _engine.tablet_manager()->load_tablet_from_meta(

                this, tablet_id, schema_hash, value, false, false, false, false);

        if (!status.ok() && !status.is<TABLE_ALREADY_DELETED_ERROR>() &&

            !status.is<ENGINE_INSERT_OLD_TABLET>()) {

            // load_tablet_from_meta() may return Status::Error<TABLE_ALREADY_DELETED_ERROR>()

            // which means the tablet status is DELETED

            // This may happen when the tablet was just deleted before the BE restarted,

            // but it has not been cleared from rocksdb. At this time, restarting the BE

            // will read the tablet in the DELETE state from rocksdb. These tablets have been

            // added to the garbage collection queue and will be automatically deleted afterwards.

            // Therefore, we believe that this situation is not a failure.

            // Besides, load_tablet_from_meta() may return Status::Error<ENGINE_INSERT_OLD_TABLET>()

            // when BE is restarting and the older tablet have been added to the

            // garbage collection queue but not deleted yet.

            // In this case, since the data_dirs are parallel loaded, a later loaded tablet

            // may be older than previously loaded one, which should not be acknowledged as a

            // failure.

            LOG(WARNING) << "load tablet from header failed. status:" << status

                         << ", tablet=" << tablet_id << "." << schema_hash;

            failed_tablet_ids.insert(tablet_id);

        } else {

            tablet_ids.insert(tablet_id);

        }

        return true;

    };

    MonotonicStopWatch tablet_timer;

    tablet_timer.start();

    Status load_tablet_status = TabletMetaManager::traverse_headers(_meta, load_tablet_func);

    tablet_timer.stop();

    if (!failed_tablet_ids.empty()) {

        LOG(WARNING) << "load tablets from header failed"

                     << ", loaded tablet: " << tablet_ids.size()

                     << ", error tablet: " << failed_tablet_ids.size() << ", path: " << _path;

        if (!config::ignore_load_tablet_failure) {

            throw Exception(Status::FatalError(

                    "load tablets encounter failure. stop BE process. path: {}", _path));

        }

    }

    if (!load_tablet_status) {

        LOG(WARNING) << "there is failure when loading tablet headers"

                     << ", loaded tablet: " << tablet_ids.size()

                     << ", error tablet: " << failed_tablet_ids.size() << ", path: " << _path;

    } else {

        LOG(INFO) << "load tablet from meta finished"

                  << ", loaded tablet: " << tablet_ids.size()

                  << ", error tablet: " << failed_tablet_ids.size()

                  << ", cost: " << tablet_timer.elapsed_time_milliseconds()

                  << " ms, path: " << _path;

    }

    for (int64_t tablet_id : tablet_ids) {

        TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(tablet_id);

        if (tablet && tablet->set_tablet_schema_into_rowset_meta()) {

            RETURN_IF_ERROR(TabletMetaManager::save(this, tablet->tablet_id(),

                                                    tablet->schema_hash(), tablet->tablet_meta()));

        }

    }

    auto load_pending_publish_info_func =

            [&engine = _engine](int64_t tablet_id, int64_t publish_version, std::string_view info) {

                PendingPublishInfoPB pending_publish_info_pb;

                bool parsed = pending_publish_info_pb.ParseFromArray(info.data(),

                                                                     cast_set<int>(info.size()));

                if (!parsed) {

                    LOG(WARNING) << "parse pending publish info failed, tablet_id: " << tablet_id

                                 << " publish_version: " << publish_version;

                }

                engine.add_async_publish_task(pending_publish_info_pb.partition_id(), tablet_id,

                                              publish_version,

                                              pending_publish_info_pb.transaction_id(), true);

                return true;

            };

    MonotonicStopWatch pending_publish_timer;

    pending_publish_timer.start();

    RETURN_IF_ERROR(

            TabletMetaManager::traverse_pending_publish(_meta, load_pending_publish_info_func));

    pending_publish_timer.stop();

    LOG(INFO) << "load pending publish task from meta finished, cost: "

              << pending_publish_timer.elapsed_time_milliseconds() << " ms, data dir: " << _path;

    int64_t rowset_partition_id_eq_0_num = 0;

    for (auto rowset_meta : dir_rowset_metas) {

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

            ++rowset_partition_id_eq_0_num;

        }

    }

    if (rowset_partition_id_eq_0_num > config::ignore_invalid_partition_id_rowset_num) {

        throw Exception(Status::FatalError(

                "rowset partition id eq 0 is {} bigger than config {}, be exit, plz check be.INFO",

                rowset_partition_id_eq_0_num, config::ignore_invalid_partition_id_rowset_num));

    }

    // traverse rowset

    // 1. add committed rowset to txn map

    // 2. add visible rowset to tablet

    // ignore any errors when load tablet or rowset, because fe will repair them after report

    int64_t invalid_rowset_counter = 0;

    for (auto&& rowset_meta : dir_rowset_metas) {

        TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(rowset_meta->tablet_id());

        // tablet maybe dropped, but not drop related rowset meta

        if (tablet == nullptr) {

            VLOG_NOTICE << "could not find tablet id: " << rowset_meta->tablet_id()

                        << ", schema hash: " << rowset_meta->tablet_schema_hash()

                        << ", for rowset: " << rowset_meta->rowset_id() << ", skip this rowset";

            ++invalid_rowset_counter;

            continue;

        }

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

            LOG(WARNING) << "skip tablet_id=" << tablet->tablet_id()

                         << " rowset: " << rowset_meta->rowset_id()

                         << " txn: " << rowset_meta->txn_id();

            continue;

        }

        RowsetSharedPtr rowset;

        Status create_status = tablet->create_rowset(rowset_meta, &rowset);

        if (!create_status) {

            LOG(WARNING) << "could not create rowset from rowsetmeta: "

                         << " rowset_id: " << rowset_meta->rowset_id()

                         << " rowset_type: " << rowset_meta->rowset_type()

                         << " rowset_state: " << rowset_meta->rowset_state();

            continue;

        }

        if (rowset_meta->rowset_state() == RowsetStatePB::COMMITTED &&

            rowset_meta->tablet_uid() == tablet->tablet_uid()) {

            if (!rowset_meta->tablet_schema()) {

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

                RETURN_IF_ERROR(RowsetMetaManager::save(_meta, rowset_meta->tablet_uid(),

                                                        rowset_meta->rowset_id(),

                                                        rowset_meta->get_rowset_pb(), false));

            }

            Status commit_txn_status = _engine.txn_manager()->commit_txn(

                    _meta, rowset_meta->partition_id(), rowset_meta->txn_id(),

                    rowset_meta->tablet_id(), rowset_meta->tablet_uid(), rowset_meta->load_id(),

                    rowset, _engine.pending_local_rowsets().add(rowset_meta->rowset_id()), true);

            if (commit_txn_status || commit_txn_status.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {

                LOG(INFO) << "successfully to add committed rowset: " << rowset_meta->rowset_id()

                          << " to tablet: " << rowset_meta->tablet_id()

                          << " schema hash: " << rowset_meta->tablet_schema_hash()

                          << " for txn: " << rowset_meta->txn_id();

            } else if (commit_txn_status.is<ErrorCode::INTERNAL_ERROR>()) {

                LOG(WARNING) << "failed to add committed rowset: " << rowset_meta->rowset_id()

                             << " to tablet: " << rowset_meta->tablet_id()

                             << " for txn: " << rowset_meta->txn_id()

                             << " error: " << commit_txn_status;

                return commit_txn_status;

            } else {

                LOG(WARNING) << "failed to add committed rowset: " << rowset_meta->rowset_id()

                             << " to tablet: " << rowset_meta->tablet_id()

                             << " for txn: " << rowset_meta->txn_id()

                             << " error: " << commit_txn_status;

            }

        } else if (rowset_meta->rowset_state() == RowsetStatePB::VISIBLE &&

                   rowset_meta->tablet_uid() == tablet->tablet_uid()) {

            if (!rowset_meta->tablet_schema()) {

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

                RETURN_IF_ERROR(RowsetMetaManager::save(_meta, rowset_meta->tablet_uid(),

                                                        rowset_meta->rowset_id(),

                                                        rowset_meta->get_rowset_pb(), false));

            }

            Status publish_status = tablet->add_rowset(rowset);

            if (!publish_status && !publish_status.is<PUSH_VERSION_ALREADY_EXIST>()) {

                LOG(WARNING) << "add visible rowset to tablet failed rowset_id:"

                             << rowset->rowset_id() << " tablet id: " << rowset_meta->tablet_id()

                             << " txn id:" << rowset_meta->txn_id()

                             << " start_version: " << rowset_meta->version().first

                             << " end_version: " << rowset_meta->version().second;

            }

        } else {

            LOG(WARNING) << "find invalid rowset: " << rowset_meta->rowset_id()

                         << " with tablet id: " << rowset_meta->tablet_id()

                         << " tablet uid: " << rowset_meta->tablet_uid()

                         << " schema hash: " << rowset_meta->tablet_schema_hash()

                         << " txn: " << rowset_meta->txn_id()

                         << " current valid tablet uid: " << tablet->tablet_uid();

            ++invalid_rowset_counter;

        }

    }

    int64_t dbm_cnt {0};

    int64_t unknown_dbm_cnt {0};

    auto load_delete_bitmap_func = [this, &dbm_cnt, &unknown_dbm_cnt](int64_t tablet_id,

                                                                      int64_t version,

                                                                      std::string_view val) {

        TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(tablet_id);

        if (!tablet) {

            return true;

        }

        const auto& all_rowsets = tablet->tablet_meta()->all_rs_metas();

        RowsetIdUnorderedSet rowset_ids;

        for (const auto& [_, rowset_meta] : all_rowsets) {

            rowset_ids.insert(rowset_meta->rowset_id());

        }

        DeleteBitmapPB delete_bitmap_pb;

        delete_bitmap_pb.ParseFromArray(val.data(), cast_set<int>(val.size()));

        int rst_ids_size = delete_bitmap_pb.rowset_ids_size();

        int seg_ids_size = delete_bitmap_pb.segment_ids_size();

        int seg_maps_size = delete_bitmap_pb.segment_delete_bitmaps_size();

        CHECK(rst_ids_size == seg_ids_size && seg_ids_size == seg_maps_size);

        for (int i = 0; i < rst_ids_size; ++i) {

            RowsetId rst_id;

            rst_id.init(delete_bitmap_pb.rowset_ids(i));

            // only process the rowset in _rs_metas

            if (rowset_ids.find(rst_id) == rowset_ids.end()) {

                ++unknown_dbm_cnt;

                continue;

            }

            ++dbm_cnt;

            auto seg_id = delete_bitmap_pb.segment_ids(i);

            auto iter = tablet->tablet_meta()->delete_bitmap().delete_bitmap.find(

                    {rst_id, seg_id, version});

            // This version of delete bitmap already exists

            if (iter != tablet->tablet_meta()->delete_bitmap().delete_bitmap.end()) {

                continue;

            }

            auto bitmap = delete_bitmap_pb.segment_delete_bitmaps(i).data();

            tablet->tablet_meta()->delete_bitmap().delete_bitmap[{rst_id, seg_id, version}] =

                    roaring::Roaring::read(bitmap);

        }

        return true;

    };

    MonotonicStopWatch dbm_timer;

    dbm_timer.start();

    RETURN_IF_ERROR(TabletMetaManager::traverse_delete_bitmap(_meta, load_delete_bitmap_func));

    dbm_timer.stop();

    LOG(INFO) << "load delete bitmap from meta finished, cost: "

              << dbm_timer.elapsed_time_milliseconds() << " ms, data dir: " << _path;

    // At startup, we only count these invalid rowset, but do not actually delete it.

    // The actual delete operation is in StorageEngine::_clean_unused_rowset_metas,

    // which is cleaned up uniformly by the background cleanup thread.

    LOG(INFO) << "finish to load tablets from " << _path

              << ", total rowset meta: " << dir_rowset_metas.size()

              << ", invalid rowset num: " << invalid_rowset_counter

              << ", visible/stale rowsets' delete bitmap count: " << dbm_cnt

              << ", invalid rowsets' delete bitmap count: " << unknown_dbm_cnt;

    return Status::OK();

}

// gc unused local tablet dir

void DataDir::_perform_tablet_gc(const std::string& tablet_schema_hash_path, int16_t shard_id) {

    if (_stop_bg_worker) {

        return;

    }

    TTabletId tablet_id = -1;

    TSchemaHash schema_hash = -1;

    bool is_valid = TabletManager::get_tablet_id_and_schema_hash_from_path(

            tablet_schema_hash_path, &tablet_id, &schema_hash);

    if (!is_valid || tablet_id < 1 || schema_hash < 1) [[unlikely]] {

        LOG(WARNING) << "[path gc] unknown path: " << tablet_schema_hash_path;

        return;

    }

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

    if (!tablet || tablet->data_dir() != this) {

        if (tablet) {

            LOG(INFO) << "The tablet in path " << tablet_schema_hash_path

                      << " is not same with the running one: " << tablet->tablet_path()

                      << ", might be the old tablet after migration, try to move it to trash";

        }

        _engine.tablet_manager()->try_delete_unused_tablet_path(this, tablet_id, schema_hash,

                                                                tablet_schema_hash_path, shard_id);

        return;

    }

    _perform_rowset_gc(tablet_schema_hash_path);

}

// gc unused local rowsets under tablet dir

void DataDir::_perform_rowset_gc(const std::string& tablet_schema_hash_path) {

    if (_stop_bg_worker) {

        return;

    }

    TTabletId tablet_id = -1;

    TSchemaHash schema_hash = -1;

    bool is_valid = doris::TabletManager::get_tablet_id_and_schema_hash_from_path(

            tablet_schema_hash_path, &tablet_id, &schema_hash);

    if (!is_valid || tablet_id < 1 || schema_hash < 1) [[unlikely]] {

        LOG(WARNING) << "[path gc] unknown path: " << tablet_schema_hash_path;

        return;

    }

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

    if (!tablet) {

        // Could not found the tablet, maybe it's a dropped tablet, will be reclaimed

        // in the next time `_perform_path_gc_by_tablet`

        return;

    }

    if (tablet->data_dir() != this) {

        // Current running tablet is not in same data_dir, maybe it's a tablet after migration,

        // will be reclaimed in the next time `_perform_path_gc_by_tablet`

        return;

    }

    bool exists;

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

    auto st = io::global_local_filesystem()->list(tablet_schema_hash_path, true, &files, &exists);

    if (!st.ok()) [[unlikely]] {

        LOG(WARNING) << "[path gc] fail to list tablet path " << tablet_schema_hash_path << " : "

                     << st;

        return;

    }

    // Rowset files excluding pending rowsets

    std::vector<std::pair<RowsetId, std::string /* filename */>> rowsets_not_pending;

    for (auto&& file : files) {

        auto rowset_id = extract_rowset_id(file.file_name);

        if (rowset_id.hi == 0) {

            continue; // Not a rowset

        }

        if (_engine.pending_local_rowsets().contains(rowset_id)) {

            continue; // Pending rowset file

        }

        rowsets_not_pending.emplace_back(rowset_id, std::move(file.file_name));

    }

    RowsetIdUnorderedSet rowsets_in_version_map;

    tablet->traverse_rowsets(

            [&rowsets_in_version_map](auto& rs) { rowsets_in_version_map.insert(rs->rowset_id()); },

            true);

    DBUG_EXECUTE_IF("DataDir::_perform_rowset_gc.simulation.slow", {

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

        if (target_tablet_id == tablet_id) {

            LOG(INFO) << "debug point wait tablet to remove rsmgr tabletId=" << tablet_id;

            DBUG_BLOCK;

        }

    });

    auto reclaim_rowset_file = [](const std::string& path) {

        auto st = io::global_local_filesystem()->delete_file(path);

        if (!st.ok()) [[unlikely]] {

            LOG(WARNING) << "[path gc] failed to delete garbage rowset file: " << st;

            return;

        }

        LOG(INFO) << "[path gc] delete garbage path: " << path; // Audit log

    };

    auto should_reclaim = [&, this](const RowsetId& rowset_id) {

        return !rowsets_in_version_map.contains(rowset_id) &&

               !_engine.check_rowset_id_in_unused_rowsets(rowset_id) &&

               RowsetMetaManager::exists(get_meta(), tablet->tablet_uid(), rowset_id)

                       .is<META_KEY_NOT_FOUND>();

    };

    // rowset_id -> is_garbage

    std::unordered_map<RowsetId, bool> checked_rowsets;

    for (auto&& [rowset_id, filename] : rowsets_not_pending) {

        if (_stop_bg_worker) {

            return;

        }

        if (auto it = checked_rowsets.find(rowset_id); it != checked_rowsets.end()) {

            if (it->second) { // Is checked garbage rowset

                reclaim_rowset_file(tablet_schema_hash_path + '/' + filename);

            }

            continue;

        }

        if (should_reclaim(rowset_id)) {

            if (config::path_gc_check_step > 0 &&

                ++_path_gc_step % config::path_gc_check_step == 0) {

                std::this_thread::sleep_for(

                        std::chrono::milliseconds(config::path_gc_check_step_interval_ms));

            }

            reclaim_rowset_file(tablet_schema_hash_path + '/' + filename);

            checked_rowsets.emplace(rowset_id, true);

        } else {

            checked_rowsets.emplace(rowset_id, false);

        }

    }

}

void DataDir::perform_path_gc() {

    if (_stop_bg_worker) {

        return;

    }

    LOG(INFO) << "start to gc data dir " << _path;

    auto data_path = fmt::format("{}/{}", _path, DATA_PREFIX);

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

    bool exists = true;

    const auto& fs = io::global_local_filesystem();

    auto st = fs->list(data_path, false, &shards, &exists);

    if (!st.ok()) [[unlikely]] {

        LOG(WARNING) << "failed to scan data dir: " << st;

        return;

    }

    for (const auto& shard : shards) {

        if (_stop_bg_worker) {

            break;

        }

        if (shard.is_file) {

            continue;

        }

        auto shard_path = fmt::format("{}/{}", data_path, shard.file_name);

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

        st = io::global_local_filesystem()->list(shard_path, false, &tablet_ids, &exists);

        if (!st.ok()) [[unlikely]] {

            LOG(WARNING) << "fail to walk dir, shard_path=" << shard_path << " : " << st;

            continue;

        }

        for (const auto& tablet_id : tablet_ids) {

            if (_stop_bg_worker) {

                break;

            }

            if (tablet_id.is_file) {

                continue;

            }

            auto tablet_id_path = fmt::format("{}/{}", shard_path, tablet_id.file_name);

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

            st = fs->list(tablet_id_path, false, &schema_hashes, &exists);

            if (!st.ok()) [[unlikely]] {

                LOG(WARNING) << "fail to walk dir, tablet_id_path=" << tablet_id_path << " : "

                             << st;

                continue;

            }

            for (auto&& schema_hash : schema_hashes) {

                if (schema_hash.is_file) {

                    continue;

                }

                if (config::path_gc_check_step > 0 &&

                    ++_path_gc_step % config::path_gc_check_step == 0) {

                    std::this_thread::sleep_for(

                            std::chrono::milliseconds(config::path_gc_check_step_interval_ms));

                }

                int16_t shard_id = -1;

                try {

                    shard_id = cast_set<int16_t>(std::stoi(shard.file_name));

                } catch (const std::exception&) {

                    LOG(WARNING) << "failed to stoi shard_id, shard name=" << shard.file_name;

                    continue;

                }

                _perform_tablet_gc(tablet_id_path + '/' + schema_hash.file_name, shard_id);

            }

        }

    }

    LOG(INFO) << "gc data dir path: " << _path << " finished";

}

Status DataDir::update_capacity() {

    RETURN_IF_ERROR(io::global_local_filesystem()->get_space_info(_path, &_disk_capacity_bytes,

                                                                  &_available_bytes));

    disks_total_capacity->set_value(_disk_capacity_bytes);

    disks_avail_capacity->set_value(_available_bytes);

    LOG(INFO) << "path: " << _path << " total capacity: " << _disk_capacity_bytes

              << ", available capacity: " << _available_bytes << ", usage: " << get_usage(0)

              << ", in_use: " << is_used();

    return Status::OK();

}

void DataDir::update_trash_capacity() {

    auto trash_path = fmt::format("{}/{}", _path, TRASH_PREFIX);

    try {

        _trash_used_bytes = _engine.get_file_or_directory_size(trash_path);

    } catch (const std::filesystem::filesystem_error& e) {

        LOG(WARNING) << "update trash capacity failed, path: " << _path << ", err: " << e.what();

        return;

    }

    disks_trash_used_capacity->set_value(_trash_used_bytes);

    LOG(INFO) << "path: " << _path << " trash capacity: " << _trash_used_bytes;

}

void DataDir::update_local_data_size(int64_t size) {

    disks_local_used_capacity->set_value(size);

}

void DataDir::update_remote_data_size(int64_t size) {

    disks_remote_used_capacity->set_value(size);

}

size_t DataDir::tablet_size() const {

    std::lock_guard<std::mutex> l(_mutex);

    return _tablet_set.size();

}

bool DataDir::reach_capacity_limit(int64_t incoming_data_size) {

    double used_pct = get_usage(incoming_data_size);

    int64_t left_bytes = _available_bytes - incoming_data_size;

    if (used_pct >= config::storage_flood_stage_usage_percent / 100.0 &&

        left_bytes <= config::storage_flood_stage_left_capacity_bytes) {

        LOG(WARNING) << "reach capacity limit. used pct: " << used_pct

                     << ", left bytes: " << left_bytes << ", path: " << _path;

        return true;

    }

    return false;

}

void DataDir::disks_compaction_score_increment(int64_t delta) {

    disks_compaction_score->increment(delta);

}

void DataDir::disks_compaction_num_increment(int64_t delta) {

    disks_compaction_num->increment(delta);

}

Status DataDir::move_to_trash(const std::string& tablet_path) {

    if (config::trash_file_expire_time_sec <= 0) {

        LOG(INFO) << "delete tablet dir " << tablet_path

                  << " directly due to trash_file_expire_time_sec is 0";

        RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(tablet_path));

        return delete_tablet_parent_path_if_empty(tablet_path);

    }

    Status res = Status::OK();

    // 1. get timestamp string

    std::string time_str;

    if ((res = gen_timestamp_string(&time_str)) != Status::OK()) {

        LOG(WARNING) << "failed to generate time_string when move file to trash.err code=" << res;

        return res;

    }

    // 2. generate new file path

    // a global counter to avoid file name duplication.

    static std::atomic<uint64_t> delete_counter(0);

    auto trash_root_path =

            fmt::format("{}/{}/{}.{}", _path, TRASH_PREFIX, time_str, delete_counter++);

    auto fs_tablet_path = io::Path(tablet_path);

    auto trash_tablet_path = trash_root_path /