// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "cloud/cloud_storage_engine.h"

#include <bvar/reducer.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <rapidjson/document.h>

#include <rapidjson/encodings.h>

#include <rapidjson/prettywriter.h>

#include <rapidjson/stringbuffer.h>

#include <algorithm>

#include <memory>

#include <variant>

#include "cloud/cloud_base_compaction.h"

#include "cloud/cloud_compaction_stop_token.h"

#include "cloud/cloud_cumulative_compaction.h"

#include "cloud/cloud_cumulative_compaction_policy.h"

#include "cloud/cloud_full_compaction.h"

#include "cloud/cloud_index_change_compaction.h"

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_snapshot_mgr.h"

#include "cloud/cloud_tablet_hotspot.h"

#include "cloud/cloud_tablet_mgr.h"

#include "cloud/cloud_txn_delete_bitmap_cache.h"

#include "cloud/cloud_warm_up_manager.h"

#include "cloud/config.h"

#include "common/config.h"

#include "common/metrics/doris_metrics.h"

#include "common/signal_handler.h"

#include "common/status.h"

#include "core/assert_cast.h"

#include "io/cache/block_file_cache_downloader.h"

#include "io/cache/block_file_cache_factory.h"

#include "io/cache/file_cache_common.h"

#include "io/fs/file_system.h"

#include "io/fs/hdfs_file_system.h"

#include "io/fs/s3_file_system.h"

#include "io/hdfs_util.h"

#include "io/io_common.h"

#include "load/memtable/memtable_flush_executor.h"

#include "runtime/exec_env.h"

#include "runtime/memory/cache_manager.h"

#include "storage/compaction/cumulative_compaction_policy.h"

#include "storage/compaction/cumulative_compaction_time_series_policy.h"

#include "storage/compaction_task_tracker.h"

#include "storage/storage_policy.h"

#include "util/parse_util.h"

#include "util/time.h"

namespace doris {

using namespace std::literals;

bvar::Adder<uint64_t> g_base_compaction_running_task_count("base_compaction_running_task_count");

bvar::Adder<uint64_t> g_full_compaction_running_task_count("full_compaction_running_task_count");

bvar::Adder<uint64_t> g_cumu_compaction_running_task_count(

        "cumulative_compaction_running_task_count");

int get_cumu_thread_num() {

    if (config::max_cumu_compaction_threads > 0) {

        return config::max_cumu_compaction_threads;

    }

    int num_cores = doris::CpuInfo::num_cores();

    return std::min(std::max(int(num_cores * config::cumu_compaction_thread_num_factor), 2), 20);

}

int get_base_thread_num() {

    if (config::max_base_compaction_threads > 0) {

        return config::max_base_compaction_threads;

    }

    int num_cores = doris::CpuInfo::num_cores();

    return std::min(std::max(int(num_cores * config::base_compaction_thread_num_factor), 1), 10);

}

CloudStorageEngine::CloudStorageEngine(const EngineOptions& options)

        : BaseStorageEngine(Type::CLOUD, options.backend_uid),

          _meta_mgr(std::make_unique<cloud::CloudMetaMgr>()),

          _tablet_mgr(std::make_unique<CloudTabletMgr>(*this)),

          _options(options) {

    _cumulative_compaction_policies[CUMULATIVE_SIZE_BASED_POLICY] =

            std::make_shared<CloudSizeBasedCumulativeCompactionPolicy>();

    _cumulative_compaction_policies[CUMULATIVE_TIME_SERIES_POLICY] =

            std::make_shared<CloudTimeSeriesCumulativeCompactionPolicy>();

    _startup_timepoint = std::chrono::system_clock::now();

}

CloudStorageEngine::~CloudStorageEngine() {

    stop();

}

static Status vault_process_error(std::string_view id,

                                  std::variant<S3Conf, cloud::HdfsVaultInfo>& vault, Status err) {

    std::stringstream ss;

    std::visit(

            [&]<typename T>(T& val) {

                if constexpr (std::is_same_v<T, S3Conf>) {

                    ss << val.to_string();

                } else if constexpr (std::is_same_v<T, cloud::HdfsVaultInfo>) {

                    val.SerializeToOstream(&ss);

                }

            },

Unexecuted instantiation: cloud_storage_engine.cpp:_ZZN5dorisL19vault_process_errorESt17basic_string_viewIcSt11char_traitsIcEERSt7variantIJNS_6S3ConfENS_5cloud13HdfsVaultInfoEEENS_6StatusEENK3$_0clIS5_EEDaRT_

Unexecuted instantiation: cloud_storage_engine.cpp:_ZZN5dorisL19vault_process_errorESt17basic_string_viewIcSt11char_traitsIcEERSt7variantIJNS_6S3ConfENS_5cloud13HdfsVaultInfoEEENS_6StatusEENK3$_0clIS7_EEDaRT_

            vault);

    return Status::IOError("Invalid vault, id {}, err {}, detail conf {}", id, err, ss.str());

}

struct VaultCreateFSVisitor {

    VaultCreateFSVisitor(const std::string& id, const cloud::StorageVaultPB_PathFormat& path_format,

                         bool check_fs)

            : id(id), path_format(path_format), check_fs(check_fs) {}

    Status operator()(const S3Conf& s3_conf) const {

        LOG(INFO) << "get new s3 info: " << s3_conf.to_string() << " resource_id=" << id

                  << " check_fs: " << check_fs;

        auto fs = DORIS_TRY(io::S3FileSystem::create(s3_conf, id));

        if (check_fs && !s3_conf.client_conf.role_arn.empty()) {

            bool res = false;

            // just check connectivity, not care object if exist

            auto st = fs->exists("not_exist_object", &res);

            if (!st.ok()) {

                LOG(FATAL) << "failed to check s3 fs, resource_id: " << id << " st: " << st

                           << "s3_conf: " << s3_conf.to_string()

                           << "add enable_check_storage_vault=false to be.conf to skip the check";

            }

        }

        put_storage_resource(id, {std::move(fs), path_format}, 0);

        LOG_INFO("successfully create s3 vault, vault id {}", id);

        return Status::OK();

    }

    // TODO(ByteYue): Make sure enable_java_support is on

    Status operator()(const cloud::HdfsVaultInfo& vault) const {

        auto hdfs_params = io::to_hdfs_params(vault);

        auto fs = DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name, id,

                                                       nullptr, vault.prefix()));

        put_storage_resource(id, {std::move(fs), path_format}, 0);

        LOG_INFO("successfully create hdfs vault, vault id {}", id);

        return Status::OK();

    }

    const std::string& id;

    const cloud::StorageVaultPB_PathFormat& path_format;

    bool check_fs;

};

struct RefreshFSVaultVisitor {

    RefreshFSVaultVisitor(const std::string& id, io::FileSystemSPtr fs,

                          const cloud::StorageVaultPB_PathFormat& path_format)

            : id(id), fs(std::move(fs)), path_format(path_format) {}

    Status operator()(const S3Conf& s3_conf) const {

        DCHECK_EQ(fs->type(), io::FileSystemType::S3) << id;

        auto s3_fs = std::static_pointer_cast<io::S3FileSystem>(fs);

        auto client_holder = s3_fs->client_holder();

        auto st = client_holder->reset(s3_conf.client_conf);

        if (!st.ok()) {

            LOG(WARNING) << "failed to update s3 fs, resource_id=" << id << ": " << st;

        }

        return st;

    }

    Status operator()(const cloud::HdfsVaultInfo& vault) const {

        auto hdfs_params = io::to_hdfs_params(vault);

        auto hdfs_fs =

                DORIS_TRY(io::HdfsFileSystem::create(hdfs_params, hdfs_params.fs_name, id, nullptr,

                                                     vault.has_prefix() ? vault.prefix() : ""));

        auto hdfs = std::static_pointer_cast<io::HdfsFileSystem>(hdfs_fs);

        put_storage_resource(id, {std::move(hdfs), path_format}, 0);

        return Status::OK();

    }

    const std::string& id;

    io::FileSystemSPtr fs;

    const cloud::StorageVaultPB_PathFormat& path_format;

};

Status CloudStorageEngine::open() {

    sync_storage_vault();

    // TODO(plat1ko): DeleteBitmapTxnManager

    _memtable_flush_executor = std::make_unique<MemTableFlushExecutor>();

    // Use file cache disks number

    _memtable_flush_executor->init(

            cast_set<int32_t>(io::FileCacheFactory::instance()->get_cache_instance_size()));

    _calc_delete_bitmap_executor = std::make_unique<CalcDeleteBitmapExecutor>();

    _calc_delete_bitmap_executor->init(config::calc_delete_bitmap_max_thread);

    _calc_delete_bitmap_executor_for_load = std::make_unique<CalcDeleteBitmapExecutor>();

    _calc_delete_bitmap_executor_for_load->init(

            config::calc_delete_bitmap_for_load_max_thread > 0

                    ? config::calc_delete_bitmap_for_load_max_thread

                    : std::max(1, CpuInfo::num_cores() / 2));

    // The default cache is set to 100MB, use memory limit to dynamic adjustment

    bool is_percent = false;

    int64_t delete_bitmap_agg_cache_cache_limit =

            ParseUtil::parse_mem_spec(config::delete_bitmap_dynamic_agg_cache_limit,

                                      MemInfo::mem_limit(), MemInfo::physical_mem(), &is_percent);

    _txn_delete_bitmap_cache = std::make_unique<CloudTxnDeleteBitmapCache>(

            delete_bitmap_agg_cache_cache_limit > config::delete_bitmap_agg_cache_capacity

                    ? delete_bitmap_agg_cache_cache_limit

                    : config::delete_bitmap_agg_cache_capacity);

    RETURN_IF_ERROR(_txn_delete_bitmap_cache->init());

    _committed_rs_mgr = std::make_unique<CloudCommittedRSMgr>();

    RETURN_IF_ERROR(_committed_rs_mgr->init());

    _file_cache_block_downloader = std::make_unique<io::FileCacheBlockDownloader>(*this);

    _cloud_warm_up_manager = std::make_unique<CloudWarmUpManager>(*this);

    _tablet_hotspot = std::make_unique<TabletHotspot>();

    _cloud_snapshot_mgr = std::make_unique<CloudSnapshotMgr>(*this);

    RETURN_NOT_OK_STATUS_WITH_WARN(

            init_stream_load_recorder(ExecEnv::GetInstance()->store_paths()[0].path),

            "init StreamLoadRecorder failed");

    // check cluster id

    RETURN_NOT_OK_STATUS_WITH_WARN(_check_all_root_path_cluster_id(), "fail to check cluster id");

    RETURN_NOT_OK_STATUS_WITH_WARN(ThreadPoolBuilder("SyncLoadForTabletsThreadPool")

                                           .set_max_threads(config::sync_load_for_tablets_thread)

                                           .set_min_threads(config::sync_load_for_tablets_thread)

                                           .build(&_sync_load_for_tablets_thread_pool),

                                   "fail to build SyncLoadForTabletsThreadPool");

    RETURN_NOT_OK_STATUS_WITH_WARN(ThreadPoolBuilder("WarmupCacheAsyncThreadPool")

                                           .set_max_threads(config::warmup_cache_async_thread)

                                           .set_min_threads(config::warmup_cache_async_thread)

                                           .build(&_warmup_cache_async_thread_pool),

                                   "fail to build WarmupCacheAsyncThreadPool");

    return Status::OK();

}

void CloudStorageEngine::stop() {

    if (_stopped) {

        return;

    }

    _stopped = true;

    _stop_background_threads_latch.count_down();

    for (auto&& t : _bg_threads) {

        if (t) {

            t->join();

        }

    }

    if (_base_compaction_thread_pool) {

        _base_compaction_thread_pool->shutdown();

    }

    if (_cumu_compaction_thread_pool) {

        _cumu_compaction_thread_pool->shutdown();

    }

    _adaptive_thread_controller.stop();

    LOG(INFO) << "Cloud storage engine is stopped.";

    if (_calc_tablet_delete_bitmap_task_thread_pool) {

        _calc_tablet_delete_bitmap_task_thread_pool->shutdown();

    }

    if (_sync_delete_bitmap_thread_pool) {

        _sync_delete_bitmap_thread_pool->shutdown();

    }

}

bool CloudStorageEngine::stopped() {

    return _stopped;

}

Result<BaseTabletSPtr> CloudStorageEngine::get_tablet(int64_t tablet_id,

                                                      SyncRowsetStats* sync_stats,

                                                      bool force_use_only_cached,

                                                      bool cache_on_miss) {

    return _tablet_mgr

            ->get_tablet(tablet_id, false, true, sync_stats, force_use_only_cached, cache_on_miss)

            .transform([](auto&& t) { return static_pointer_cast<BaseTablet>(std::move(t)); });

}

Status CloudStorageEngine::get_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta,

                                           bool force_use_only_cached) {

    if (tablet_meta == nullptr) {

        return Status::InvalidArgument("tablet_meta output is null");

    }

#if 0

    if (_tablet_mgr && _tablet_mgr->peek_tablet_meta(tablet_id, tablet_meta)) {

        return Status::OK();

    }

    if (force_use_only_cached) {

        return Status::NotFound("tablet meta {} not found in cache", tablet_id);

    }

#endif

    if (_meta_mgr == nullptr) {

        return Status::InternalError("cloud meta manager is not initialized");

    }

    return _meta_mgr->get_tablet_meta(tablet_id, tablet_meta);

}

Status CloudStorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sptr) {

    RETURN_IF_ERROR(Thread::create(

            "CloudStorageEngine", "refresh_s3_info_thread",

            [this]() { this->_refresh_storage_vault_info_thread_callback(); },

            &_bg_threads.emplace_back()));

    LOG(INFO) << "refresh s3 info thread started";

    RETURN_IF_ERROR(Thread::create(

            "CloudStorageEngine", "vacuum_stale_rowsets_thread",

            [this]() { this->_vacuum_stale_rowsets_thread_callback(); },

            &_bg_threads.emplace_back()));

    LOG(INFO) << "vacuum stale rowsets thread started";

    RETURN_IF_ERROR(Thread::create(

            "CloudStorageEngine", "sync_tablets_thread",

            [this]() { this->_sync_tablets_thread_callback(); }, &_bg_threads.emplace_back()));

    LOG(INFO) << "sync tablets thread started";

    RETURN_IF_ERROR(Thread::create(

            "CloudStorageEngine", "evict_querying_rowset_thread",

            [this]() { this->_evict_quring_rowset_thread_callback(); },

            &_evict_quering_rowset_thread));

    LOG(INFO) << "evict quering thread started";

    // add calculate tablet delete bitmap task thread pool

    RETURN_IF_ERROR(ThreadPoolBuilder("TabletCalDeleteBitmapThreadPool")

                            .set_min_threads(config::calc_tablet_delete_bitmap_task_max_thread)

                            .set_max_threads(config::calc_tablet_delete_bitmap_task_max_thread)

                            .build(&_calc_tablet_delete_bitmap_task_thread_pool));

    RETURN_IF_ERROR(ThreadPoolBuilder("SyncDeleteBitmapThreadPool")

                            .set_min_threads(config::sync_delete_bitmap_task_max_thread)

                            .set_max_threads(config::sync_delete_bitmap_task_max_thread)

                            .build(&_sync_delete_bitmap_thread_pool));

    // TODO(plat1ko): check_bucket_enable_versioning_thread

    // compaction tasks producer thread

    int base_thread_num = get_base_thread_num();

    int cumu_thread_num = get_cumu_thread_num();

    RETURN_IF_ERROR(ThreadPoolBuilder("BaseCompactionTaskThreadPool")

                            .set_min_threads(base_thread_num)

                            .set_max_threads(base_thread_num)

                            .build(&_base_compaction_thread_pool));

    RETURN_IF_ERROR(ThreadPoolBuilder("CumuCompactionTaskThreadPool")

                            .set_min_threads(cumu_thread_num)

                            .set_max_threads(cumu_thread_num)

                            .build(&_cumu_compaction_thread_pool));

    RETURN_IF_ERROR(Thread::create(

            "StorageEngine", "compaction_tasks_producer_thread",

            [this]() { this->_compaction_tasks_producer_callback(); },

            &_bg_threads.emplace_back()));

    LOG(INFO) << "compaction tasks producer thread started,"

              << " base thread num " << base_thread_num << " cumu thread num " << cumu_thread_num;

    RETURN_IF_ERROR(Thread::create(

            "StorageEngine", "lease_compaction_thread",

            [this]() { this->_lease_compaction_thread_callback(); }, &_bg_threads.emplace_back()));

    LOG(INFO) << "lease compaction thread started";

    RETURN_IF_ERROR(Thread::create(

            "StorageEngine", "check_tablet_delete_bitmap_score_thread",

            [this]() { this->_check_tablet_delete_bitmap_score_callback(); },

            &_bg_threads.emplace_back()));

    LOG(INFO) << "check tablet delete bitmap score thread started";

    _start_adaptive_thread_controller();

    return Status::OK();

}

void CloudStorageEngine::sync_storage_vault() {

    cloud::StorageVaultInfos vault_infos;

    bool enable_storage_vault = false;

    auto st = _meta_mgr->get_storage_vault_info(&vault_infos, &enable_storage_vault);

    if (!st.ok()) {

        LOG(WARNING) << "failed to get storage vault info. err=" << st;

        return;

    }

    if (vault_infos.empty()) {

        LOG(WARNING) << "empty storage vault info";

        return;

    }

    bool check_storage_vault = false;

    bool expected = false;

    if (first_sync_storage_vault.compare_exchange_strong(expected, true)) {

        check_storage_vault = config::enable_check_storage_vault;

        LOG(INFO) << "first sync storage vault info, BE try to check iam role connectivity, "

                     "check_storage_vault="

                  << check_storage_vault;

    }

    for (auto& [id, vault_info, path_format] : vault_infos) {

        auto fs = get_filesystem(id);

        auto status =

                (fs == nullptr)

                        ? std::visit(VaultCreateFSVisitor {id, path_format, check_storage_vault},

                                     vault_info)

                        : std::visit(RefreshFSVaultVisitor {id, std::move(fs), path_format},

                                     vault_info);

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

            LOG(WARNING) << vault_process_error(id, vault_info, std::move(st));

        }

    }

    if (auto& id = std::get<0>(vault_infos.back());

        (latest_fs() == nullptr || latest_fs()->id() != id) && !enable_storage_vault) {

        set_latest_fs(get_filesystem(id));

    }

}

// We should enable_java_support if we want to use hdfs vault

void CloudStorageEngine::_refresh_storage_vault_info_thread_callback() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::seconds(config::refresh_s3_info_interval_s))) {

        sync_storage_vault();

        // The other place that rebuilds the S3 rate limiter is S3ClientFactory::create(), which

        // is not called when an existing vault's conf is unchanged. Trigger the check here as well

        // so that dynamically modified s3_{get,put}_* rate limiter configs take effect within

        // refresh_s3_info_interval_s even when no vault is created or its conf does not change.

        // Gate it behind enable_s3_rate_limiter so that clusters with rate limiting disabled

        // (e.g. HDFS-only vaults) do not force-initialize S3ClientFactory / the AWS SDK here.

        if (config::enable_s3_rate_limiter) {

            check_s3_rate_limiter_config_changed();

        }

    }

}

void CloudStorageEngine::_vacuum_stale_rowsets_thread_callback() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::seconds(config::vacuum_stale_rowsets_interval_s))) {

        _tablet_mgr->vacuum_stale_rowsets(_stop_background_threads_latch);

    }

}

void CloudStorageEngine::_sync_tablets_thread_callback() {

    while (!_stop_background_threads_latch.wait_for(

            std::chrono::seconds(config::schedule_sync_tablets_interval_s))) {

        _tablet_mgr->sync_tablets(_stop_background_threads_latch);

    }

}

void CloudStorageEngine::get_cumu_compaction(

        int64_t tablet_id, std::vector<std::shared_ptr<CloudCumulativeCompaction>>& res) {

    std::lock_guard lock(_compaction_mtx);

    if (auto it = _submitted_cumu_compactions.find(tablet_id);

        it != _submitted_cumu_compactions.end()) {

        res = it->second;

    }

}

Status CloudStorageEngine::_adjust_compaction_thread_num() {

    int base_thread_num = get_base_thread_num();

    if (!_base_compaction_thread_pool || !_cumu_compaction_thread_pool) {

        LOG(WARNING) << "base or cumu compaction thread pool is not created";

        return Status::Error<ErrorCode::INTERNAL_ERROR, false>("");

    }

    if (_base_compaction_thread_pool->max_threads() != base_thread_num) {

        int old_max_threads = _base_compaction_thread_pool->max_threads();

        Status status = _base_compaction_thread_pool->set_max_threads(base_thread_num);

        if (status.ok()) {

            VLOG_NOTICE << "update base compaction thread pool max_threads from " << old_max_threads

                        << " to " << base_thread_num;

        }

    }

    if (_base_compaction_thread_pool->min_threads() != base_thread_num) {

        int old_min_threads = _base_compaction_thread_pool->min_threads();

        Status status = _base_compaction_thread_pool->set_min_threads(base_thread_num);

        if (status.ok()) {

            VLOG_NOTICE << "update base compaction thread pool min_threads from " << old_min_threads

                        << " to " << base_thread_num;

        }

    }

    int cumu_thread_num = get_cumu_thread_num();

    if (_cumu_compaction_thread_pool->max_threads() != cumu_thread_num) {

        int old_max_threads = _cumu_compaction_thread_pool->max_threads();

        Status status = _cumu_compaction_thread_pool->set_max_threads(cumu_thread_num);

        if (status.ok()) {

            VLOG_NOTICE << "update cumu compaction thread pool max_threads from " << old_max_threads

                        << " to " << cumu_thread_num;

        }

    }

    if (_cumu_compaction_thread_pool->min_threads() != cumu_thread_num) {

        int old_min_threads = _cumu_compaction_thread_pool->min_threads();

        Status status = _cumu_compaction_thread_pool->set_min_threads(cumu_thread_num);

        if (status.ok()) {

            VLOG_NOTICE << "update cumu compaction thread pool min_threads from " << old_min_threads

                        << " to " << cumu_thread_num;

        }

    }

    return Status::OK();

}

void CloudStorageEngine::_compaction_tasks_producer_callback() {

    LOG(INFO) << "try to start compaction producer process!";

    int round = 0;

    CompactionType compaction_type;

    // Used to record the time when the score metric was last updated.

    // The update of the score metric is accompanied by the logic of selecting the tablet.

    // If there is no slot available, the logic of selecting the tablet will be terminated,

    // which causes the score metric update to be terminated.

    // In order to avoid this situation, we need to update the score regularly.

    int64_t last_cumulative_score_update_time = 0;

    int64_t last_base_score_update_time = 0;

    static const int64_t check_score_interval_ms = 5000; // 5 secs

    int64_t interval = config::generate_compaction_tasks_interval_ms;

    do {

        int64_t cur_time = UnixMillis();

        if (!config::disable_auto_compaction) {

            Status st = _adjust_compaction_thread_num();

            if (!st.ok()) {

                break;

            }

            bool check_score = false;

            if (round < config::cumulative_compaction_rounds_for_each_base_compaction_round) {

                compaction_type = CompactionType::CUMULATIVE_COMPACTION;

                round++;

                if (cur_time - last_cumulative_score_update_time >= check_score_interval_ms) {

                    check_score = true;

                    last_cumulative_score_update_time = cur_time;

                }

            } else {

                compaction_type = CompactionType::BASE_COMPACTION;

                round = 0;

                if (cur_time - last_base_score_update_time >= check_score_interval_ms) {

                    check_score = true;

                    last_base_score_update_time = cur_time;

                }

            }

            std::unique_ptr<ThreadPool>& thread_pool =

                    (compaction_type == CompactionType::CUMULATIVE_COMPACTION)

                            ? _cumu_compaction_thread_pool

                            : _base_compaction_thread_pool;

            VLOG_CRITICAL << "compaction thread pool. type: "

                          << (compaction_type == CompactionType::CUMULATIVE_COMPACTION ? "CUMU"

                                                                                       : "BASE")

                          << ", num_threads: " << thread_pool->num_threads()

                          << ", num_threads_pending_start: "

                          << thread_pool->num_threads_pending_start()

                          << ", num_active_threads: " << thread_pool->num_active_threads()

                          << ", max_threads: " << thread_pool->max_threads()

                          << ", min_threads: " << thread_pool->min_threads()

                          << ", num_total_queued_tasks: " << thread_pool->get_queue_size();

            std::vector<CloudTabletSPtr> tablets_compaction =

                    _generate_cloud_compaction_tasks(compaction_type, check_score);

            /// Regardless of whether the tablet is submitted for compaction or not,

            /// we need to call 'reset_compaction' to clean up the base_compaction or cumulative_compaction objects

            /// in the tablet, because these two objects store the tablet's own shared_ptr.

            /// If it is not cleaned up, the reference count of the tablet will always be greater than 1,

            /// thus cannot be collected by the garbage collector. (TabletManager::start_trash_sweep)

            for (const auto& tablet : tablets_compaction) {

                Status status = submit_compaction_task(tablet, compaction_type);

                if (status.ok()) continue;

                if ((!status.is<ErrorCode::BE_NO_SUITABLE_VERSION>() &&

                     !status.is<ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION>()) ||

                    VLOG_DEBUG_IS_ON) {

                    LOG(WARNING) << "failed to submit compaction task for tablet: "

                                 << tablet->tablet_id() << ", err: " << status;

                }

            }

            interval = config::generate_compaction_tasks_interval_ms;

        } else {

            interval = config::check_auto_compaction_interval_seconds * 1000;

        }

        int64_t end_time = UnixMillis();

        DorisMetrics::instance()->compaction_producer_callback_a_round_time->set_value(end_time -

                                                                                       cur_time);

    } while (!_stop_background_threads_latch.wait_for(std::chrono::milliseconds(interval)));

}

void CloudStorageEngine::unregister_index_change_compaction(int64_t tablet_id,

                                                            bool is_base_compact) {

    std::lock_guard lock(_compaction_mtx);

    if (is_base_compact) {

        _submitted_index_change_base_compaction.erase(tablet_id);

    } else {

        _submitted_index_change_cumu_compaction.erase(tablet_id);

    }

}

bool CloudStorageEngine::register_index_change_compaction(

        std::shared_ptr<CloudIndexChangeCompaction> compact, int64_t tablet_id,

        bool is_base_compact, std::string& err_reason) {

    std::lock_guard lock(_compaction_mtx);

    if (is_base_compact) {

        if (_submitted_base_compactions.contains(tablet_id) ||

            _submitted_full_compactions.contains(tablet_id) ||

            _submitted_index_change_base_compaction.contains(tablet_id)) {

            std::stringstream ss;

            ss << "reason:" << ((int)_submitted_base_compactions.contains(tablet_id)) << ", "

               << ((int)_submitted_full_compactions.contains(tablet_id)) << ", "

               << ((int)_submitted_index_change_base_compaction.contains(tablet_id));

            err_reason = ss.str();

            return false;

        } else {

            _submitted_index_change_base_compaction[tablet_id] = compact;

            return true;

        }

    } else {

        if (_tablet_preparing_cumu_compaction.contains(tablet_id) ||

            _submitted_cumu_compactions.contains(tablet_id) ||

            _submitted_index_change_cumu_compaction.contains(tablet_id)) {

            std::stringstream ss;

            ss << "reason:" << ((int)_tablet_preparing_cumu_compaction.contains(tablet_id)) << ", "

               << ((int)_submitted_cumu_compactions.contains(tablet_id)) << ", "

               << ((int)_submitted_index_change_cumu_compaction.contains(tablet_id));

            err_reason = ss.str();

            return false;

        } else {

            _submitted_index_change_cumu_compaction[tablet_id] = compact;

        }

        return true;

    }

}

std::vector<CloudTabletSPtr> CloudStorageEngine::_generate_cloud_compaction_tasks(

        CompactionType compaction_type, bool check_score) {

    std::vector<std::shared_ptr<CloudTablet>> tablets_compaction;

    int64_t max_compaction_score = 0;

    std::unordered_set<int64_t> tablet_preparing_cumu_compaction;

    std::unordered_map<int64_t, std::vector<std::shared_ptr<CloudCumulativeCompaction>>>

            submitted_cumu_compactions;

    std::unordered_map<int64_t, std::shared_ptr<CloudBaseCompaction>> submitted_base_compactions;

    std::unordered_map<int64_t, std::shared_ptr<CloudFullCompaction>> submitted_full_compactions;

    std::unordered_map<int64_t, std::shared_ptr<CloudIndexChangeCompaction>>

            submitted_index_change_cumu_compactions;

    std::unordered_map<int64_t, std::shared_ptr<CloudIndexChangeCompaction>>

            submitted_index_change_base_compactions;

    {

        std::lock_guard lock(_compaction_mtx);

        tablet_preparing_cumu_compaction = _tablet_preparing_cumu_compaction;

        submitted_cumu_compactions = _submitted_cumu_compactions;

        submitted_base_compactions = _submitted_base_compactions;

        submitted_full_compactions = _submitted_full_compactions;

        submitted_index_change_cumu_compactions = _submitted_index_change_cumu_compaction;

        submitted_index_change_base_compactions = _submitted_index_change_base_compaction;

    }

    bool need_pick_tablet = true;

    int thread_per_disk =

            config::compaction_task_num_per_fast_disk; // all disks are fast in cloud mode

    int num_cumu =

            std::accumulate(submitted_cumu_compactions.begin(), submitted_cumu_compactions.end(), 0,

                            [](int a, auto& b) { return a + b.second.size(); });

    int num_base =

            cast_set<int>(submitted_base_compactions.size() + submitted_full_compactions.size());

    int n = thread_per_disk - num_cumu - num_base;

    if (compaction_type == CompactionType::BASE_COMPACTION) {

        // We need to reserve at least one thread for cumulative compaction,

        // because base compactions may take too long to complete, which may

        // leads to "too many rowsets" error.

        int base_n = std::min(config::max_base_compaction_task_num_per_disk, thread_per_disk - 1) -

                     num_base;

        n = std::min(base_n, n);

    }

    if (n <= 0) { // No threads available

        if (!check_score) return tablets_compaction;

        need_pick_tablet = false;

        n = 0;

    }

    // Return true for skipping compaction

    std::function<bool(CloudTablet*)> filter_out;

    if (compaction_type == CompactionType::BASE_COMPACTION) {

        filter_out = [&submitted_base_compactions, &submitted_full_compactions,

                      &submitted_index_change_base_compactions](CloudTablet* t) {

            return submitted_base_compactions.contains(t->tablet_id()) ||

                   submitted_full_compactions.contains(t->tablet_id()) ||

                   submitted_index_change_base_compactions.contains(t->tablet_id()) ||

                   t->tablet_state() != TABLET_RUNNING;

        };

    } else if (config::enable_parallel_cumu_compaction) {

        filter_out = [&tablet_preparing_cumu_compaction,

                      &submitted_index_change_cumu_compactions](CloudTablet* t) {

            return tablet_preparing_cumu_compaction.contains(t->tablet_id()) ||

                   submitted_index_change_cumu_compactions.contains(t->tablet_id()) ||

                   (t->tablet_state() != TABLET_RUNNING &&

                    (!config::enable_new_tablet_do_compaction || t->alter_version() == -1));

        };

    } else {

        filter_out = [&tablet_preparing_cumu_compaction, &submitted_cumu_compactions,

                      &submitted_index_change_cumu_compactions](CloudTablet* t) {

            return tablet_preparing_cumu_compaction.contains(t->tablet_id()) ||

                   submitted_index_change_cumu_compactions.contains(t->tablet_id()) ||

                   submitted_cumu_compactions.contains(t->tablet_id()) ||

                   (t->tablet_state() != TABLET_RUNNING &&

                    (!config::enable_new_tablet_do_compaction || t->alter_version() == -1));

        };

    }

    // Even if need_pick_tablet is false, we still need to call find_best_tablet_to_compaction(),

    // So that we can update the max_compaction_score metric.

    do {

        std::vector<CloudTabletSPtr> tablets;

        auto st = tablet_mgr().get_topn_tablets_to_compact(n, compaction_type, filter_out, &tablets,

                                                           &max_compaction_score);

        if (!st.ok()) {

            LOG(WARNING) << "failed to get tablets to compact, err=" << st;

            break;

        }

        if (!need_pick_tablet) break;

        tablets_compaction = std::move(tablets);

    } while (false);

    if (max_compaction_score > 0) {

        if (compaction_type == CompactionType::BASE_COMPACTION) {

            DorisMetrics::instance()->tablet_base_max_compaction_score->set_value(

                    max_compaction_score);

        } else {

            DorisMetrics::instance()->tablet_cumulative_max_compaction_score->set_value(

                    max_compaction_score);

        }

    }

    return tablets_compaction;

}

Status CloudStorageEngine::_request_tablet_global_compaction_lock(

        ReaderType compaction_type, const CloudTabletSPtr& tablet,

        std::shared_ptr<CloudCompactionMixin> compaction) {

    long now = duration_cast<std::chrono::milliseconds>(

                       std::chrono::system_clock::now().time_since_epoch())

                       .count();

    if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) {

        auto cumu_compaction = static_pointer_cast<CloudCumulativeCompaction>(compaction);

        if (auto st = cumu_compaction->request_global_lock(); !st.ok()) {

            LOG_WARNING("failed to request cumu compactoin global lock")

                    .tag("tablet id", tablet->tablet_id())

                    .tag("msg", st.to_string());

            tablet->set_last_cumu_compaction_failure_time(now);

            return st;

        }

        {

            std::lock_guard lock(_compaction_mtx);

            _executing_cumu_compactions[tablet->tablet_id()].push_back(cumu_compaction);

        }

        return Status::OK();

    } else if (compaction_type == ReaderType::READER_BASE_COMPACTION) {

        auto base_compaction = static_pointer_cast<CloudBaseCompaction>(compaction);

        if (auto st = base_compaction->request_global_lock(); !st.ok()) {

            LOG_WARNING("failed to request base compactoin global lock")

                    .tag("tablet id", tablet->tablet_id())

                    .tag("msg", st.to_string());

            tablet->set_last_base_compaction_failure_time(now);

            return st;

        }

        {

            std::lock_guard lock(_compaction_mtx);

            _executing_base_compactions[tablet->tablet_id()] = base_compaction;

        }

        return Status::OK();

    } else if (compaction_type == ReaderType::READER_FULL_COMPACTION) {

        auto full_compaction = static_pointer_cast<CloudFullCompaction>(compaction);

        if (auto st = full_compaction->request_global_lock(); !st.ok()) {

            LOG_WARNING("failed to request full compactoin global lock")

                    .tag("tablet id", tablet->tablet_id())

                    .tag("msg", st.to_string());

            tablet->set_last_full_compaction_failure_time(now);

            return st;

        }

        {

            std::lock_guard lock(_compaction_mtx);

            _executing_full_compactions[tablet->tablet_id()] = full_compaction;

        }

        return Status::OK();

    } else {

        LOG(WARNING) << "unsupport compaction task for tablet: " << tablet->tablet_id()

                     << ", compaction name: " << compaction->compaction_name();

        return Status::NotFound("Unsupport compaction type {}", compaction->compaction_name());

    }

}

Status CloudStorageEngine::_submit_base_compaction_task(const CloudTabletSPtr& tablet,

                                                        int trigger_method) {

    using namespace std::chrono;

    {

        std::lock_guard lock(_compaction_mtx);

        // Take a placeholder for base compaction

        auto [_, success] = _submitted_base_compactions.emplace(tablet->tablet_id(), nullptr);

        if (!success) {

            return Status::AlreadyExist(

                    "other base compaction or full compaction is submitted, tablet_id={}",

                    tablet->tablet_id());

        }

    }

    auto compaction = std::make_shared<CloudBaseCompaction>(*this, tablet);

    auto st = compaction->prepare_compact();

    if (!st.ok()) {

        long now = duration_cast<std::chrono::milliseconds>(

                           std::chrono::system_clock::now().time_since_epoch())

                           .count();

        tablet->set_last_base_compaction_failure_time(now);

        std::lock_guard lock(_compaction_mtx);

        _submitted_base_compactions.erase(tablet->tablet_id());

        return st;

    }

    // Register task with CompactionTaskTracker as PENDING

    auto* tracker = CompactionTaskTracker::instance();

    int64_t compaction_id = compaction->compaction_id();

    {

        CompactionTaskInfo info;

        info.compaction_id = compaction_id;

        info.tablet_id = tablet->tablet_id();

        info.table_id = tablet->table_id();

        info.partition_id = tablet->partition_id();

        info.compaction_type = CompactionProfileType::BASE;

        info.status = CompactionTaskStatus::PENDING;

        info.trigger_method = static_cast<TriggerMethod>(trigger_method);

        info.scheduled_time_ms =

                duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

        info.backend_id = BackendOptions::get_backend_id();

        info.compaction_score = tablet->get_real_compaction_score();

        info.input_rowsets_count = compaction->input_rowsets_count();

        info.input_row_num = compaction->input_row_num_value();

        info.input_data_size = compaction->input_rowsets_data_size();

        info.input_index_size = compaction->input_rowsets_index_size();

        info.input_total_size = compaction->input_rowsets_total_size();

        info.input_segments_num = compaction->input_segments_num_value();

        info.input_version_range = compaction->input_version_range_str();

        info.is_vertical = compaction->is_vertical();

        tracker->register_task(std::move(info));

    }

    {

        std::lock_guard lock(_compaction_mtx);

        _submitted_base_compactions[tablet->tablet_id()] = compaction;

    }

    st = _base_compaction_thread_pool->submit_func([=, this, compaction = std::move(compaction)]() {

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

        DorisMetrics::instance()->base_compaction_task_pending_total->set_value(

                _base_compaction_thread_pool->get_queue_size());

        g_base_compaction_running_task_count << 1;

        signal::tablet_id = tablet->tablet_id();

        Defer defer {[&]() {

            // Idempotent cleanup: remove task from tracker

            CompactionTaskTracker::instance()->remove_task(compaction_id);

            g_base_compaction_running_task_count << -1;

            std::lock_guard lock(_compaction_mtx);

            _submitted_base_compactions.erase(tablet->tablet_id());

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

            DorisMetrics::instance()->base_compaction_task_pending_total->set_value(

                    _base_compaction_thread_pool->get_queue_size());

        }};

        auto st = _request_tablet_global_compaction_lock(ReaderType::READER_BASE_COMPACTION, tablet,

                                                         compaction);

        if (!st.ok()) return;

        // Update tracker to RUNNING after acquiring global lock

        {

            RunningStats rs;

            rs.start_time_ms =

                    duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

            CompactionTaskTracker::instance()->update_to_running(compaction_id, rs);

        }

        st = compaction->execute_compact();

        if (!st.ok()) {

            // Error log has been output in `execute_compact`

            long now = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

            tablet->set_last_base_compaction_failure_time(now);

        }

        std::lock_guard lock(_compaction_mtx);

        _executing_base_compactions.erase(tablet->tablet_id());

    });

    DorisMetrics::instance()->base_compaction_task_pending_total->set_value(

            _base_compaction_thread_pool->get_queue_size());

    if (!st.ok()) {

        tracker->remove_task(compaction_id);

        std::lock_guard lock(_compaction_mtx);

        _submitted_base_compactions.erase(tablet->tablet_id());

        return Status::InternalError("failed to submit base compaction, tablet_id={}",

                                     tablet->tablet_id());

    }

    return st;

}

Status CloudStorageEngine::_submit_cumulative_compaction_task(const CloudTabletSPtr& tablet,

                                                              int trigger_method) {

    using namespace std::chrono;

    {

        std::lock_guard lock(_compaction_mtx);

        if (!config::enable_parallel_cumu_compaction &&

            _submitted_cumu_compactions.count(tablet->tablet_id())) {

            return Status::AlreadyExist("other cumu compaction is submitted, tablet_id={}",

                                        tablet->tablet_id());

        }

        auto [_, success] = _tablet_preparing_cumu_compaction.insert(tablet->tablet_id());

        if (!success) {

            return Status::AlreadyExist("other cumu compaction is preparing, tablet_id={}",

                                        tablet->tablet_id());

        }

    }

    auto compaction = std::make_shared<CloudCumulativeCompaction>(*this, tablet);

    auto st = compaction->prepare_compact();

    if (!st.ok()) {

        long now = duration_cast<std::chrono::milliseconds>(

                           std::chrono::system_clock::now().time_since_epoch())

                           .count();

        if (!st.is<ErrorCode::CUMULATIVE_MEET_DELETE_VERSION>()) {

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

                // Backoff strategy if no suitable version

                tablet->last_cumu_no_suitable_version_ms = now;

            } else {

                tablet->set_last_cumu_compaction_failure_time(now);

            }

        }

        std::lock_guard lock(_compaction_mtx);

        _tablet_preparing_cumu_compaction.erase(tablet->tablet_id());

        return st;

    }

    // Register task with CompactionTaskTracker as PENDING

    // IMPORTANT: use compaction->compaction_id(), NOT tracker->next_compaction_id(),

    // because the Compaction constructor already allocated an ID via the tracker.

    auto* tracker = CompactionTaskTracker::instance();

    int64_t compaction_id = compaction->compaction_id();

    {

        CompactionTaskInfo info;

        info.compaction_id = compaction_id;

        info.tablet_id = tablet->tablet_id();

        info.table_id = tablet->table_id();

        info.partition_id = tablet->partition_id();

        info.compaction_type = CompactionProfileType::CUMULATIVE;

        info.status = CompactionTaskStatus::PENDING;

        info.trigger_method = static_cast<TriggerMethod>(trigger_method);

        info.scheduled_time_ms =

                duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

        info.backend_id = BackendOptions::get_backend_id();

        info.compaction_score = tablet->get_real_compaction_score();

        info.input_rowsets_count = compaction->input_rowsets_count();

        info.input_row_num = compaction->input_row_num_value();

        info.input_data_size = compaction->input_rowsets_data_size();

        info.input_index_size = compaction->input_rowsets_index_size();

        info.input_total_size = compaction->input_rowsets_total_size();

        info.input_segments_num = compaction->input_segments_num_value();

        info.input_version_range = compaction->input_version_range_str();

        info.is_vertical = compaction->is_vertical();

        tracker->register_task(std::move(info));

    }

    {

        std::lock_guard lock(_compaction_mtx);

        _tablet_preparing_cumu_compaction.erase(tablet->tablet_id());

        _submitted_cumu_compactions[tablet->tablet_id()].push_back(compaction);

    }

    auto erase_submitted_cumu_compaction = [=, this]() {

        std::lock_guard lock(_compaction_mtx);

        auto it = _submitted_cumu_compactions.find(tablet->tablet_id());

        DCHECK(it != _submitted_cumu_compactions.end());

        auto& compactions = it->second;

        auto it1 = std::find(compactions.begin(), compactions.end(), compaction);

        DCHECK(it1 != compactions.end());

        compactions.erase(it1);

        if (compactions.empty()) { // No compactions on this tablet, erase key

            _submitted_cumu_compactions.erase(it);

            // No cumu compaction on this tablet, reset `last_cumu_no_suitable_version_ms` to enable this tablet to

            // enter the compaction scheduling candidate set. The purpose of doing this is to have at least one BE perform