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

#include <absl/strings/str_split.h>

#include <curl/curl.h>

#include <fcntl.h>

#include <fmt/format.h>

#include <gen_cpp/AgentService_types.h>

#include <gen_cpp/BackendService.h>

#include <gen_cpp/HeartbeatService_types.h>

#include <gen_cpp/MasterService_types.h>

#include <gen_cpp/Status_types.h>

#include <gen_cpp/Types_constants.h>

#include <sys/stat.h>

#include <filesystem>

#include <memory>

#include <mutex>

#include <ostream>

#include <shared_mutex>

#include <system_error>

#include <unordered_map>

#include <unordered_set>

#include <utility>

#include <vector>

#include "common/config.h"

#include "common/logging.h"

#include "io/fs/file_system.h"

#include "io/fs/local_file_system.h"

#include "io/fs/path.h"

#include "runtime/cluster_info.h"

#include "runtime/memory/mem_tracker_limiter.h"

#include "runtime/thread_context.h"

#include "service/http/http_client.h"

#include "service/http/utils.h"

#include "storage/binlog.h"

#include "storage/data_dir.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/pb_helper.h"

#include "storage/rowset/rowset.h"

#include "storage/snapshot/snapshot_manager.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_manager.h"

#include "storage/tablet/tablet_meta.h"

#include "util/client_cache.h"

#include "util/debug_points.h"

#include "util/defer_op.h"

#include "util/network_util.h"

#include "util/security.h"

#include "util/stopwatch.hpp"

#include "util/thrift_rpc_helper.h"

#include "util/trace.h"

using std::stringstream;

namespace doris {

using namespace ErrorCode;

namespace {

/// if binlog file exist, then check if binlog file md5sum equal

/// if equal, then skip link file

/// if not equal, then return error

/// return value: if binlog file not exist, then return to binlog file path

Result<std::string> check_dest_binlog_valid(const std::string& tablet_dir,

                                            const std::string& clone_dir,

                                            const std::string& clone_file, bool* skip_link_file) {

    std::string from, to;

    std::string new_clone_file = clone_file;

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

        // change clone_file suffix from .binlog to .dat

        new_clone_file.replace(clone_file.size() - 7, 7, ".dat");

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

        // change clone_file suffix from .binlog-index to .idx

        new_clone_file.replace(clone_file.size() - 13, 13, ".idx");

    }

    from = fmt::format("{}/{}", clone_dir, clone_file);

    to = fmt::format("{}/_binlog/{}", tablet_dir, new_clone_file);

    // check to to file exist

    bool exists = true;

    auto status = io::global_local_filesystem()->exists(to, &exists);

    if (!status.ok()) {

        return ResultError(std::move(status));

    }

    if (!exists) {

        return to;

    }

    LOG(WARNING) << "binlog file already exist. "

                 << "tablet_dir=" << tablet_dir << ", clone_file=" << from << ", to=" << to;

    std::string clone_file_md5sum;

    status = io::global_local_filesystem()->md5sum(from, &clone_file_md5sum);

    if (!status.ok()) {

        return ResultError(std::move(status));

    }

    std::string to_file_md5sum;

    status = io::global_local_filesystem()->md5sum(to, &to_file_md5sum);

    if (!status.ok()) {

        return ResultError(std::move(status));

    }

    if (clone_file_md5sum == to_file_md5sum) {

        // if md5sum equal, then skip link file

        *skip_link_file = true;

        return to;

    } else {

        auto err_msg = fmt::format(

                "binlog file already exist, but md5sum not equal. "

                "tablet_dir={}, clone_file={}",

                tablet_dir, clone_file);

        LOG(WARNING) << err_msg;

        return ResultError(Status::InternalError(std::move(err_msg)));

    }

}

} // namespace

#define RETURN_IF_ERROR_(status, stmt) \

    do {                               \

        status = (stmt);               \

        if (UNLIKELY(!status.ok())) {  \

            return status;             \

        }                              \

    } while (false)

EngineCloneTask::EngineCloneTask(StorageEngine& engine, const TCloneReq& clone_req,

                                 const ClusterInfo* cluster_info, int64_t signature,

                                 std::vector<TTabletInfo>* tablet_infos)

        : _engine(engine),

          _clone_req(clone_req),

          _tablet_infos(tablet_infos),

          _signature(signature),

          _cluster_info(cluster_info) {

    _mem_tracker = MemTrackerLimiter::create_shared(

            MemTrackerLimiter::Type::OTHER,

            "EngineCloneTask#tabletId=" + std::to_string(_clone_req.tablet_id));

}

Status EngineCloneTask::execute() {

    // register the tablet to avoid it is deleted by gc thread during clone process

    Status st = _do_clone();

    _engine.tablet_manager()->update_partitions_visible_version(

            {{_clone_req.partition_id, _clone_req.version}});

    return st;

}

Status EngineCloneTask::_do_clone() {

    DBUG_EXECUTE_IF("EngineCloneTask.wait_clone", {

        auto duration = std::chrono::milliseconds(dp->param("duration", 10 * 1000));

        std::this_thread::sleep_for(duration);

    });

    DBUG_EXECUTE_IF("EngineCloneTask.failed_clone", {

        LOG_WARNING("EngineCloneTask.failed_clone")

                .tag("tablet_id", _clone_req.tablet_id)

                .tag("replica_id", _clone_req.replica_id)

                .tag("version", _clone_req.version);

        return Status::InternalError(

                "in debug point, EngineCloneTask.failed_clone tablet={}, replica={}, version={}",

                _clone_req.tablet_id, _clone_req.replica_id, _clone_req.version);

    });

    Status status = Status::OK();

    std::string src_file_path;

    TBackend src_host;

    int32_t copy_type =

            _clone_req.__isset.copy_type ? _clone_req.copy_type : TabletCopyType::DEFAULT;

    RETURN_IF_ERROR(TabletCopyType::validate(copy_type));

    bool copy_row_binlog = TabletCopyType::has(copy_type, TTabletCopyType::ROW_BINLOG);

    RETURN_IF_ERROR(

            _engine.tablet_manager()->register_transition_tablet(_clone_req.tablet_id, "clone"));

    Defer defer {[&]() {

        _engine.tablet_manager()->unregister_transition_tablet(_clone_req.tablet_id, "clone");

    }};

    // Check local tablet exist or not

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

    // The status of a tablet is not ready, indicating that it is a residual tablet after a schema

    // change failure. Clone a new tablet from remote be to overwrite it. This situation basically only

    // occurs when the be_rebalancer_fuzzy_test configuration is enabled.

    if (tablet && tablet->tablet_state() == TABLET_NOTREADY) {

        LOG(WARNING) << "tablet state is not ready when clone, need to drop old tablet, tablet_id="

                     << tablet->tablet_id();

        RETURN_IF_ERROR(_engine.tablet_manager()->drop_tablet(tablet->tablet_id(),

                                                              tablet->replica_id(), false));

        tablet.reset();

    }

    _is_new_tablet = tablet == nullptr;

    // try to incremental clone

    Versions missed_versions;

    // try to repair a tablet with missing version

    if (tablet != nullptr) {

        std::shared_lock migration_rlock(tablet->get_migration_lock(), std::try_to_lock);

        if (!migration_rlock.owns_lock()) {

            return Status::Error<TRY_LOCK_FAILED>(

                    "EngineCloneTask::_do_clone meet try lock failed");

        }

        if (tablet->replica_id() < _clone_req.replica_id) {

            // `tablet` may be a dropped replica in FE, e.g:

            //   BE1 migrates replica of tablet_1 to BE2, but before BE1 drop this replica, another new replica of tablet_1 is migrated to BE1.

            // Clone can still continue in this case. But to keep `replica_id` consitent with FE, MUST reset `replica_id` with request `replica_id`.

            tablet->tablet_meta()->set_replica_id(_clone_req.replica_id);

        }

        // get download path

        auto local_data_path = fmt::format("{}/{}", tablet->tablet_path(), CLONE_PREFIX);

        bool allow_incremental_clone = false;

        int64_t specified_version = _clone_req.version;

        if (tablet->enable_unique_key_merge_on_write()) {

            int64_t min_pending_ver = _engine.get_pending_publish_min_version(tablet->tablet_id());

            if (min_pending_ver - 1 < specified_version) {

                LOG(INFO) << "use min pending publish version for clone, min_pending_ver: "

                          << min_pending_ver << " visible_version: " << _clone_req.version;

                specified_version = min_pending_ver - 1;

            }

        }

        missed_versions = tablet->get_missed_versions(specified_version);

        // if missed version size is 0, then it is useless to clone from remote be, it means local data is

        // completed. Or remote be will just return header not the rowset files. clone will failed.

        if (missed_versions.empty()) {

            LOG(INFO) << "missed version size = 0, skip clone and return success. tablet_id="

                      << _clone_req.tablet_id << " replica_id=" << _clone_req.replica_id;

            RETURN_IF_ERROR(_set_tablet_info());

            return Status::OK();

        }

        LOG(INFO) << "clone to existed tablet. missed_versions_size=" << missed_versions.size()

                  << ", allow_incremental_clone=" << allow_incremental_clone

                  << ", signature=" << _signature << ", tablet_id=" << _clone_req.tablet_id

                  << ", visible_version=" << _clone_req.version

                  << ", replica_id=" << _clone_req.replica_id;

        // try to download missing version from src backend.

        // if tablet on src backend does not contains missing version, it will download all versions,

        // and set allow_incremental_clone to false

        RETURN_IF_ERROR(_make_and_download_snapshots(*(tablet->data_dir()), local_data_path,

                                                     &src_host, &src_file_path, missed_versions,

                                                     &allow_incremental_clone));

        RETURN_IF_ERROR(_finish_clone(tablet.get(), local_data_path, specified_version,

                                      allow_incremental_clone, copy_row_binlog));

    } else {

        LOG(INFO) << "clone tablet not exist, begin clone a new tablet from remote be. "

                  << "signature=" << _signature << ", tablet_id=" << _clone_req.tablet_id

                  << ", visible_version=" << _clone_req.version

                  << ", req replica=" << _clone_req.replica_id;

        // create a new tablet in this be

        // Get local disk from olap

        std::string local_shard_root_path;

        DataDir* store = nullptr;

        RETURN_IF_ERROR(_engine.obtain_shard_path(_clone_req.storage_medium,

                                                  _clone_req.dest_path_hash, &local_shard_root_path,

                                                  &store, _clone_req.partition_id));

        auto tablet_dir = fmt::format("{}/{}/{}", local_shard_root_path, _clone_req.tablet_id,

                                      _clone_req.schema_hash);

        Defer remove_useless_dir {[&] {

            if (status.ok()) {

                return;

            }

            LOG(INFO) << "clone failed. want to delete local dir: " << tablet_dir

                      << ". signature: " << _signature;

            WARN_IF_ERROR(io::global_local_filesystem()->delete_directory(tablet_dir),

                          "failed to delete useless clone dir ");

            WARN_IF_ERROR(DataDir::delete_tablet_parent_path_if_empty(tablet_dir),

                          "failed to delete parent dir");

        }};

        bool exists = true;

        Status exists_st = io::global_local_filesystem()->exists(tablet_dir, &exists);

        if (!exists_st) {

            LOG(WARNING) << "cant get path=" << tablet_dir << " state, st=" << exists_st;

            return exists_st;

        }

        if (exists) {

            LOG(WARNING) << "before clone dest path=" << tablet_dir << " exist, remove it first";

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

        }

        bool allow_incremental_clone = false;

        RETURN_IF_ERROR_(status,

                         _make_and_download_snapshots(*store, tablet_dir, &src_host, &src_file_path,

                                                      missed_versions, &allow_incremental_clone));

        LOG(INFO) << "clone copy done. src_host: " << src_host.host

                  << " src_file_path: " << src_file_path;

        auto tablet_manager = _engine.tablet_manager();

        RETURN_IF_ERROR_(status, tablet_manager->load_tablet_from_dir(store, _clone_req.tablet_id,

                                                                      _clone_req.schema_hash,

                                                                      tablet_dir, false));

        auto nested_tablet = tablet_manager->get_tablet(_clone_req.tablet_id);

        if (!nested_tablet) {

            status = Status::NotFound("tablet not found, tablet_id={}", _clone_req.tablet_id);

            return status;

        }

        // MUST reset `replica_id` to request `replica_id` to keep consistent with FE

        nested_tablet->tablet_meta()->set_replica_id(_clone_req.replica_id);

        // clone success, delete .hdr file because tablet meta is stored in rocksdb

        std::string header_path =

                TabletMeta::construct_header_file_path(tablet_dir, _clone_req.tablet_id);

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

    }

    return _set_tablet_info();

}

Status EngineCloneTask::_set_tablet_info() {

    // Get clone tablet info

    TTabletInfo tablet_info;

    tablet_info.__set_tablet_id(_clone_req.tablet_id);

    tablet_info.__set_replica_id(_clone_req.replica_id);

    tablet_info.__set_schema_hash(_clone_req.schema_hash);

    RETURN_IF_ERROR(_engine.tablet_manager()->report_tablet_info(&tablet_info));

    if (_clone_req.__isset.version && tablet_info.version < _clone_req.version) {

        // if it is a new tablet and clone failed, then remove the tablet

        // if it is incremental clone, then must not drop the tablet

        if (_is_new_tablet) {

            // we need to check if this cloned table's version is what we expect.

            // if not, maybe this is a stale remaining table which is waiting for drop.

            // we drop it.

            LOG(WARNING) << "begin to drop the stale tablet. tablet_id:" << _clone_req.tablet_id

                         << ", replica_id:" << _clone_req.replica_id

                         << ", schema_hash:" << _clone_req.schema_hash

                         << ", signature:" << _signature << ", version:" << tablet_info.version

                         << ", expected_version: " << _clone_req.version;

            WARN_IF_ERROR(_engine.tablet_manager()->drop_tablet(_clone_req.tablet_id,

                                                                _clone_req.replica_id, false),

                          "drop stale cloned table failed");

        }

        return Status::InternalError("unexpected version. tablet version: {}, expected version: {}",

                                     tablet_info.version, _clone_req.version);

    }

    LOG(INFO) << "clone get tablet info success. tablet_id:" << _clone_req.tablet_id

              << ", schema_hash:" << _clone_req.schema_hash << ", signature:" << _signature

              << ", replica id:" << _clone_req.replica_id << ", version:" << tablet_info.version;

    _tablet_infos->push_back(tablet_info);

    return Status::OK();

}

/// This method will do following things:

/// 1. Make snapshots on source BE.

/// 2. Download all snapshots to CLONE dir.

/// 3. Convert rowset ids of downloaded snapshots(would also change the replica id).

/// 4. Release the snapshots on source BE.

Status EngineCloneTask::_make_and_download_snapshots(DataDir& data_dir,

                                                     const std::string& local_data_path,

                                                     TBackend* src_host, std::string* snapshot_path,

                                                     const std::vector<Version>& missed_versions,

                                                     bool* allow_incremental_clone) {

    Status status;

    const auto& token = _cluster_info->token;

    int timeout_s = 0;

    if (_clone_req.__isset.timeout_s) {

        timeout_s = _clone_req.timeout_s;

    }

    for (auto&& src : _clone_req.src_backends) {

        // Make snapshot in remote olap engine

        *src_host = src;

        // make snapshot

        status = _make_snapshot(src.host, src.be_port, _clone_req.tablet_id, _clone_req.schema_hash,

                                timeout_s, missed_versions, snapshot_path, allow_incremental_clone);

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

            LOG_WARNING("failed to make snapshot in remote BE")

                    .tag("host", src.host)

                    .tag("port", src.be_port)

                    .tag("tablet", _clone_req.tablet_id)

                    .tag("signature", _signature)

                    .tag("missed_versions", missed_versions)

                    .error(status);

            continue; // Try another BE

        }

        LOG_INFO("successfully make snapshot in remote BE")

                .tag("host", src.host)

                .tag("port", src.be_port)

                .tag("tablet", _clone_req.tablet_id)

                .tag("snapshot_path", *snapshot_path)

                .tag("signature", _signature)

                .tag("missed_versions", missed_versions);

        Defer defer {[host = src.host, port = src.be_port, &snapshot_path = *snapshot_path, this] {

            // TODO(plat1ko): Async release snapshot

            auto st = _release_snapshot(host, port, snapshot_path);

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

                LOG_WARNING("failed to release snapshot in remote BE")

                        .tag("host", host)

                        .tag("port", port)

                        .tag("snapshot_path", snapshot_path)

                        .error(st);

            }

        }};

        std::string remote_dir;

        {

            std::stringstream ss;

            if (snapshot_path->back() == '/') {

                ss << *snapshot_path << _clone_req.tablet_id << "/" << _clone_req.schema_hash

                   << "/";

            } else {

                ss << *snapshot_path << "/" << _clone_req.tablet_id << "/" << _clone_req.schema_hash

                   << "/";

            }

            remote_dir = ss.str();

        }

        std::string address = get_host_port(src.host, src.http_port);

        int32_t copy_type =

                _clone_req.__isset.copy_type ? _clone_req.copy_type : TabletCopyType::DEFAULT;

        RETURN_IF_ERROR(TabletCopyType::validate(copy_type));

        bool copy_row_binlog = TabletCopyType::has(copy_type, TTabletCopyType::ROW_BINLOG);

        if (config::enable_batch_download && is_support_batch_download(address).ok()) {

            // download files via batch api.

            LOG_INFO("remote BE supports batch download, use batch file download")

                    .tag("address", address)

                    .tag("remote_dir", remote_dir);

            status = _batch_download_files(&data_dir, address, remote_dir, local_data_path);

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

                LOG_WARNING("failed to download snapshot from remote BE in batch")

                        .tag("address", address)

                        .tag("remote_dir", remote_dir)

                        .error(status);

                continue; // Try another BE

            }

            if (copy_row_binlog) {

                std::string row_binlog_remote_dir =

                        fmt::format("{}{}/", remote_dir, FDRowBinlogSuffix);

                std::string row_binlog_local_path =

                        fmt::format("{}/{}", local_data_path, FDRowBinlogSuffix);

                status = _batch_download_files(&data_dir, address, row_binlog_remote_dir,

                                               row_binlog_local_path);

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

                    LOG_WARNING("failed to download row binlog snapshot from remote BE in batch")

                            .tag("address", address)

                            .tag("remote_dir", row_binlog_remote_dir)

                            .error(status);

                    continue; // Try another BE

                }

            }

        } else {

            if (config::enable_batch_download) {

                LOG_INFO("remote BE does not support batch download, use single file download")

                        .tag("address", address)

                        .tag("remote_dir", remote_dir);

            } else {

                LOG_INFO("batch download is disabled, use single file download")

                        .tag("address", address)

                        .tag("remote_dir", remote_dir);

            }

            std::string remote_url_prefix;

            {

                std::stringstream ss;

                ss << "http://" << address << HTTP_REQUEST_PREFIX << HTTP_REQUEST_TOKEN_PARAM

                   << token << HTTP_REQUEST_FILE_PARAM << remote_dir;

                remote_url_prefix = ss.str();

            }

            status = _download_files(&data_dir, remote_url_prefix, local_data_path);

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

                LOG_WARNING("failed to download snapshot from remote BE")

                        .tag("url", mask_token(remote_url_prefix))

                        .error(status);

                continue; // Try another BE

            }

            if (copy_row_binlog) {

                std::string row_binlog_remote_url_prefix;

                {

                    std::stringstream ss;

                    ss << "http://" << address << HTTP_REQUEST_PREFIX << HTTP_REQUEST_TOKEN_PARAM

                       << token << HTTP_REQUEST_FILE_PARAM << remote_dir << FDRowBinlogSuffix

                       << "/";

                    row_binlog_remote_url_prefix = ss.str();

                }

                std::string row_binlog_local_path =

                        fmt::format("{}/{}", local_data_path, FDRowBinlogSuffix);

                status = _download_files(&data_dir, row_binlog_remote_url_prefix,

                                         row_binlog_local_path);

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

                    LOG_WARNING("failed to download row binlog snapshot from remote BE")

                            .tag("url", mask_token(row_binlog_remote_url_prefix))

                            .error(status);

                    continue; // Try another BE

                }

            }

        }

        // No need to try again with another BE

        _pending_rs_guards = DORIS_TRY(_engine.snapshot_mgr()->convert_rowset_ids(

                local_data_path, _clone_req.tablet_id, _clone_req.replica_id, _clone_req.table_id,

                _clone_req.partition_id, _clone_req.schema_hash));

        break;

    } // clone copy from one backend

    return status;

}

Status EngineCloneTask::_make_snapshot(const std::string& ip, int port, TTableId tablet_id,

                                       TSchemaHash schema_hash, int timeout_s,

                                       const std::vector<Version>& missed_versions,

                                       std::string* snapshot_path, bool* allow_incremental_clone) {

    TSnapshotRequest request;

    request.__set_tablet_id(tablet_id);

    request.__set_schema_hash(schema_hash);

    request.__set_preferred_snapshot_version(g_Types_constants.TPREFER_SNAPSHOT_REQ_VERSION);

    request.__set_version(_clone_req.version);

    int32_t copy_type =

            _clone_req.__isset.copy_type ? _clone_req.copy_type : TabletCopyType::DEFAULT;

    RETURN_IF_ERROR(TabletCopyType::validate(copy_type));

    request.__set_copy_type(copy_type);

    request.__set_is_copy_binlog(TabletCopyType::has(copy_type, TTabletCopyType::CCR_BINLOG));

    // TODO: missing version composed of singleton delta.

    // if not, this place should be rewrote.

    // we make every TSnapshotRequest sent from be with __isset.missing_version = true

    // then if one be received one req with __isset.missing_version = false it means

    // this req is sent from FE(FE would never set this field)

    request.__isset.missing_version = true;

    for (auto& version : missed_versions) {

        request.missing_version.push_back(version.first);

    }

    if (timeout_s > 0) {

        request.__set_timeout(timeout_s);

    }

    TAgentResult result;

    RETURN_IF_ERROR(ThriftRpcHelper::rpc<BackendServiceClient>(

            ip, port, [&request, &result](BackendServiceConnection& client) {

                client->make_snapshot(result, request);

            }));

    if (result.status.status_code != TStatusCode::OK) {

        return Status::create(result.status);

    }

    if (!result.__isset.snapshot_path) {

        return Status::InternalError("success snapshot request without snapshot path");

    }

    *snapshot_path = result.snapshot_path;

    if (snapshot_path->at(snapshot_path->length() - 1) != '/') {

        snapshot_path->append("/");

    }

    if (result.__isset.allow_incremental_clone) {

        // During upgrading, some BE nodes still be installed an old previous old.

        // which incremental clone is not ready in those nodes.

        // should add a symbol to indicate it.

        *allow_incremental_clone = result.allow_incremental_clone;

    }

    return Status::OK();

}

Status EngineCloneTask::_release_snapshot(const std::string& ip, int port,

                                          const std::string& snapshot_path) {

    TAgentResult result;

    RETURN_IF_ERROR(ThriftRpcHelper::rpc<BackendServiceClient>(

            ip, port, [&snapshot_path, &result](BackendServiceConnection& client) {

                client->release_snapshot(result, snapshot_path);

            }));

    return Status::create(result.status);

}

Status EngineCloneTask::_download_files(DataDir* data_dir, const std::string& remote_url_prefix,

                                        const std::string& local_path) {

    // Check local path exist, if exist, remove it, then create the dir

    // local_file_full_path = tabletid/clone， for a specific tablet, there should be only one folder

    // if this folder exists, then should remove it

    // for example, BE clone from BE 1 to download file 1 with version (2,2), but clone from BE 1 failed

    // then it will try to clone from BE 2, but it will find the file 1 already exist, but file 1 with same

    // name may have different versions.

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

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

    // Get remote dir file list

    std::string file_list_str;

    auto list_files_cb = [&remote_url_prefix, &file_list_str](HttpClient* client) {

        RETURN_IF_ERROR(client->init(remote_url_prefix));

        client->set_timeout_ms(LIST_REMOTE_FILE_TIMEOUT * 1000);

        return client->execute(&file_list_str);

    };

    RETURN_IF_ERROR(HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, list_files_cb));

    std::vector<std::string> file_name_list =

            absl::StrSplit(file_list_str, "\n", absl::SkipWhitespace());

    // If the header file is not exist, the table couldn't loaded by olap engine.

    // Avoid of data is not complete, we copy the header file at last.

    // The header file's name is end of .hdr.

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

        if (file_name_list[i].ends_with(".hdr")) {

            std::swap(file_name_list[i], file_name_list[file_name_list.size() - 1]);

            break;

        }

    }

    // Get copy from remote

    uint64_t total_file_size = 0;

    MonotonicStopWatch watch;

    watch.start();

    for (auto& file_name : file_name_list) {

        auto remote_file_url = remote_url_prefix + file_name;

        // get file length

        uint64_t file_size = 0;

        auto get_file_size_cb = [&remote_file_url, &file_size](HttpClient* client) {

            RETURN_IF_ERROR(client->init(remote_file_url));

            client->set_timeout_ms(GET_LENGTH_TIMEOUT * 1000);

            RETURN_IF_ERROR(client->head());

            RETURN_IF_ERROR(client->get_content_length(&file_size));

            return Status::OK();

        };

        RETURN_IF_ERROR(

                HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, get_file_size_cb));

        // check disk capacity

        if (data_dir->reach_capacity_limit(file_size)) {

            return Status::Error<EXCEEDED_LIMIT>(

                    "reach the capacity limit of path {}, file_size={}", data_dir->path(),

                    file_size);

        }

        total_file_size += file_size;

        uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024;

        if (estimate_timeout < config::download_low_speed_time) {

            estimate_timeout = config::download_low_speed_time;

        }

        std::string local_file_path = local_path + "/" + file_name;

        LOG(INFO) << "clone begin to download file from: " << mask_token(remote_file_url)

                  << " to: " << local_file_path << ". size(B): " << file_size

                  << ", timeout(s): " << estimate_timeout;

        auto download_cb = [&remote_file_url, estimate_timeout, &local_file_path,

                            file_size](HttpClient* client) {

            RETURN_IF_ERROR(client->init(remote_file_url));

            client->set_timeout_ms(estimate_timeout * 1000);

            RETURN_IF_ERROR(client->download(local_file_path));

            std::error_code ec;

            // Check file length

            uint64_t local_file_size = std::filesystem::file_size(local_file_path, ec);

            if (ec) {

                LOG(WARNING) << "download file error" << ec.message();

                return Status::IOError("can't retrive file_size of {}, due to {}", local_file_path,

                                       ec.message());

            }

            if (local_file_size != file_size) {

                LOG(WARNING) << "download file length error"

                             << ", remote_path=" << mask_token(remote_file_url)

                             << ", file_size=" << file_size

                             << ", local_file_size=" << local_file_size;

                return Status::InternalError("downloaded file size is not equal");

            }

            return io::global_local_filesystem()->permission(local_file_path,

                                                             io::LocalFileSystem::PERMS_OWNER_RW);

        };

        RETURN_IF_ERROR(HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, download_cb));

    } // Clone files from remote backend

    uint64_t total_time_ms = watch.elapsed_time() / 1000 / 1000;

    total_time_ms = total_time_ms > 0 ? total_time_ms : 0;

    double copy_rate = 0.0;

    if (total_time_ms > 0) {

        copy_rate = total_file_size / ((double)total_time_ms) / 1000;

    }

    _copy_size = (int64_t)total_file_size;

    _copy_time_ms = (int64_t)total_time_ms;

    LOG(INFO) << "succeed to copy tablet " << _signature

              << ", total files: " << file_name_list.size()

              << ", total file size: " << total_file_size << " B, cost: " << total_time_ms << " ms"

              << ", rate: " << copy_rate << " MB/s";

    return Status::OK();

}

Status EngineCloneTask::_batch_download_files(DataDir* data_dir, const std::string& address,

                                              const std::string& remote_dir,

                                              const std::string& local_dir) {

    constexpr size_t BATCH_FILE_SIZE = 64 << 20; // 64MB

    constexpr size_t BATCH_FILE_NUM = 64;

    // Check local path exist, if exist, remove it, then create the dir

    // local_file_full_path = tabletid/clone， for a specific tablet, there should be only one folder

    // if this folder exists, then should remove it

    // for example, BE clone from BE 1 to download file 1 with version (2,2), but clone from BE 1 failed

    // then it will try to clone from BE 2, but it will find the file 1 already exist, but file 1 with same

    // name may have different versions.

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

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

    const std::string& token = _cluster_info->token;

    std::vector<std::pair<std::string, size_t>> file_info_list;

    RETURN_IF_ERROR(list_remote_files_v2(address, token, remote_dir, &file_info_list));

    // If the header file is not exist, the table couldn't loaded by olap engine.

    // Avoid of data is not complete, we copy the header file at last.

    // The header file's name is end of .hdr.

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

        if (file_info_list[i].first.ends_with(".hdr")) {

            std::swap(file_info_list[i], file_info_list[file_info_list.size() - 1]);

            break;

        }

    }

    MonotonicStopWatch watch;

    watch.start();

    size_t total_file_size = 0;

    size_t total_files = file_info_list.size();

    std::vector<std::pair<std::string, size_t>> batch_files;

    for (size_t i = 0; i < total_files;) {

        size_t batch_file_size = 0;

        for (size_t j = i; j < total_files; j++) {

            // Split batchs by file number and file size,

            if (BATCH_FILE_NUM <= batch_files.size() || BATCH_FILE_SIZE <= batch_file_size ||

                // ... or separate the last .hdr file into a single batch.

                (j + 1 == total_files && !batch_files.empty())) {

                break;

            }

            batch_files.push_back(file_info_list[j]);

            batch_file_size += file_info_list[j].second;

        }

        // check disk capacity

        if (data_dir->reach_capacity_limit(batch_file_size)) {

            return Status::Error<EXCEEDED_LIMIT>(

                    "reach the capacity limit of path {}, file_size={}", data_dir->path(),

                    batch_file_size);

        }

        RETURN_IF_ERROR(download_files_v2(address, token, remote_dir, local_dir, batch_files));

        total_file_size += batch_file_size;

        i += batch_files.size();

        batch_files.clear();

    }

    uint64_t total_time_ms = watch.elapsed_time() / 1000 / 1000;

    total_time_ms = total_time_ms > 0 ? total_time_ms : 0;

    double copy_rate = 0.0;

    if (total_time_ms > 0) {

        copy_rate = total_file_size / ((double)total_time_ms) / 1000;

    }

    _copy_size = (int64_t)total_file_size;

    _copy_time_ms = (int64_t)total_time_ms;

    LOG(INFO) << "succeed to copy tablet " << _signature

              << ", total files: " << file_info_list.size()

              << ", total file size: " << total_file_size << " B, cost: " << total_time_ms << " ms"

              << ", rate: " << copy_rate << " MB/s";

    return Status::OK();

}

/// This method will only be called if tablet already exist in this BE when doing clone.

/// This method will do the following things:

/// 1. Link all files from CLONE dir to tablet dir if file does not exist in tablet dir

/// 2. Call _finish_xx_clone() to revise the tablet meta.

Status EngineCloneTask::_finish_clone(Tablet* tablet, const std::string& clone_dir, int64_t version,

                                      bool is_incremental_clone, bool copy_row_binlog) {

    Defer remove_clone_dir {[&]() {

        std::error_code ec;

        std::filesystem::remove_all(clone_dir, ec);

        if (ec) {

            LOG(WARNING) << "failed to remove=" << clone_dir << " msg=" << ec.message();

        }

    }};

    // check clone dir existed

    bool exists = true;

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

    if (!exists) {

        return Status::InternalError("clone dir not existed. clone_dir={}", clone_dir);

    }

    // Load src header.

    // The tablet meta info is downloaded from source BE as .hdr file.

    // So we load it and generate cloned_tablet_meta.

    auto cloned_tablet_meta_file = fmt::format("{}/{}.hdr", clone_dir, tablet->tablet_id());

    auto cloned_tablet_meta = std::make_shared<TabletMeta>();

    RETURN_IF_ERROR(cloned_tablet_meta->create_from_file(cloned_tablet_meta_file));

    // remove the cloned meta file

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

    // remove rowset binlog metas

    const auto& tablet_dir = tablet->tablet_path();

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

    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) {

        std::error_code ec;

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

        if (ec) {

            LOG(WARNING) << "get file size error" << ec.message();

            return Status::IOError("can't retrive file_size of {}, due to {}", binlog_metas_file,

                                   ec.message());

        }

        if (binlog_meta_filesize > 0) {

            contain_binlog = true;

            RETURN_IF_ERROR(read_pb(binlog_metas_file, &rowset_binlog_metas_pb));

        }

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

    }

    if (contain_binlog) {

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

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

    }

    enum class CloneFileType { DATA, ROW_BINLOG, CCR_BINLOG };

    // check all files in /clone and /tablet

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

    RETURN_IF_ERROR(io::global_local_filesystem()->list(clone_dir, true, &clone_files, &exists));

    std::vector<std::pair<CloneFileType, std::string>> clone_file_names;

    for (auto& file : clone_files) {

        CloneFileType file_type = CloneFileType::DATA;

        if (file.file_name.ends_with(".binlog") || file.file_name.ends_with(".binlog-index")) {

            file_type = CloneFileType::CCR_BINLOG;

        }

        clone_file_names.emplace_back(file_type, file.file_name);

    }

    auto row_binlog_clone_dir = fmt::format("{}/{}", clone_dir, FDRowBinlogSuffix);

    if (copy_row_binlog) {

        clone_files.clear();

        RETURN_IF_ERROR(io::global_local_filesystem()->list(row_binlog_clone_dir, true,

                                                            &clone_files, &exists));

        if (!exists) {

            return Status::InternalError("row binlog clone dir not existed. clone_dir={}",

                                         row_binlog_clone_dir);

        }

        for (auto& file : clone_files) {

            clone_file_names.emplace_back(CloneFileType::ROW_BINLOG, file.file_name);

        }

    }

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

    RETURN_IF_ERROR(io::global_local_filesystem()->list(tablet_dir, true, &local_files, &exists));

    std::unordered_set<std::string> data_local_file_names;

    for (auto& file : local_files) {

        data_local_file_names.insert(file.file_name);

    }

    auto row_binlog_dir = tablet->row_binlog_path();

    std::unordered_set<std::string> row_binlog_local_file_names;

    if (copy_row_binlog) {

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

        local_files.clear();

        RETURN_IF_ERROR(

                io::global_local_filesystem()->list(row_binlog_dir, true, &local_files, &exists));

        for (auto& file : local_files) {

            row_binlog_local_file_names.insert(file.file_name);

        }

    }

    Status status;

    std::vector<std::string> linked_success_files;

    Defer remove_linked_files {[&]() { // clear linked files if errors happen

        if (!status.ok()) {

            std::vector<io::Path> paths;