be/src/service/backend_service.cpp

Source
// 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 = 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());

    // Binlog is only supported in local mode. Segment-list rowsets are cloud-only, so binlog files
    // are still named by contiguous segment indexes rather than persisted segment_ids.
    // 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)
                .tag("schema_hash", snapshot_request.schema_hash)
                .error(status);
    } else {
        LOG_INFO("successfully make snapshot")
                .tag("tablet_id", snapshot_request.tablet_id)
                .tag("schema_hash", snapshot_request.schema_hash)
                .tag("snapshot_path", snapshot_path);
        return_value.__set_snapshot_path(snapshot_path);
        return_value.__set_allow_incremental_clone(allow_incremental_clone);
    }

    status.to_thrift(&return_value.status);
    return_value.__set_snapshot_version(snapshot_request.preferred_snapshot_version);
}

void BackendService::release_snapshot(TAgentResult& return_value,
                                      const std::string& snapshot_path) {
    Status status = _engine.snapshot_mgr()->release_snapshot(snapshot_path);
    if (!status) {
        LOG_WARNING("failed to release snapshot").tag("snapshot_path", snapshot_path).error(status);
    } else {
        LOG_INFO("successfully release snapshot").tag("snapshot_path", snapshot_path);
    }
    status.to_thrift(&return_value.status);
}

void BackendService::ingest_binlog(TIngestBinlogResult& result,
                                   const TIngestBinlogRequest& request) {
    LOG(INFO) << "ingest binlog. request: " << apache::thrift::ThriftDebugString(request);

    TStatus tstatus;
    Defer defer {[&result, &tstatus]() {
        result.__set_status(tstatus);
        LOG(INFO) << "ingest binlog. result: " << apache::thrift::ThriftDebugString(result);
    }};

    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));
    };

    if (!config::enable_feature_binlog) {
        set_tstatus(TStatusCode::RUNTIME_ERROR, "enable feature binlog is false");
        return;
    }

    /// Check args: txn_id, remote_tablet_id, binlog_version, remote_host, remote_port, partition_id, load_id
    if (!request.__isset.txn_id) {
        auto error_msg = "txn_id is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.remote_tablet_id) {
        auto error_msg = "remote_tablet_id is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.binlog_version) {
        auto error_msg = "binlog_version is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.remote_host) {
        auto error_msg = "remote_host is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.remote_port) {
        auto error_msg = "remote_port is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.partition_id) {
        auto error_msg = "partition_id is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.local_tablet_id) {
        auto error_msg = "local_tablet_id is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.load_id) {
        auto error_msg = "load_id is empty";
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg);
        return;
    }

    auto txn_id = request.txn_id;
    // Step 1: get local tablet
    auto const& local_tablet_id = request.local_tablet_id;
    auto local_tablet = _engine.tablet_manager()->get_tablet(local_tablet_id);
    if (local_tablet == nullptr) {
        auto error_msg = fmt::format("tablet {} not found", local_tablet_id);
        LOG(WARNING) << error_msg;
        set_tstatus(TStatusCode::TABLET_MISSING, std::move(error_msg));
        return;
    }

    // Step 2: check txn, create txn, prepare_txn will check it
    auto partition_id = request.partition_id;
    auto& load_id = request.load_id;
    auto is_ingrest = true;
    PUniqueId p_load_id;
    p_load_id.set_hi(load_id.hi);
    p_load_id.set_lo(load_id.lo);

    {
        // TODO: Before push_lock is not held, but I think it should hold.
        auto status = local_tablet->prepare_txn(partition_id, txn_id, p_load_id, is_ingrest);
        if (!status.ok()) {
            LOG(WARNING) << "prepare txn failed. txn_id=" << txn_id
                         << ", status=" << status.to_string();
            status.to_thrift(&tstatus);
            return;
        }
    }

    bool is_async = (_ingest_binlog_workers != nullptr);
    result.__set_is_async(is_async);

    auto ingest_binlog_func = [=, this, tstatus = &tstatus]() {
        IngestBinlogArg ingest_binlog_arg = {
                .txn_id = txn_id,
                .partition_id = partition_id,
                .local_tablet_id = local_tablet_id,
                .local_tablet = local_tablet,

                .request = request,
                .tstatus = is_async ? nullptr : tstatus,
        };

        _ingest_binlog(_engine, &ingest_binlog_arg);
    };

    if (is_async) {
        auto status = _ingest_binlog_workers->submit_func(std::move(ingest_binlog_func));
        if (!status.ok()) {
            status.to_thrift(&tstatus);
            return;
        }
    } else {
        ingest_binlog_func();
    }
}

void BackendService::query_ingest_binlog(TQueryIngestBinlogResult& result,
                                         const TQueryIngestBinlogRequest& request) {
    LOG(INFO) << "query ingest binlog. request: " << apache::thrift::ThriftDebugString(request);

    auto set_result = [&](TIngestBinlogStatus::type status, std::string error_msg) {
        result.__set_status(status);
        result.__set_err_msg(std::move(error_msg));
    };

    /// Check args: txn_id, partition_id, tablet_id, load_id
    if (!request.__isset.txn_id) {
        auto error_msg = "txn_id is empty";
        LOG(WARNING) << error_msg;
        set_result(TIngestBinlogStatus::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.partition_id) {
        auto error_msg = "partition_id is empty";
        LOG(WARNING) << error_msg;
        set_result(TIngestBinlogStatus::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.tablet_id) {
        auto error_msg = "tablet_id is empty";
        LOG(WARNING) << error_msg;
        set_result(TIngestBinlogStatus::ANALYSIS_ERROR, error_msg);
        return;
    }
    if (!request.__isset.load_id) {
        auto error_msg = "load_id is empty";
        LOG(WARNING) << error_msg;
        set_result(TIngestBinlogStatus::ANALYSIS_ERROR, error_msg);
        return;
    }

    auto partition_id = request.partition_id;
    auto txn_id = request.txn_id;
    auto tablet_id = request.tablet_id;

    // Step 1: get local tablet
    auto local_tablet = _engine.tablet_manager()->get_tablet(tablet_id);
    if (local_tablet == nullptr) {
        auto error_msg = fmt::format("tablet {} not found", tablet_id);
        LOG(WARNING) << error_msg;
        set_result(TIngestBinlogStatus::NOT_FOUND, std::move(error_msg));
        return;
    }

    // Step 2: get txn state
    auto tablet_uid = local_tablet->tablet_uid();
    auto txn_state =
            _engine.txn_manager()->get_txn_state(partition_id, txn_id, tablet_id, tablet_uid);
    switch (txn_state) {
    case TxnState::NOT_FOUND:
        result.__set_status(TIngestBinlogStatus::NOT_FOUND);
        break;
    case TxnState::PREPARED:
        result.__set_status(TIngestBinlogStatus::DOING);
        break;
    case TxnState::COMMITTED:
        result.__set_status(TIngestBinlogStatus::OK);
        break;
    case TxnState::ROLLEDBACK:
        result.__set_status(TIngestBinlogStatus::FAILED);
        break;
    case TxnState::ABORTED:
        result.__set_status(TIngestBinlogStatus::FAILED);
        break;
    case TxnState::DELETED:
        result.__set_status(TIngestBinlogStatus::FAILED);
        break;
    }
}

void BaseBackendService::get_tablet_stat(TTabletStatResult& result) {
    LOG(ERROR) << "get_tablet_stat is not implemented";
}

int64_t BaseBackendService::get_trash_used_capacity() {
    LOG(ERROR) << "get_trash_used_capacity is not implemented";
    return 0;
}

void BaseBackendService::get_stream_load_record(TStreamLoadRecordResult& result,
                                                int64_t last_stream_record_time) {
    LOG(ERROR) << "get_stream_load_record is not implemented";
}

void BaseBackendService::get_stream_load_record(
        TStreamLoadRecordResult& result, int64_t last_stream_record_time,
        std::shared_ptr<StreamLoadRecorder> stream_load_recorder) {
    if (stream_load_recorder != nullptr) {
        std::map<std::string, std::string> records;
        auto st = stream_load_recorder->get_batch(std::to_string(last_stream_record_time),
                                                  config::stream_load_record_batch_size, &records);
        if (st.ok()) {
            LOG(INFO) << "get_batch stream_load_record rocksdb successfully. records size: "
                      << records.size()
                      << ", last_stream_load_timestamp: " << last_stream_record_time;
            std::map<std::string, TStreamLoadRecord> stream_load_record_batch;
            auto it = records.begin();
            for (; it != records.end(); ++it) {
                TStreamLoadRecord stream_load_item;
                StreamLoadContext::parse_stream_load_record(it->second, stream_load_item);
                stream_load_record_batch.emplace(it->first.c_str(), stream_load_item);
            }
            result.__set_stream_load_record(stream_load_record_batch);
        }
    } else {
        LOG(WARNING) << "stream_load_recorder is null.";
    }
}

void BaseBackendService::get_disk_trash_used_capacity(std::vector<TDiskTrashInfo>& diskTrashInfos) {
    LOG(ERROR) << "get_disk_trash_used_capacity is not implemented";
}

void BaseBackendService::make_snapshot(TAgentResult& return_value,
                                       const TSnapshotRequest& snapshot_request) {
    LOG(ERROR) << "make_snapshot is not implemented";
    return_value.__set_status(Status::NotSupported("make_snapshot is not implemented").to_thrift());
}

void BaseBackendService::release_snapshot(TAgentResult& return_value,
                                          const std::string& snapshot_path) {
    LOG(ERROR) << "release_snapshot is not implemented";
    return_value.__set_status(
            Status::NotSupported("release_snapshot is not implemented").to_thrift());
}

void BaseBackendService::check_storage_format(TCheckStorageFormatResult& result) {
    LOG(ERROR) << "check_storage_format is not implemented";
}

void BaseBackendService::ingest_binlog(TIngestBinlogResult& result,
                                       const TIngestBinlogRequest& request) {
    LOG(ERROR) << "ingest_binlog is not implemented";
    result.__set_status(Status::NotSupported("ingest_binlog is not implemented").to_thrift());
}

void BaseBackendService::query_ingest_binlog(TQueryIngestBinlogResult& result,
                                             const TQueryIngestBinlogRequest& request) {
    LOG(ERROR) << "query_ingest_binlog is not implemented";
    result.__set_status(TIngestBinlogStatus::UNKNOWN);
    result.__set_err_msg("query_ingest_binlog is not implemented");
}

void BaseBackendService::warm_up_cache_async(TWarmUpCacheAsyncResponse& response,
                                             const TWarmUpCacheAsyncRequest& request) {
    LOG(ERROR) << "warm_up_cache_async is not implemented";
    response.__set_status(
            Status::NotSupported("warm_up_cache_async is not implemented").to_thrift());
}

void BaseBackendService::check_warm_up_cache_async(TCheckWarmUpCacheAsyncResponse& response,
                                                   const TCheckWarmUpCacheAsyncRequest& request) {
    LOG(ERROR) << "check_warm_up_cache_async is not implemented";
    response.__set_status(
            Status::NotSupported("check_warm_up_cache_async is not implemented").to_thrift());
}

void BaseBackendService::sync_load_for_tablets(TSyncLoadForTabletsResponse& response,
                                               const TSyncLoadForTabletsRequest& request) {
    LOG(ERROR) << "sync_load_for_tablets is not implemented";
}

void BaseBackendService::get_top_n_hot_partitions(TGetTopNHotPartitionsResponse& response,
                                                  const TGetTopNHotPartitionsRequest& request) {
    LOG(ERROR) << "get_top_n_hot_partitions is not implemented";
}

void BaseBackendService::warm_up_tablets(TWarmUpTabletsResponse& response,
                                         const TWarmUpTabletsRequest& request) {
    LOG(ERROR) << "warm_up_tablets is not implemented";
    response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}

void BaseBackendService::get_realtime_exec_status(TGetRealtimeExecStatusResponse& response,
                                                  const TGetRealtimeExecStatusRequest& request) {
    if (!request.__isset.id) {
        LOG_WARNING("Invalidate argument, id is empty");
        response.__set_status(Status::InvalidArgument("id is empty").to_thrift());
        return;
    }

    RuntimeProfile::Counter get_realtime_timer {TUnit::TIME_NS};

    Defer _print_log([&]() {
        LOG_INFO("Getting realtime exec status of query {} , cost time {}", print_id(request.id),
                 PrettyPrinter::print(get_realtime_timer.value(), get_realtime_timer.type()));
    });

    SCOPED_TIMER(&get_realtime_timer);

    std::unique_ptr<TReportExecStatusParams> report_exec_status_params =
            std::make_unique<TReportExecStatusParams>();
    std::unique_ptr<TQueryStatistics> query_stats = std::make_unique<TQueryStatistics>();

    std::string req_type = request.__isset.req_type ? request.req_type : "profile";
    Status st;
    if (req_type == "stats") {
        st = ExecEnv::GetInstance()->fragment_mgr()->get_query_statistics(request.id,
                                                                          query_stats.get());
        if (st.ok()) {
            response.__set_query_stats(*query_stats);
        }
    } else {
        // default is "profile"
        st = ExecEnv::GetInstance()->fragment_mgr()->get_realtime_exec_status(
                request.id, report_exec_status_params.get());
        if (st.ok()) {
            response.__set_report_exec_status_params(*report_exec_status_params);
        }
    }

    report_exec_status_params->__set_query_id(TUniqueId());
    report_exec_status_params->__set_done(false);
    response.__set_status(st.to_thrift());
}

void BaseBackendService::get_dictionary_status(TDictionaryStatusList& result,
                                               const std::vector<int64_t>& dictionary_ids) {
    std::vector<TDictionaryStatus> dictionary_status;
    ExecEnv::GetInstance()->dict_factory()->get_dictionary_status(dictionary_status,
                                                                  dictionary_ids);
    result.__set_dictionary_status_list(dictionary_status);
    LOG(INFO) << "query for dictionary status, return " << result.dictionary_status_list.size()
              << " rows";
}

void BaseBackendService::test_storage_connectivity(TTestStorageConnectivityResponse& response,
                                                   const TTestStorageConnectivityRequest& request) {
    Status status = io::StorageConnectivityTester::test(request.type, request.properties);
    response.__set_status(status.to_thrift());
}

void BaseBackendService::get_python_envs(std::vector<TPythonEnvInfo>& result) {
    result = PythonVersionManager::instance().env_infos_to_thrift();
}

void BaseBackendService::get_python_packages(std::vector<TPythonPackageInfo>& result,
                                             const std::string& python_version) {
    PythonVersion version;
    auto& manager = PythonVersionManager::instance();
    THROW_IF_ERROR(manager.get_version(python_version, &version));

    std::vector<std::pair<std::string, std::string>> packages;
    THROW_IF_ERROR(list_installed_packages(version, &packages));
    result = manager.package_infos_to_thrift(packages);
}

} // namespace doris

Coverage Report

Created: 2026-08-10 11:21