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

#include <absl/strings/str_split.h>

#include <arrow/record_batch.h>

#include <fmt/format.h>

#include <gen_cpp/BackendService.h>

#include <gen_cpp/BackendService_types.h>

#include <gen_cpp/Data_types.h>

#include <gen_cpp/DorisExternalService_types.h>

#include <gen_cpp/FrontendService_types.h>

#include <gen_cpp/Metrics_types.h>

#include <gen_cpp/PaloInternalService_types.h>

#include <gen_cpp/Planner_types.h>

#include <gen_cpp/Status_types.h>

#include <gen_cpp/Types_types.h>

#include <sys/types.h>

#include <thrift/concurrency/ThreadFactory.h>

#include <thrift/protocol/TDebugProtocol.h>

#include <time.h>

#include <cstdint>

#include <map>

#include <memory>

#include <ostream>

#include <ranges>

#include <string>

#include <thread>

#include <utility>

#include <vector>

#include "absl/strings/substitute.h"

#include "cloud/config.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "exprs/function/dictionary_factory.h"

#include "format/arrow/arrow_row_batch.h"

#include "io/fs/connectivity/storage_connectivity_tester.h"

#include "io/fs/local_file_system.h"

#include "load/routine_load/routine_load_task_executor.h"

#include "load/stream_load/stream_load_context.h"

#include "load/stream_load/stream_load_recorder.h"

#include "runtime/exec_env.h"

#include "runtime/external_scan_context_mgr.h"

#include "runtime/fragment_mgr.h"

#include "runtime/result_queue_mgr.h"

#include "runtime/runtime_profile.h"

#include "service/http/http_client.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowset/beta_rowset.h"

#include "storage/rowset/pending_rowset_helper.h"

#include "storage/rowset/rowset_factory.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/snapshot/snapshot_manager.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_manager.h"

#include "storage/tablet/tablet_meta.h"

#include "storage/txn/txn_manager.h"

#include "udf/python/python_env.h"

#include "util/defer_op.h"

#include "util/threadpool.h"

#include "util/thrift_server.h"

#include "util/uid_util.h"

#include "util/url_coding.h"

namespace apache {

namespace thrift {

class TException;

class TMultiplexedProcessor;

class TProcessor;

namespace transport {

class TTransportException;

} // namespace transport

} // namespace thrift

} // namespace apache

namespace doris {

namespace {

bvar::LatencyRecorder g_ingest_binlog_latency("doris_backend_service", "ingest_binlog");

struct IngestBinlogArg {

    int64_t txn_id;

    int64_t partition_id;

    int64_t local_tablet_id;

    TabletSharedPtr local_tablet;

    TIngestBinlogRequest request;

    TStatus* tstatus;

};

Status _exec_http_req(std::optional<HttpClient>& client, int retry_times, int sleep_time,

                      const std::function<Status(HttpClient*)>& callback) {

    if (client.has_value()) {

        return client->execute(retry_times, sleep_time, callback);

    } else {

        return HttpClient::execute_with_retry(retry_times, sleep_time, callback);

    }

}

Status _download_binlog_segment_file(HttpClient* client, const std::string& get_segment_file_url,

                                     const std::string& segment_path, uint64_t segment_file_size,

                                     uint64_t estimate_timeout,

                                     std::vector<std::string>& download_success_files) {

    RETURN_IF_ERROR(client->init(get_segment_file_url));

    client->set_timeout_ms(estimate_timeout * 1000);

    RETURN_IF_ERROR(client->download(segment_path));

    download_success_files.push_back(segment_path);

    std::string remote_file_md5;

    RETURN_IF_ERROR(client->get_content_md5(&remote_file_md5));

    LOG(INFO) << "download segment file to " << segment_path << ", remote md5: " << remote_file_md5

              << ", remote size: " << segment_file_size;

    std::error_code ec;

    // Check file length

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

    if (ec) {

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

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

                               ec.message());

    }

    if (local_file_size != segment_file_size) {

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

                     << ", get_segment_file_url=" << get_segment_file_url

                     << ", file_size=" << segment_file_size

                     << ", local_file_size=" << local_file_size;

        return Status::RuntimeError("downloaded file size is not equal, local={}, remote={}",

                                    local_file_size, segment_file_size);

    }

    if (!remote_file_md5.empty()) { // keep compatibility

        std::string local_file_md5;

        RETURN_IF_ERROR(io::global_local_filesystem()->md5sum(segment_path, &local_file_md5));

        if (local_file_md5 != remote_file_md5) {

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

                         << ", get_segment_file_url=" << get_segment_file_url

                         << ", remote_file_md5=" << remote_file_md5

                         << ", local_file_md5=" << local_file_md5;

            return Status::RuntimeError("download file md5 is not equal, local={}, remote={}",

                                        local_file_md5, remote_file_md5);

        }

    }

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

                                                     io::LocalFileSystem::PERMS_OWNER_RW);

}

Status _download_binlog_index_file(HttpClient* client,

                                   const std::string& get_segment_index_file_url,

                                   const std::string& local_segment_index_path,

                                   uint64_t segment_index_file_size, uint64_t estimate_timeout,

                                   std::vector<std::string>& download_success_files) {

    RETURN_IF_ERROR(client->init(get_segment_index_file_url));

    client->set_timeout_ms(estimate_timeout * 1000);

    RETURN_IF_ERROR(client->download(local_segment_index_path));

    download_success_files.push_back(local_segment_index_path);

    std::string remote_file_md5;

    RETURN_IF_ERROR(client->get_content_md5(&remote_file_md5));

    LOG(INFO) << "download segment index file to " << local_segment_index_path

              << ", remote md5: " << remote_file_md5

              << ", remote size: " << segment_index_file_size;

    std::error_code ec;

    // Check file length

    uint64_t local_index_file_size = std::filesystem::file_size(local_segment_index_path, ec);

    if (ec) {

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

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

                               ec.message());

    }

    if (local_index_file_size != segment_index_file_size) {

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

                     << ", get_segment_index_file_url=" << get_segment_index_file_url

                     << ", index_file_size=" << segment_index_file_size

                     << ", local_index_file_size=" << local_index_file_size;

        return Status::RuntimeError("downloaded index file size is not equal, local={}, remote={}",

                                    local_index_file_size, segment_index_file_size);

    }

    if (!remote_file_md5.empty()) { // keep compatibility

        std::string local_file_md5;

        RETURN_IF_ERROR(

                io::global_local_filesystem()->md5sum(local_segment_index_path, &local_file_md5));

        if (local_file_md5 != remote_file_md5) {

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

                         << ", get_segment_index_file_url=" << get_segment_index_file_url

                         << ", remote_file_md5=" << remote_file_md5

                         << ", local_file_md5=" << local_file_md5;

            return Status::RuntimeError("download file md5 is not equal, local={}, remote={}",

                                        local_file_md5, remote_file_md5);

        }

    }

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

                                                     io::LocalFileSystem::PERMS_OWNER_RW);

}

void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) {

    std::optional<HttpClient> client;

    if (config::enable_ingest_binlog_with_persistent_connection) {

        // Save the http client instance for persistent connection

        client = std::make_optional<HttpClient>();

    }

    auto txn_id = arg->txn_id;

    auto partition_id = arg->partition_id;

    auto local_tablet_id = arg->local_tablet_id;

    const auto& local_tablet = arg->local_tablet;

    const auto& local_tablet_uid = local_tablet->tablet_uid();

    std::shared_ptr<MemTrackerLimiter> mem_tracker = MemTrackerLimiter::create_shared(

            MemTrackerLimiter::Type::OTHER, fmt::format("IngestBinlog#TxnId={}", txn_id));

    SCOPED_ATTACH_TASK(mem_tracker);

    auto& request = arg->request;

    MonotonicStopWatch watch(true);

    int64_t total_download_bytes = 0;

    int64_t total_download_files = 0;

    TStatus tstatus;

    std::vector<std::string> download_success_files;

    std::unordered_map<std::string_view, uint64_t> elapsed_time_map;

    Defer defer {[=, &engine, &tstatus, ingest_binlog_tstatus = arg->tstatus, &watch,

                  &total_download_bytes, &total_download_files, &elapsed_time_map]() {

        g_ingest_binlog_latency << watch.elapsed_time_microseconds();

        auto elapsed_time_ms = static_cast<int64_t>(watch.elapsed_time_milliseconds());

        double copy_rate = 0.0;

        if (elapsed_time_ms > 0) {

            copy_rate = (double)total_download_bytes / ((double)elapsed_time_ms) / 1000;

        }

        LOG(INFO) << "ingest binlog elapsed " << elapsed_time_ms << " ms, download "

                  << total_download_files << " files, total " << total_download_bytes

                  << " bytes, avg rate " << copy_rate

                  << " MB/s. result: " << apache::thrift::ThriftDebugString(tstatus);

        if (config::ingest_binlog_elapsed_threshold_ms >= 0 &&

            elapsed_time_ms > config::ingest_binlog_elapsed_threshold_ms) {

            auto elapsed_details_view =

                    elapsed_time_map | std::views::transform([](const auto& pair) {

                        return fmt::format("{}:{}", pair.first, pair.second);

                    });

            std::string elapsed_details = fmt::format("{}", fmt::join(elapsed_details_view, ", "));

            LOG(WARNING) << "ingest binlog elapsed " << elapsed_time_ms << " ms, "

                         << elapsed_details;

        }

        if (tstatus.status_code != TStatusCode::OK) {

            // abort txn

            engine.txn_manager()->abort_txn(partition_id, txn_id, local_tablet_id,

                                            local_tablet_uid);

            // delete all successfully downloaded files

            LOG(WARNING) << "will delete downloaded success files due to error " << tstatus;

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

            for (const auto& file : download_success_files) {

                paths.emplace_back(file);

                LOG(WARNING) << "will delete downloaded success file " << file << " due to error";

            }

            static_cast<void>(io::global_local_filesystem()->batch_delete(paths));

            LOG(WARNING) << "done delete downloaded success files due to error " << tstatus;

        }

        if (ingest_binlog_tstatus) {

            *ingest_binlog_tstatus = std::move(tstatus);

        }

    }};

    auto set_tstatus = [&tstatus](TStatusCode::type code, std::string error_msg) {

        tstatus.__set_status_code(code);

        tstatus.__isset.error_msgs = true;

        tstatus.error_msgs.push_back(std::move(error_msg));

    };

    auto estimate_download_timeout = [](int64_t 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;

        }

        return estimate_timeout;

    };

    // Step 3: get binlog info

    auto binlog_api_url = fmt::format("http://{}:{}/api/_binlog/_download", request.remote_host,

                                      request.remote_port);

    constexpr int max_retry = 3;

    auto get_binlog_info_url =

            fmt::format("{}?method={}&tablet_id={}&binlog_version={}", binlog_api_url,

                        "get_binlog_info", request.remote_tablet_id, request.binlog_version);

    std::string binlog_info;

    auto get_binlog_info_cb = [&get_binlog_info_url, &binlog_info](HttpClient* client) {

        RETURN_IF_ERROR(client->init(get_binlog_info_url));

        client->set_timeout_ms(config::download_binlog_meta_timeout_ms);

        return client->execute(&binlog_info);

    };

    auto status = _exec_http_req(client, max_retry, 1, get_binlog_info_cb);

    if (!status.ok()) {

        LOG(WARNING) << "failed to get binlog info from " << get_binlog_info_url

                     << ", status=" << status.to_string();

        status.to_thrift(&tstatus);

        return;

    }

    elapsed_time_map.emplace("get_binlog_info", watch.elapsed_time_microseconds());

    std::vector<std::string> binlog_info_parts = absl::StrSplit(binlog_info, ":");

    if (binlog_info_parts.size() != 2) {

        status = Status::RuntimeError("failed to parse binlog info into 2 parts: {}", binlog_info);

        LOG(WARNING) << "failed to get binlog info from " << get_binlog_info_url

                     << ", status=" << status.to_string();

        status.to_thrift(&tstatus);

        return;

    }

    std::string remote_rowset_id = std::move(binlog_info_parts[0]);

    int64_t num_segments = -1;

    try {

        num_segments = std::stoll(binlog_info_parts[1]);

    } catch (std::exception& e) {

        status = Status::RuntimeError("failed to parse num segments from binlog info {}: {}",

                                      binlog_info, e.what());

        LOG(WARNING) << "failed to get binlog info from " << get_binlog_info_url

                     << ", status=" << status;

        status.to_thrift(&tstatus);

        return;

    }

    // Step 4: get rowset meta

    auto get_rowset_meta_url = fmt::format(

            "{}?method={}&tablet_id={}&rowset_id={}&binlog_version={}", binlog_api_url,

            "get_rowset_meta", request.remote_tablet_id, remote_rowset_id, request.binlog_version);

    std::string rowset_meta_str;

    auto get_rowset_meta_cb = [&get_rowset_meta_url, &rowset_meta_str](HttpClient* client) {

        RETURN_IF_ERROR(client->init(get_rowset_meta_url));

        client->set_timeout_ms(config::download_binlog_meta_timeout_ms);

        return client->execute(&rowset_meta_str);

    };

    status = _exec_http_req(client, max_retry, 1, get_rowset_meta_cb);

    if (!status.ok()) {

        LOG(WARNING) << "failed to get rowset meta from " << get_rowset_meta_url

                     << ", status=" << status.to_string();

        status.to_thrift(&tstatus);

        return;

    }

    elapsed_time_map.emplace("get_rowset_meta", watch.elapsed_time_microseconds());

    RowsetMetaPB rowset_meta_pb;

    if (!rowset_meta_pb.ParseFromString(rowset_meta_str)) {

        LOG(WARNING) << "failed to parse rowset meta from " << get_rowset_meta_url;

        status = Status::InternalError("failed to parse rowset meta");

        status.to_thrift(&tstatus);

        return;

    }

    // save source rowset id and tablet id

    rowset_meta_pb.set_source_rowset_id(remote_rowset_id);

    rowset_meta_pb.set_source_tablet_id(request.remote_tablet_id);

    // rewrite rowset meta

    rowset_meta_pb.set_tablet_id(local_tablet_id);

    rowset_meta_pb.set_partition_id(partition_id);

    rowset_meta_pb.set_tablet_schema_hash(local_tablet->tablet_meta()->schema_hash());

    rowset_meta_pb.set_txn_id(txn_id);

    rowset_meta_pb.set_rowset_state(RowsetStatePB::COMMITTED);

    auto rowset_meta = std::make_shared<RowsetMeta>();

    if (!rowset_meta->init_from_pb(rowset_meta_pb)) {

        LOG(WARNING) << "failed to init rowset meta from " << get_rowset_meta_url;

        status = Status::InternalError("failed to init rowset meta");

        status.to_thrift(&tstatus);

        return;

    }

    RowsetId new_rowset_id = engine.next_rowset_id();

    auto pending_rs_guard = engine.pending_local_rowsets().add(new_rowset_id);

    rowset_meta->set_rowset_id(new_rowset_id);

    rowset_meta->set_tablet_uid(local_tablet->tablet_uid());

    // Step 5: get all segment files

    // Step 5.1: get all segment files size

    std::vector<std::string> segment_file_urls;

    segment_file_urls.reserve(num_segments);

    std::vector<uint64_t> segment_file_sizes;

    segment_file_sizes.reserve(num_segments);

    for (int64_t segment_index = 0; segment_index < num_segments; ++segment_index) {

        auto get_segment_file_size_url = fmt::format(

                "{}?method={}&tablet_id={}&rowset_id={}&segment_index={}", binlog_api_url,

                "get_segment_file", request.remote_tablet_id, remote_rowset_id, segment_index);

        uint64_t segment_file_size;

        auto get_segment_file_size_cb = [&get_segment_file_size_url,

                                         &segment_file_size](HttpClient* client) {

            RETURN_IF_ERROR(client->init(get_segment_file_size_url));

            client->set_timeout_ms(config::download_binlog_meta_timeout_ms);

            RETURN_IF_ERROR(client->head());

            return client->get_content_length(&segment_file_size);

        };

        status = _exec_http_req(client, max_retry, 1, get_segment_file_size_cb);

        if (!status.ok()) {

            LOG(WARNING) << "failed to get segment file size from " << get_segment_file_size_url

                         << ", status=" << status.to_string();

            status.to_thrift(&tstatus);

            return;

        }

        segment_file_sizes.push_back(segment_file_size);

        segment_file_urls.push_back(std::move(get_segment_file_size_url));

    }

    elapsed_time_map.emplace("get_segment_file_size", watch.elapsed_time_microseconds());

    // Step 5.2: check data capacity

    uint64_t total_size = std::accumulate(segment_file_sizes.begin(), segment_file_sizes.end(),

                                          0ULL); // NOLINT(bugprone-fold-init-type)

    if (!local_tablet->can_add_binlog(total_size)) {

        LOG(WARNING) << "failed to add binlog, no enough space, total_size=" << total_size

                     << ", tablet=" << local_tablet->tablet_id();

        status = Status::InternalError("no enough space");

        status.to_thrift(&tstatus);

        return;

    }

    total_download_bytes = total_size;

    total_download_files = num_segments;

    // Step 5.3: get all segment files

    for (int64_t segment_index = 0; segment_index < num_segments; ++segment_index) {

        auto segment_file_size = segment_file_sizes[segment_index];

        auto get_segment_file_url = segment_file_urls[segment_index];

        if (config::enable_download_md5sum_check) {

            get_segment_file_url = fmt::format("{}&acquire_md5=true", get_segment_file_url);

        }

        auto segment_path = local_segment_path(local_tablet->tablet_path(),

                                               rowset_meta->rowset_id().to_string(), segment_index);

        LOG(INFO) << "download segment file from " << get_segment_file_url << " to "

                  << segment_path;

        uint64_t estimate_timeout = estimate_download_timeout(segment_file_size);

        auto get_segment_file_cb = [&get_segment_file_url, &segment_path, segment_file_size,

                                    estimate_timeout, &download_success_files](HttpClient* client) {

            return _download_binlog_segment_file(client, get_segment_file_url, segment_path,

                                                 segment_file_size, estimate_timeout,

                                                 download_success_files);

        };

        status = _exec_http_req(client, max_retry, 1, get_segment_file_cb);

        if (!status.ok()) {

            LOG(WARNING) << "failed to get segment file from " << get_segment_file_url

                         << ", status=" << status.to_string();

            status.to_thrift(&tstatus);

            return;

        }

    }

    elapsed_time_map.emplace("get_segment_files", watch.elapsed_time_microseconds());

    // Step 6: get all segment index files

    // Step 6.1: get all segment index files size

    std::vector<std::string> segment_index_file_urls;

    std::vector<uint64_t> segment_index_file_sizes;

    std::vector<std::string> segment_index_file_names;

    auto tablet_schema = rowset_meta->tablet_schema();

    if (tablet_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) {

        for (const auto& index : tablet_schema->inverted_indexes()) {

            auto index_id = index->index_id();

            for (int64_t segment_index = 0; segment_index < num_segments; ++segment_index) {

                auto get_segment_index_file_size_url = fmt::format(

                        "{}?method={}&tablet_id={}&rowset_id={}&segment_index={}&segment_index_id={"

                        "}",

                        binlog_api_url, "get_segment_index_file", request.remote_tablet_id,

                        remote_rowset_id, segment_index, index_id);

                uint64_t segment_index_file_size;

                auto get_segment_index_file_size_cb =

                        [&get_segment_index_file_size_url,

                         &segment_index_file_size](HttpClient* client) {

                            RETURN_IF_ERROR(client->init(get_segment_index_file_size_url));

                            client->set_timeout_ms(config::download_binlog_meta_timeout_ms);

                            RETURN_IF_ERROR(client->head());

                            return client->get_content_length(&segment_index_file_size);

                        };

                auto segment_path =

                        local_segment_path(local_tablet->tablet_path(),

                                           rowset_meta->rowset_id().to_string(), segment_index);

                segment_index_file_names.push_back(InvertedIndexDescriptor::get_index_file_path_v1(

                        InvertedIndexDescriptor::get_index_file_path_prefix(segment_path), index_id,

                        index->get_index_suffix()));

                status = _exec_http_req(client, max_retry, 1, get_segment_index_file_size_cb);

                if (!status.ok()) {

                    LOG(WARNING) << "failed to get segment file size from "

                                 << get_segment_index_file_size_url

                                 << ", status=" << status.to_string();

                    status.to_thrift(&tstatus);

                    return;

                }

                segment_index_file_sizes.push_back(segment_index_file_size);

                segment_index_file_urls.push_back(std::move(get_segment_index_file_size_url));

            }

        }

    } else {

        for (int64_t segment_index = 0; segment_index < num_segments; ++segment_index) {

            if (tablet_schema->has_inverted_index() || tablet_schema->has_ann_index()) {

                auto get_segment_index_file_size_url = fmt::format(

                        "{}?method={}&tablet_id={}&rowset_id={}&segment_index={}&segment_index_id={"

                        "}",

                        binlog_api_url, "get_segment_index_file", request.remote_tablet_id,

                        remote_rowset_id, segment_index, -1);

                uint64_t segment_index_file_size;

                auto get_segment_index_file_size_cb =

                        [&get_segment_index_file_size_url,

                         &segment_index_file_size](HttpClient* client) {

                            RETURN_IF_ERROR(client->init(get_segment_index_file_size_url));

                            client->set_timeout_ms(config::download_binlog_meta_timeout_ms);

                            RETURN_IF_ERROR(client->head());

                            return client->get_content_length(&segment_index_file_size);

                        };

                auto segment_path =

                        local_segment_path(local_tablet->tablet_path(),

                                           rowset_meta->rowset_id().to_string(), segment_index);

                segment_index_file_names.push_back(InvertedIndexDescriptor::get_index_file_path_v2(

                        InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)));

                status = _exec_http_req(client, max_retry, 1, get_segment_index_file_size_cb);

                if (!status.ok()) {

                    LOG(WARNING) << "failed to get segment file size from "

                                 << get_segment_index_file_size_url

                                 << ", status=" << status.to_string();

                    status.to_thrift(&tstatus);

                    return;

                }

                segment_index_file_sizes.push_back(segment_index_file_size);

                segment_index_file_urls.push_back(std::move(get_segment_index_file_size_url));

            }

        }

    }

    elapsed_time_map.emplace("get_segment_index_file_size", watch.elapsed_time_microseconds());

    // Step 6.2: check data capacity

    uint64_t total_index_size =

            std::accumulate(segment_index_file_sizes.begin(), segment_index_file_sizes.end(),

                            0ULL); // NOLINT(bugprone-fold-init-type)

    if (!local_tablet->can_add_binlog(total_index_size)) {

        LOG(WARNING) << "failed to add binlog, no enough space, total_index_size="

                     << total_index_size << ", tablet=" << local_tablet->tablet_id();

        status = Status::InternalError("no enough space");

        status.to_thrift(&tstatus);

        return;

    }

    total_download_bytes += total_index_size;

    total_download_files += segment_index_file_urls.size();

    // Step 6.3: get all segment index files

    DCHECK(segment_index_file_sizes.size() == segment_index_file_names.size());

    DCHECK(segment_index_file_names.size() == segment_index_file_urls.size());

    for (int64_t i = 0; i < segment_index_file_urls.size(); ++i) {

        auto segment_index_file_size = segment_index_file_sizes[i];

        auto get_segment_index_file_url = segment_index_file_urls[i];

        if (config::enable_download_md5sum_check) {

            get_segment_index_file_url =

                    fmt::format("{}&acquire_md5=true", get_segment_index_file_url);

        }

        uint64_t estimate_timeout = estimate_download_timeout(segment_index_file_size);

        auto local_segment_index_path = segment_index_file_names[i];

        LOG(INFO) << fmt::format("download segment index file from {} to {}",

                                 get_segment_index_file_url, local_segment_index_path);

        auto get_segment_index_file_cb = [&get_segment_index_file_url, &local_segment_index_path,

                                          segment_index_file_size, estimate_timeout,

                                          &download_success_files](HttpClient* client) {

            return _download_binlog_index_file(client, get_segment_index_file_url,

                                               local_segment_index_path, segment_index_file_size,

                                               estimate_timeout, download_success_files);

        };

        status = _exec_http_req(client, max_retry, 1, get_segment_index_file_cb);

        if (!status.ok()) {

            LOG(WARNING) << "failed to get segment index file from " << get_segment_index_file_url

                         << ", status=" << status.to_string();

            status.to_thrift(&tstatus);

            return;

        }

    }

    elapsed_time_map.emplace("get_segment_index_files", watch.elapsed_time_microseconds());

    // Step 7: create rowset && calculate delete bitmap && commit

    // Step 7.1: create rowset

    RowsetSharedPtr rowset;

    status = RowsetFactory::create_rowset(local_tablet->tablet_schema(),

                                          local_tablet->tablet_path(), rowset_meta, &rowset);

    if (!status) {

        LOG(WARNING) << "failed to create rowset from rowset meta for remote tablet"

                     << ". rowset_id: " << rowset_meta_pb.rowset_id()

                     << ", rowset_type: " << rowset_meta_pb.rowset_type()

                     << ", remote_tablet_id=" << rowset_meta_pb.tablet_id() << ", txn_id=" << txn_id

                     << ", status=" << status.to_string();

        status.to_thrift(&tstatus);

        return;

    }

    // Step 7.2 calculate delete bitmap before commit

    auto calc_delete_bitmap_token = engine.calc_delete_bitmap_executor()->create_token();

    DeleteBitmapPtr delete_bitmap = std::make_shared<DeleteBitmap>(local_tablet_id);

    RowsetIdUnorderedSet pre_rowset_ids;

    if (local_tablet->enable_unique_key_merge_on_write()) {

        auto beta_rowset = reinterpret_cast<BetaRowset*>(rowset.get());

        std::vector<segment_v2::SegmentSharedPtr> segments;

        status = beta_rowset->load_segments(&segments);

        if (!status) {

            LOG(WARNING) << "failed to load segments from rowset"

                         << ". rowset_id: " << beta_rowset->rowset_id() << ", txn_id=" << txn_id

                         << ", status=" << status.to_string();

            status.to_thrift(&tstatus);

            return;

        }

        elapsed_time_map.emplace("load_segments", watch.elapsed_time_microseconds());

        if (segments.size() > 1) {

            // calculate delete bitmap between segments

            status = local_tablet->calc_delete_bitmap_between_segments(

                    rowset->tablet_schema(), rowset->rowset_id(), segments, delete_bitmap);

            if (!status) {

                LOG(WARNING) << "failed to calculate delete bitmap"

                             << ". tablet_id: " << local_tablet->tablet_id()

                             << ". rowset_id: " << rowset->rowset_id() << ", txn_id=" << txn_id

                             << ", status=" << status.to_string();

                status.to_thrift(&tstatus);

                return;

            }

            elapsed_time_map.emplace("calc_delete_bitmap", watch.elapsed_time_microseconds());

        }

        static_cast<void>(BaseTablet::commit_phase_update_delete_bitmap(

                local_tablet, rowset, pre_rowset_ids, delete_bitmap, segments, txn_id,

                calc_delete_bitmap_token.get(), nullptr));

        elapsed_time_map.emplace("commit_phase_update_delete_bitmap",

                                 watch.elapsed_time_microseconds());

        static_cast<void>(calc_delete_bitmap_token->wait());

        elapsed_time_map.emplace("wait_delete_bitmap", watch.elapsed_time_microseconds());

    }

    // Step 7.3: commit txn

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

            local_tablet->data_dir()->get_meta(), rowset_meta->partition_id(),

            rowset_meta->txn_id(), rowset_meta->tablet_id(), local_tablet->tablet_uid(),

            rowset_meta->load_id(), rowset, std::move(pending_rs_guard), false);

    if (!commit_txn_status && !commit_txn_status.is<ErrorCode::PUSH_TRANSACTION_ALREADY_EXIST>()) {

        auto err_msg = fmt::format(

                "failed to commit txn for remote tablet. rowset_id: {}, remote_tablet_id={}, "

                "txn_id={}, status={}",

                rowset_meta->rowset_id().to_string(), rowset_meta->tablet_id(),

                rowset_meta->txn_id(), commit_txn_status.to_string());

        LOG(WARNING) << err_msg;

        set_tstatus(TStatusCode::RUNTIME_ERROR, std::move(err_msg));

        return;

    }

    elapsed_time_map.emplace("commit_txn", watch.elapsed_time_microseconds());

    if (local_tablet->enable_unique_key_merge_on_write()) {

        engine.txn_manager()->set_txn_related_delete_bitmap(partition_id, txn_id, local_tablet_id,

                                                            local_tablet->tablet_uid(), true,

                                                            delete_bitmap, pre_rowset_ids, nullptr);

        elapsed_time_map.emplace("set_txn_related_delete_bitmap",

                                 watch.elapsed_time_microseconds());

    }

    tstatus.__set_status_code(TStatusCode::OK);

}

} // namespace

BaseBackendService::BaseBackendService(ExecEnv* exec_env)

        : _exec_env(exec_env), _agent_server(new AgentServer(exec_env, exec_env->cluster_info())) {}

BaseBackendService::~BaseBackendService() = default;

BackendService::BackendService(StorageEngine& engine, ExecEnv* exec_env)

        : BaseBackendService(exec_env), _engine(engine) {}

BackendService::~BackendService() = default;

Status BackendService::start_thrift_dependencies() {

    _agent_server->start_workers(_engine, _exec_env);

    auto thread_num = config::ingest_binlog_work_pool_size;

    if (thread_num < 0) {

        LOG(INFO) << fmt::format("ingest binlog thread pool size is {}, so we will in sync mode",

                                 thread_num);

        return Status::OK();

    }

    if (thread_num == 0) {

        thread_num = std::thread::hardware_concurrency();

    }

    static_cast<void>(doris::ThreadPoolBuilder("IngestBinlog")

                              .set_min_threads(thread_num)

                              .set_max_threads(thread_num * 2)

                              .build(&_ingest_binlog_workers));

    LOG(INFO) << fmt::format("ingest binlog thread pool size is {}, in async mode", thread_num);

    return Status::OK();

}

void BackendService::get_tablet_stat(TTabletStatResult& result) {

    _engine.tablet_manager()->get_tablet_stat(&result);

}

int64_t BackendService::get_trash_used_capacity() {

    int64_t result = 0;

    std::vector<DataDirInfo> data_dir_infos;

    static_cast<void>(_engine.get_all_data_dir_info(&data_dir_infos, false /*do not update */));

    // uses excute sql `show trash`, then update backend trash capacity too.

    _engine.notify_listener("REPORT_DISK_STATE");

    for (const auto& root_path_info : data_dir_infos) {

        result += root_path_info.trash_used_capacity;

    }

    return result;

}

void BackendService::get_disk_trash_used_capacity(std::vector<TDiskTrashInfo>& diskTrashInfos) {

    std::vector<DataDirInfo> data_dir_infos;

    static_cast<void>(_engine.get_all_data_dir_info(&data_dir_infos, false /*do not update */));

    // uses excute sql `show trash on <be>`, then update backend trash capacity too.

    _engine.notify_listener("REPORT_DISK_STATE");

    for (const auto& root_path_info : data_dir_infos) {

        TDiskTrashInfo diskTrashInfo;

        diskTrashInfo.__set_root_path(root_path_info.path);

        diskTrashInfo.__set_state(root_path_info.is_used ? "ONLINE" : "OFFLINE");

        diskTrashInfo.__set_trash_used_capacity(root_path_info.trash_used_capacity);

        diskTrashInfos.push_back(diskTrashInfo);

    }

}

void BaseBackendService::submit_routine_load_task(TStatus& t_status,

                                                  const std::vector<TRoutineLoadTask>& tasks) {

    for (auto& task : tasks) {

        Status st = _exec_env->routine_load_task_executor()->submit_task(task);

        if (!st.ok()) {

            LOG(WARNING) << "failed to submit routine load task. job id: " << task.job_id

                         << " task id: " << task.id;

            return st.to_thrift(&t_status);

        }

    }

    return Status::OK().to_thrift(&t_status);

}

/*

 * 1. validate user privilege (todo)

 * 2. FragmentMgr#exec_plan_fragment

 */

void BaseBackendService::open_scanner(TScanOpenResult& result_, const TScanOpenParams& params) {

    TStatus t_status;

    TUniqueId fragment_instance_id = generate_uuid();

    // A query_id is randomly generated to replace t_query_plan_info.query_id.

    // external query does not need to report anything to FE, so the query_id can be changed.

    // Otherwise, multiple independent concurrent open tablet scanners have the same query_id.

    // when one of the scanners ends, the other scanners will be canceled through FragmentMgr.cancel(query_id).

    TUniqueId query_id = generate_uuid();

    std::shared_ptr<ScanContext> p_context;

    static_cast<void>(_exec_env->external_scan_context_mgr()->create_scan_context(&p_context));

    p_context->fragment_instance_id = fragment_instance_id;

    p_context->offset = 0;

    p_context->last_access_time = time(nullptr);

    if (params.__isset.keep_alive_min) {

        p_context->keep_alive_min = params.keep_alive_min;

    } else {

        p_context->keep_alive_min = 5;

    }

    Status exec_st;

    TQueryPlanInfo t_query_plan_info;

    {

        const std::string& opaqued_query_plan = params.opaqued_query_plan;

        std::string query_plan_info;

        // base64 decode query plan

        if (!base64_decode(opaqued_query_plan, &query_plan_info)) {

            LOG(WARNING) << "open context error: base64_decode decode opaqued_query_plan failure";

            std::stringstream msg;

            msg << "query_plan_info: " << query_plan_info

                << " validate error, should not be modified after returned Doris FE processed";

            exec_st = Status::InvalidArgument(msg.str());

        }

        const uint8_t* buf = (const uint8_t*)query_plan_info.data();

        uint32_t len = (uint32_t)query_plan_info.size();

        // deserialize TQueryPlanInfo

        auto st = deserialize_thrift_msg(buf, &len, false, &t_query_plan_info);

        if (!st.ok()) {

            LOG(WARNING) << "open context error: deserialize TQueryPlanInfo failure";

            std::stringstream msg;

            msg << "query_plan_info: " << query_plan_info

                << " deserialize error, should not be modified after returned Doris FE processed";

            exec_st = Status::InvalidArgument(msg.str());

        }

        p_context->query_id = query_id;

    }

    std::vector<TScanColumnDesc> selected_columns;

    if (exec_st.ok()) {

        // start the scan procedure

        LOG(INFO) << fmt::format(

                "exec external scanner, old_query_id = {}, new_query_id = {}, fragment_instance_id "

                "= {}",

                print_id(t_query_plan_info.query_id), print_id(query_id),

                print_id(fragment_instance_id));

        exec_st = _exec_env->fragment_mgr()->exec_external_plan_fragment(

                params, t_query_plan_info, query_id, fragment_instance_id, &selected_columns);

    }

    exec_st.to_thrift(&t_status);

    //return status

    // t_status.status_code = TStatusCode::OK;

    result_.status = t_status;

    result_.__set_context_id(p_context->context_id);

    result_.__set_selected_columns(selected_columns);

}

// fetch result from polling the queue, should always maintain the context offset, otherwise inconsistent result

void BaseBackendService::get_next(TScanBatchResult& result_, const TScanNextBatchParams& params) {

    std::string context_id = params.context_id;

    u_int64_t offset = params.offset;

    TStatus t_status;

    std::shared_ptr<ScanContext> context;

    Status st = _exec_env->external_scan_context_mgr()->get_scan_context(context_id, &context);

    if (!st.ok()) {

        st.to_thrift(&t_status);

        result_.status = t_status;

        return;

    }

    if (offset != context->offset) {

        LOG(ERROR) << "getNext error: context offset [" << context->offset << " ]"

                   << " ,client offset [ " << offset << " ]";

        // invalid offset

        t_status.status_code = TStatusCode::NOT_FOUND;

        t_status.error_msgs.push_back(

                absl::Substitute("context_id=$0, send_offset=$1, context_offset=$2", context_id,

                                 offset, context->offset));

        result_.status = t_status;

    } else {

        // during accessing, should disabled last_access_time

        context->last_access_time = -1;

        TUniqueId fragment_instance_id = context->fragment_instance_id;

        std::shared_ptr<arrow::RecordBatch> record_batch;

        bool eos;

        st = _exec_env->result_queue_mgr()->fetch_result(fragment_instance_id, &record_batch, &eos);

        if (st.ok()) {

            result_.__set_eos(eos);

            if (!eos) {

                std::string record_batch_str;

                st = serialize_record_batch(*record_batch, &record_batch_str);

                st.to_thrift(&t_status);

                if (st.ok()) {

                    // avoid copy large string

                    result_.rows = std::move(record_batch_str);

                    // set __isset

                    result_.__isset.rows = true;

                    context->offset += record_batch->num_rows();

                }

            }

        } else {

            LOG(WARNING) << "fragment_instance_id [" << print_id(fragment_instance_id)

                         << "] fetch result status [" << st.to_string() + "]";

            st.to_thrift(&t_status);

            result_.status = t_status;

        }

    }

    context->last_access_time = time(nullptr);

}

void BaseBackendService::close_scanner(TScanCloseResult& result_, const TScanCloseParams& params) {

    std::string context_id = params.context_id;

    TStatus t_status;

    Status st = _exec_env->external_scan_context_mgr()->clear_scan_context(context_id);

    st.to_thrift(&t_status);

    result_.status = t_status;

}

void BackendService::get_stream_load_record(TStreamLoadRecordResult& result,

                                            int64_t last_stream_record_time) {

    BaseBackendService::get_stream_load_record(result, last_stream_record_time,

                                               _engine.get_stream_load_recorder());

}

void BackendService::check_storage_format(TCheckStorageFormatResult& result) {

    _engine.tablet_manager()->get_all_tablets_storage_format(&result);

}

void BackendService::make_snapshot(TAgentResult& return_value,

                                   const TSnapshotRequest& snapshot_request) {

    std::string snapshot_path;

    bool allow_incremental_clone = false;

    Status status = _engine.snapshot_mgr()->make_snapshot(snapshot_request, &snapshot_path,

                                                          &allow_incremental_clone);

    if (!status) {

        LOG_WARNING("failed to make snapshot")

                .tag("tablet_id", snapshot_request.tablet_id)