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

#include <assert.h>

#include <brpc/closure_guard.h>

#include <brpc/controller.h>

#include <bthread/bthread.h>

#include <bthread/types.h>

#include <butil/errno.h>

#include <butil/iobuf.h>

#include <fcntl.h>

#include <fmt/core.h>

#include <gen_cpp/DataSinks_types.h>

#include <gen_cpp/MasterService_types.h>

#include <gen_cpp/PaloInternalService_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/Status_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/internal_service.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <gen_cpp/segment_v2.pb.h>

#include <gen_cpp/types.pb.h>

#include <google/protobuf/stubs/callback.h>

#include <stddef.h>

#include <stdint.h>

#include <sys/stat.h>

#include <vec/exec/vjdbc_connector.h>

#include <algorithm>

#include <exception>

#include <filesystem>

#include <memory>

#include <set>

#include <sstream>

#include <string>

#include <unordered_map>

#include <utility>

#include <vector>

#include "common/config.h"

#include "common/consts.h"

#include "common/exception.h"

#include "common/logging.h"

#include "common/signal_handler.h"

#include "common/status.h"

#include "exec/rowid_fetcher.h"

#include "gen_cpp/BackendService.h"

#include "gen_cpp/PaloInternalService_types.h"

#include "gen_cpp/internal_service.pb.h"

#include "gutil/integral_types.h"

#include "http/http_client.h"

#include "io/fs/local_file_system.h"

#include "io/fs/stream_load_pipe.h"

#include "io/io_common.h"

#include "olap/data_dir.h"

#include "olap/olap_common.h"

#include "olap/olap_define.h"

#include "olap/rowset/beta_rowset.h"

#include "olap/rowset/rowset.h"

#include "olap/rowset/rowset_factory.h"

#include "olap/rowset/rowset_meta.h"

#include "olap/rowset/segment_v2/column_reader.h"

#include "olap/rowset/segment_v2/common.h"

#include "olap/rowset/segment_v2/inverted_index_desc.h"

#include "olap/rowset/segment_v2/segment.h"

#include "olap/rowset/segment_v2/segment_iterator.h"

#include "olap/segment_loader.h"

#include "olap/storage_engine.h"

#include "olap/tablet.h"

#include "olap/tablet_fwd.h"

#include "olap/tablet_manager.h"

#include "olap/tablet_schema.h"

#include "olap/txn_manager.h"

#include "olap/utils.h"

#include "olap/wal/wal_manager.h"

#include "runtime/buffer_control_block.h"

#include "runtime/cache/result_cache.h"

#include "runtime/define_primitive_type.h"

#include "runtime/descriptors.h"

#include "runtime/exec_env.h"

#include "runtime/fold_constant_executor.h"

#include "runtime/fragment_mgr.h"

#include "runtime/load_channel_mgr.h"

#include "runtime/load_stream_mgr.h"

#include "runtime/result_buffer_mgr.h"

#include "runtime/routine_load/routine_load_task_executor.h"

#include "runtime/stream_load/new_load_stream_mgr.h"

#include "runtime/stream_load/stream_load_context.h"

#include "runtime/thread_context.h"

#include "runtime/types.h"

#include "service/backend_options.h"

#include "service/point_query_executor.h"

#include "util/arrow/row_batch.h"

#include "util/async_io.h"

#include "util/brpc_client_cache.h"

#include "util/doris_metrics.h"

#include "util/md5.h"

#include "util/metrics.h"

#include "util/network_util.h"

#include "util/proto_util.h"

#include "util/ref_count_closure.h"

#include "util/runtime_profile.h"

#include "util/stopwatch.hpp"

#include "util/string_util.h"

#include "util/thrift_util.h"

#include "util/time.h"

#include "util/uid_util.h"

#include "vec/columns/column.h"

#include "vec/columns/column_string.h"

#include "vec/common/schema_util.h"

#include "vec/core/block.h"

#include "vec/core/column_with_type_and_name.h"

#include "vec/data_types/data_type.h"

#include "vec/exec/format/avro//avro_jni_reader.h"

#include "vec/exec/format/csv/csv_reader.h"

#include "vec/exec/format/generic_reader.h"

#include "vec/exec/format/json/new_json_reader.h"

#include "vec/exec/format/orc/vorc_reader.h"

#include "vec/exec/format/parquet/vparquet_reader.h"

#include "vec/jsonb/serialize.h"

#include "vec/runtime/vdata_stream_mgr.h"

namespace google {

namespace protobuf {

class RpcController;

} // namespace protobuf

} // namespace google

namespace doris {

using namespace ErrorCode;

const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_pool_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_active_threads, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_active_threads, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_max_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_pool_max_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_max_threads, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_max_threads, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_pool_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_active_threads, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_pool_max_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_max_threads, MetricUnit::NOUNIT);

bthread_key_t btls_key;

static void thread_context_deleter(void* d) {

    delete static_cast<ThreadContext*>(d);

}

template <typename T>

class NewHttpClosure : public ::google::protobuf::Closure {

public:

    NewHttpClosure(google::protobuf::Closure* done) : _done(done) {}

    NewHttpClosure(T* request, google::protobuf::Closure* done) : _request(request), _done(done) {}

Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_28PTabletWriterAddBlockRequestEEC2EPS1_PN6google8protobuf7ClosureE

Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_19PTransmitDataParamsEEC2EPS1_PN6google8protobuf7ClosureE

    void Run() override {

        if (_request != nullptr) {

            delete _request;

            _request = nullptr;

        }

        if (_done != nullptr) {

            _done->Run();

        }

        delete this;

    }

Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_28PTabletWriterAddBlockRequestEE3RunEv

Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_19PTransmitDataParamsEE3RunEv

private:

    T* _request = nullptr;

    google::protobuf::Closure* _done = nullptr;

};

PInternalService::PInternalService(ExecEnv* exec_env)

        : _exec_env(exec_env),

          // heavy threadpool is used for load process and other process that will read disk or access network.

          _heavy_work_pool(config::brpc_heavy_work_pool_threads != -1

                                   ? config::brpc_heavy_work_pool_threads

                                   : std::max(128, CpuInfo::num_cores() * 4),

                           config::brpc_heavy_work_pool_max_queue_size != -1

                                   ? config::brpc_heavy_work_pool_max_queue_size

                                   : std::max(10240, CpuInfo::num_cores() * 320),

                           "brpc_heavy"),

          // light threadpool should be only used in query processing logic. All hanlers should be very light, not locked, not access disk.

          _light_work_pool(config::brpc_light_work_pool_threads != -1

                                   ? config::brpc_light_work_pool_threads

                                   : std::max(128, CpuInfo::num_cores() * 4),

                           config::brpc_light_work_pool_max_queue_size != -1

                                   ? config::brpc_light_work_pool_max_queue_size

                                   : std::max(10240, CpuInfo::num_cores() * 320),

                           "brpc_light"),

          _arrow_flight_work_pool(config::brpc_arrow_flight_work_pool_threads != -1

                                          ? config::brpc_arrow_flight_work_pool_threads

                                          : std::max(512, CpuInfo::num_cores() * 2),

                                  config::brpc_arrow_flight_work_pool_max_queue_size != -1

                                          ? config::brpc_arrow_flight_work_pool_max_queue_size

                                          : std::max(20480, CpuInfo::num_cores() * 640),

                                  "brpc_arrow_flight") {

    REGISTER_HOOK_METRIC(heavy_work_pool_queue_size,

                         [this]() { return _heavy_work_pool.get_queue_size(); });

    REGISTER_HOOK_METRIC(light_work_pool_queue_size,

                         [this]() { return _light_work_pool.get_queue_size(); });

    REGISTER_HOOK_METRIC(heavy_work_active_threads,

                         [this]() { return _heavy_work_pool.get_active_threads(); });

    REGISTER_HOOK_METRIC(light_work_active_threads,

                         [this]() { return _light_work_pool.get_active_threads(); });

    REGISTER_HOOK_METRIC(heavy_work_pool_max_queue_size,

                         []() { return config::brpc_heavy_work_pool_max_queue_size; });

    REGISTER_HOOK_METRIC(light_work_pool_max_queue_size,

                         []() { return config::brpc_light_work_pool_max_queue_size; });

    REGISTER_HOOK_METRIC(heavy_work_max_threads,

                         []() { return config::brpc_heavy_work_pool_threads; });

    REGISTER_HOOK_METRIC(light_work_max_threads,

                         []() { return config::brpc_light_work_pool_threads; });

    REGISTER_HOOK_METRIC(arrow_flight_work_pool_queue_size,

                         [this]() { return _arrow_flight_work_pool.get_queue_size(); });

    REGISTER_HOOK_METRIC(arrow_flight_work_active_threads,

                         [this]() { return _arrow_flight_work_pool.get_active_threads(); });

    REGISTER_HOOK_METRIC(arrow_flight_work_pool_max_queue_size,

                         []() { return config::brpc_arrow_flight_work_pool_max_queue_size; });

    REGISTER_HOOK_METRIC(arrow_flight_work_max_threads,

                         []() { return config::brpc_arrow_flight_work_pool_threads; });

    _exec_env->load_stream_mgr()->set_heavy_work_pool(&_heavy_work_pool);

    CHECK_EQ(0, bthread_key_create(&btls_key, thread_context_deleter));

    CHECK_EQ(0, bthread_key_create(&AsyncIO::btls_io_ctx_key, AsyncIO::io_ctx_key_deleter));

}

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

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

PInternalServiceImpl::~PInternalServiceImpl() = default;

PInternalService::~PInternalService() {

    DEREGISTER_HOOK_METRIC(heavy_work_pool_queue_size);

    DEREGISTER_HOOK_METRIC(light_work_pool_queue_size);

    DEREGISTER_HOOK_METRIC(heavy_work_active_threads);

    DEREGISTER_HOOK_METRIC(light_work_active_threads);

    DEREGISTER_HOOK_METRIC(heavy_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(light_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(heavy_work_max_threads);

    DEREGISTER_HOOK_METRIC(light_work_max_threads);

    DEREGISTER_HOOK_METRIC(arrow_flight_work_pool_queue_size);

    DEREGISTER_HOOK_METRIC(arrow_flight_work_active_threads);

    DEREGISTER_HOOK_METRIC(arrow_flight_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(arrow_flight_work_max_threads);

    CHECK_EQ(0, bthread_key_delete(btls_key));

    CHECK_EQ(0, bthread_key_delete(AsyncIO::btls_io_ctx_key));

}

void PInternalService::transmit_data(google::protobuf::RpcController* controller,

                                     const PTransmitDataParams* request,

                                     PTransmitDataResult* response,

                                     google::protobuf::Closure* done) {}

void PInternalService::transmit_data_by_http(google::protobuf::RpcController* controller,

                                             const PEmptyRequest* request,

                                             PTransmitDataResult* response,

                                             google::protobuf::Closure* done) {}

void PInternalService::_transmit_data(google::protobuf::RpcController* controller,

                                      const PTransmitDataParams* request,

                                      PTransmitDataResult* response,

                                      google::protobuf::Closure* done, const Status& extract_st) {}

void PInternalService::tablet_writer_open(google::protobuf::RpcController* controller,

                                          const PTabletWriterOpenRequest* request,

                                          PTabletWriterOpenResult* response,

                                          google::protobuf::Closure* done) {

    bool ret = _heavy_work_pool.try_offer([this, request, response, done]() {

        VLOG_RPC << "tablet writer open, id=" << request->id()

                 << ", index_id=" << request->index_id() << ", txn_id=" << request->txn_id();

        signal::set_signal_task_id(request->id());

        brpc::ClosureGuard closure_guard(done);

        auto st = _exec_env->load_channel_mgr()->open(*request);

        if (!st.ok()) {

            LOG(WARNING) << "load channel open failed, message=" << st << ", id=" << request->id()

                         << ", index_id=" << request->index_id()

                         << ", txn_id=" << request->txn_id();

        }

        st.to_protobuf(response->mutable_status());

    });

    if (!ret) {

        offer_failed(response, done, _heavy_work_pool);

        return;

    }

}

void PInternalService::exec_plan_fragment(google::protobuf::RpcController* controller,

                                          const PExecPlanFragmentRequest* request,

                                          PExecPlanFragmentResult* response,

                                          google::protobuf::Closure* done) {

    timeval tv {};

    gettimeofday(&tv, nullptr);

    response->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);

    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {

        _exec_plan_fragment_in_pthread(controller, request, response, done);

    });

    if (!ret) {

        offer_failed(response, done, _light_work_pool);

        return;

    }

}

void PInternalService::_exec_plan_fragment_in_pthread(google::protobuf::RpcController* controller,

                                                      const PExecPlanFragmentRequest* request,

                                                      PExecPlanFragmentResult* response,

                                                      google::protobuf::Closure* done) {

    timeval tv1 {};

    gettimeofday(&tv1, nullptr);

    response->set_execution_time(tv1.tv_sec * 1000LL + tv1.tv_usec / 1000);

    brpc::ClosureGuard closure_guard(done);

    auto st = Status::OK();

    bool compact = request->has_compact() ? request->compact() : false;

    PFragmentRequestVersion version =

            request->has_version() ? request->version() : PFragmentRequestVersion::VERSION_1;

    try {

        st = _exec_plan_fragment_impl(request->request(), version, compact);

    } catch (const Exception& e) {

        st = e.to_status();

    } catch (...) {

        st = Status::Error(ErrorCode::INTERNAL_ERROR,

                           "_exec_plan_fragment_impl meet unknown error");

    }

    if (!st.ok()) {

        LOG(WARNING) << "exec plan fragment failed, errmsg=" << st;

    }

    st.to_protobuf(response->mutable_status());

    timeval tv2 {};

    gettimeofday(&tv2, nullptr);

    response->set_execution_done_time(tv2.tv_sec * 1000LL + tv2.tv_usec / 1000);

}

void PInternalService::exec_plan_fragment_prepare(google::protobuf::RpcController* controller,

                                                  const PExecPlanFragmentRequest* request,

                                                  PExecPlanFragmentResult* response,

                                                  google::protobuf::Closure* done) {

    timeval tv {};

    gettimeofday(&tv, nullptr);

    response->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);

    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {

        _exec_plan_fragment_in_pthread(controller, request, response, done);

    });

    if (!ret) {

        offer_failed(response, done, _light_work_pool);

        return;

    }

}

void PInternalService::exec_plan_fragment_start(google::protobuf::RpcController* /*controller*/,

                                                const PExecPlanFragmentStartRequest* request,

                                                PExecPlanFragmentResult* result,

                                                google::protobuf::Closure* done) {

    timeval tv {};

    gettimeofday(&tv, nullptr);

    result->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);

    bool ret = _light_work_pool.try_offer([this, request, result, done]() {

        timeval tv1 {};

        gettimeofday(&tv1, nullptr);

        result->set_execution_time(tv1.tv_sec * 1000LL + tv1.tv_usec / 1000);

        brpc::ClosureGuard closure_guard(done);

        auto st = _exec_env->fragment_mgr()->start_query_execution(request);

        st.to_protobuf(result->mutable_status());

        timeval tv2 {};

        gettimeofday(&tv2, nullptr);

        result->set_execution_done_time(tv2.tv_sec * 1000LL + tv2.tv_usec / 1000);

    });

    if (!ret) {

        offer_failed(result, done, _light_work_pool);

        return;

    }

}

void PInternalService::open_load_stream(google::protobuf::RpcController* controller,

                                        const POpenLoadStreamRequest* request,

                                        POpenLoadStreamResponse* response,

                                        google::protobuf::Closure* done) {

    bool ret = _heavy_work_pool.try_offer([this, controller, request, response, done]() {

        signal::set_signal_task_id(request->load_id());

        brpc::ClosureGuard done_guard(done);

        brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);

        brpc::StreamOptions stream_options;

        LOG(INFO) << "open load stream, load_id=" << request->load_id()

                  << ", src_id=" << request->src_id();

        for (const auto& req : request->tablets()) {

            BaseTabletSPtr tablet;

            if (auto res = ExecEnv::get_tablet(req.tablet_id()); !res.has_value()) [[unlikely]] {

                auto st = std::move(res).error();

                st.to_protobuf(response->mutable_status());

                cntl->SetFailed(st.to_string());

                return;

            } else {

                tablet = std::move(res).value();

            }

            auto resp = response->add_tablet_schemas();

            resp->set_index_id(req.index_id());

            resp->set_enable_unique_key_merge_on_write(tablet->enable_unique_key_merge_on_write());

            tablet->tablet_schema()->to_schema_pb(resp->mutable_tablet_schema());

        }

        LoadStream* load_stream = nullptr;

        auto st = _exec_env->load_stream_mgr()->open_load_stream(request, load_stream);

        if (!st.ok()) {

            st.to_protobuf(response->mutable_status());

            return;

        }

        stream_options.handler = load_stream;

        stream_options.idle_timeout_ms = request->idle_timeout_ms();

        DBUG_EXECUTE_IF("PInternalServiceImpl.open_load_stream.set_idle_timeout",

                        { stream_options.idle_timeout_ms = 1; });

        StreamId streamid;

        if (brpc::StreamAccept(&streamid, *cntl, &stream_options) != 0) {

            st = Status::Cancelled("Fail to accept stream {}", streamid);

            st.to_protobuf(response->mutable_status());

            cntl->SetFailed(st.to_string());

            return;

        }

        VLOG_DEBUG << "get streamid =" << streamid;

        st.to_protobuf(response->mutable_status());

    });

    if (!ret) {

        offer_failed(response, done, _heavy_work_pool);

    }

}

void PInternalService::tablet_writer_add_block_by_http(google::protobuf::RpcController* controller,

                                                       const ::doris::PEmptyRequest* request,

                                                       PTabletWriterAddBlockResult* response,

                                                       google::protobuf::Closure* done) {

    PTabletWriterAddBlockRequest* new_request = new PTabletWriterAddBlockRequest();

    google::protobuf::Closure* new_done =

            new NewHttpClosure<PTabletWriterAddBlockRequest>(new_request, done);

    brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);

    Status st = attachment_extract_request_contain_block<PTabletWriterAddBlockRequest>(new_request,

                                                                                       cntl);

    if (st.ok()) {

        tablet_writer_add_block(controller, new_request, response, new_done);

    } else {

        st.to_protobuf(response->mutable_status());

    }

}

void PInternalService::tablet_writer_add_block(google::protobuf::RpcController* controller,

                                               const PTabletWriterAddBlockRequest* request,

                                               PTabletWriterAddBlockResult* response,

                                               google::protobuf::Closure* done) {

    int64_t submit_task_time_ns = MonotonicNanos();

    bool ret = _heavy_work_pool.try_offer([request, response, done, submit_task_time_ns, this]() {

        int64_t wait_execution_time_ns = MonotonicNanos() - submit_task_time_ns;

        brpc::ClosureGuard closure_guard(done);

        int64_t execution_time_ns = 0;

        {

            SCOPED_RAW_TIMER(&execution_time_ns);

            signal::set_signal_task_id(request->id());

            auto st = _exec_env->load_channel_mgr()->add_batch(*request, response);

            if (!st.ok()) {

                LOG(WARNING) << "tablet writer add block failed, message=" << st

                             << ", id=" << request->id() << ", index_id=" << request->index_id()

                             << ", sender_id=" << request->sender_id()

                             << ", backend id=" << request->backend_id();

            }

            st.to_protobuf(response->mutable_status());

        }

        response->set_execution_time_us(execution_time_ns / NANOS_PER_MICRO);

        response->set_wait_execution_time_us(wait_execution_time_ns / NANOS_PER_MICRO);

    });

    if (!ret) {

        offer_failed(response, done, _heavy_work_pool);

        return;

    }

}

void PInternalService::tablet_writer_cancel(google::protobuf::RpcController* controller,

                                            const PTabletWriterCancelRequest* request,

                                            PTabletWriterCancelResult* response,

                                            google::protobuf::Closure* done) {

    bool ret = _heavy_work_pool.try_offer([this, request, done]() {

        VLOG_RPC << "tablet writer cancel, id=" << request->id()

                 << ", index_id=" << request->index_id() << ", sender_id=" << request->sender_id();

        signal::set_signal_task_id(request->id());

        brpc::ClosureGuard closure_guard(done);

        auto st = _exec_env->load_channel_mgr()->cancel(*request);

        if (!st.ok()) {

            LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()

                         << ", index_id=" << request->index_id()

                         << ", sender_id=" << request->sender_id();

        }

    });

    if (!ret) {

        offer_failed(response, done, _heavy_work_pool);

        return;

    }

}

Status PInternalService::_exec_plan_fragment_impl(

        const std::string& ser_request, PFragmentRequestVersion version, bool compact,

        const std::function<void(RuntimeState*, Status*)>& cb) {

    // Sometimes the BE do not receive the first heartbeat message and it receives request from FE

    // If BE execute this fragment, it will core when it wants to get some property from master info.

    if (ExecEnv::GetInstance()->cluster_info() == nullptr) {

        return Status::InternalError(

                "Have not receive the first heartbeat message from master, not ready to provide "

                "service");

    }

    if (version == PFragmentRequestVersion::VERSION_1) {

        // VERSION_1 should be removed in v1.2

        TExecPlanFragmentParams t_request;

        {

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

            uint32_t len = ser_request.size();

            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));

        }

        if (cb) {

            return _exec_env->fragment_mgr()->exec_plan_fragment(

                    t_request, QuerySource::INTERNAL_FRONTEND, cb);

        } else {

            return _exec_env->fragment_mgr()->exec_plan_fragment(t_request,

                                                                 QuerySource::INTERNAL_FRONTEND);

        }

    } else if (version == PFragmentRequestVersion::VERSION_2) {

        TExecPlanFragmentParamsList t_request;

        {

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

            uint32_t len = ser_request.size();

            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));

        }

        const auto& fragment_list = t_request.paramsList;

        MonotonicStopWatch timer;

        timer.start();

        for (const TExecPlanFragmentParams& params : t_request.paramsList) {

            if (cb) {

                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(

                        params, QuerySource::INTERNAL_FRONTEND, cb));

            } else {

                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(

                        params, QuerySource::INTERNAL_FRONTEND));

            }

        }

        timer.stop();

        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;

        if (cost_secs > 5) {

            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",

                        fragment_list.size(), print_id(fragment_list.front().params.query_id),

                        cost_secs);

        }

        return Status::OK();

    } else if (version == PFragmentRequestVersion::VERSION_3) {

        TPipelineFragmentParamsList t_request;

        {

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

            uint32_t len = ser_request.size();

            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));

        }

        const auto& fragment_list = t_request.params_list;

        if (fragment_list.empty()) {

            return Status::InternalError("Invalid TPipelineFragmentParamsList!");

        }

        MonotonicStopWatch timer;

        timer.start();

        for (const TPipelineFragmentParams& fragment : fragment_list) {

            if (cb) {

                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(

                        fragment, QuerySource::INTERNAL_FRONTEND, cb));

            } else {

                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(

                        fragment, QuerySource::INTERNAL_FRONTEND));

            }

        }

        timer.stop();

        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;

        if (cost_secs > 5) {

            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",

                        fragment_list.size(), print_id(fragment_list.front().query_id), cost_secs);

        }

        return Status::OK();

    } else {

        return Status::InternalError("invalid version");

    }

}

void PInternalService::cancel_plan_fragment(google::protobuf::RpcController* /*controller*/,

                                            const PCancelPlanFragmentRequest* request,

                                            PCancelPlanFragmentResult* result,

                                            google::protobuf::Closure* done) {

    bool ret = _light_work_pool.try_offer([this, request, result, done]() {

        brpc::ClosureGuard closure_guard(done);

        TUniqueId tid;

        tid.__set_hi(request->finst_id().hi());

        tid.__set_lo(request->finst_id().lo());

        signal::set_signal_task_id(tid);

        Status st = Status::OK();

        const bool has_cancel_reason = request->has_cancel_reason();

        const bool has_cancel_status = request->has_cancel_status();

        // During upgrade only LIMIT_REACH is used, other reason is changed to internal error

        Status actual_cancel_status = Status::OK();

        // Convert PPlanFragmentCancelReason to Status

        if (has_cancel_status) {

            // If fe set cancel status, then it is new FE now, should use cancel status.

            actual_cancel_status = Status::create(request->cancel_status());

        } else if (has_cancel_reason) {

            // If fe not set cancel status, but set cancel reason, should convert cancel reason

            // to cancel status here.

            if (request->cancel_reason() == PPlanFragmentCancelReason::LIMIT_REACH) {

                actual_cancel_status = Status::Error<ErrorCode::LIMIT_REACH>("limit reach");

            } else {

                // Use cancel reason as error message

                actual_cancel_status = Status::InternalError(

                        PPlanFragmentCancelReason_Name(request->cancel_reason()));

            }

        } else {

            actual_cancel_status = Status::InternalError("unknown error");

        }

        TUniqueId query_id;

        query_id.__set_hi(request->query_id().hi());

        query_id.__set_lo(request->query_id().lo());

        LOG(INFO) << fmt::format("Cancel query {}, reason: {}", print_id(query_id),

                                 actual_cancel_status.to_string());

        _exec_env->fragment_mgr()->cancel_query(query_id, actual_cancel_status);

        // TODO: the logic seems useless, cancel only return Status::OK. remove it

        st.to_protobuf(result->mutable_status());

    });

    if (!ret) {

        offer_failed(result, done, _light_work_pool);

        return;

    }

}

void PInternalService::fetch_data(google::protobuf::RpcController* controller,

                                  const PFetchDataRequest* request, PFetchDataResult* result,

                                  google::protobuf::Closure* done) {

    // fetch_data is a light operation which will put a request rather than wait inplace when there's no data ready.

    // when there's data ready, use brpc to send. there's queue in brpc service. won't take it too long.

    auto* cntl = static_cast<brpc::Controller*>(controller);

    auto* ctx = new GetResultBatchCtx(cntl, result, done);

    _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);

}

void PInternalService::fetch_arrow_data(google::protobuf::RpcController* controller,

                                        const PFetchArrowDataRequest* request,

                                        PFetchArrowDataResult* result,

                                        google::protobuf::Closure* done) {

    bool ret = _arrow_flight_work_pool.try_offer([this, controller, request, result, done]() {

        brpc::ClosureGuard closure_guard(done);

        auto* cntl = static_cast<brpc::Controller*>(controller);

        auto* ctx = new GetArrowResultBatchCtx(cntl, result, done);

        _exec_env->result_mgr()->fetch_arrow_data(request->finst_id(), ctx);

    });

    if (!ret) {

        offer_failed(result, done, _arrow_flight_work_pool);

        return;

    }

}

void PInternalService::outfile_write_success(google::protobuf::RpcController* controller,

                                             const POutfileWriteSuccessRequest* request,

                                             POutfileWriteSuccessResult* result,

                                             google::protobuf::Closure* done) {

    bool ret = _heavy_work_pool.try_offer([request, result, done]() {

        VLOG_RPC << "outfile write success file";

        brpc::ClosureGuard closure_guard(done);

        TResultFileSink result_file_sink;

        Status st = Status::OK();

        {

            const uint8_t* buf = (const uint8_t*)(request->result_file_sink().data());

            uint32_t len = request->result_file_sink().size();

            st = deserialize_thrift_msg(buf, &len, false, &result_file_sink);

            if (!st.ok()) {

                LOG(WARNING) << "outfile write success file failed, errmsg = " << st;

                st.to_protobuf(result->mutable_status());

                return;

            }

        }

        TResultFileSinkOptions file_options = result_file_sink.file_options;

        std::stringstream ss;

        ss << file_options.file_path << file_options.success_file_name;

        std::string file_name = ss.str();

        if (result_file_sink.storage_backend_type == TStorageBackendType::LOCAL) {

            // For local file writer, the file_path is a local dir.

            // Here we do a simple security verification by checking whether the file exists.

            // Because the file path is currently arbitrarily specified by the user,

            // Doris is not responsible for ensuring the correctness of the path.

            // This is just to prevent overwriting the existing file.

            bool exists = true;

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

            if (!st.ok()) {

                LOG(WARNING) << "outfile write success filefailed, errmsg = " << st;

                st.to_protobuf(result->mutable_status());

                return;

            }

            if (exists) {

                st = Status::InternalError("File already exists: {}", file_name);

            }

            if (!st.ok()) {

                LOG(WARNING) << "outfile write success file failed, errmsg = " << st;

                st.to_protobuf(result->mutable_status());

                return;

            }

        }

        auto&& res = FileFactory::create_file_writer(

                FileFactory::convert_storage_type(result_file_sink.storage_backend_type),

                ExecEnv::GetInstance(), file_options.broker_addresses,

                file_options.broker_properties, file_name,

                {

                        .write_file_cache = false,

                        .sync_file_data = false,

                });

        using T = std::decay_t<decltype(res)>;

        if (!res.has_value()) [[unlikely]] {

            st = std::forward<T>(res).error();

            st.to_protobuf(result->mutable_status());

            return;

        }

        std::unique_ptr<doris::io::FileWriter> _file_writer_impl = std::forward<T>(res).value();

        // must write somthing because s3 file writer can not writer empty file

        st = _file_writer_impl->append({"success"});

        if (!st.ok()) {

            LOG(WARNING) << "outfile write success filefailed, errmsg=" << st;

            st.to_protobuf(result->mutable_status());

            return;

        }

        st = _file_writer_impl->close();

        if (!st.ok()) {

            LOG(WARNING) << "outfile write success filefailed, errmsg=" << st;

            st.to_protobuf(result->mutable_status());

            return;

        }

    });

    if (!ret) {

        offer_failed(result, done, _heavy_work_pool);

        return;

    }

}

void PInternalService::fetch_table_schema(google::protobuf::RpcController* controller,

                                          const PFetchTableSchemaRequest* request,

                                          PFetchTableSchemaResult* result,

                                          google::protobuf::Closure* done) {

    bool ret = _heavy_work_pool.try_offer([request, result, done]() {

        VLOG_RPC << "fetch table schema";

        brpc::ClosureGuard closure_guard(done);

        TFileScanRange file_scan_range;

        Status st = Status::OK();

        {

            const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());

            uint32_t len = request->file_scan_range().size();

            st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);

            if (!st.ok()) {

                LOG(WARNING) << "fetch table schema failed, errmsg=" << st;

                st.to_protobuf(result->mutable_status());

                return;

            }

        }

        if (file_scan_range.__isset.ranges == false) {

            st = Status::InternalError("can not get TFileRangeDesc.");

            st.to_protobuf(result->mutable_status());

            return;

        }

        if (file_scan_range.__isset.params == false) {

            st = Status::InternalError("can not get TFileScanRangeParams.");

            st.to_protobuf(result->mutable_status());

            return;

        }

        const TFileRangeDesc& range = file_scan_range.ranges.at(0);

        const TFileScanRangeParams& params = file_scan_range.params;

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

                MemTrackerLimiter::Type::OTHER,

                fmt::format("InternalService::fetch_table_schema:{}#{}", params.format_type,

                            params.file_type));

        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(mem_tracker);

        // make sure profile is desctructed after reader cause PrefetchBufferedReader

        // might asynchronouslly access the profile

        std::unique_ptr<RuntimeProfile> profile =

                std::make_unique<RuntimeProfile>("FetchTableSchema");

        std::unique_ptr<vectorized::GenericReader> reader(nullptr);

        io::IOContext io_ctx;

        io::FileCacheStatistics file_cache_statis;

        io_ctx.file_cache_stats = &file_cache_statis;

        switch (params.format_type) {

        case TFileFormatType::FORMAT_CSV_PLAIN:

        case TFileFormatType::FORMAT_CSV_GZ:

        case TFileFormatType::FORMAT_CSV_BZ2:

        case TFileFormatType::FORMAT_CSV_LZ4FRAME:

        case TFileFormatType::FORMAT_CSV_LZ4BLOCK:

        case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:

        case TFileFormatType::FORMAT_CSV_LZOP:

        case TFileFormatType::FORMAT_CSV_DEFLATE: {

            // file_slots is no use

            std::vector<SlotDescriptor*> file_slots;

            reader = vectorized::CsvReader::create_unique(profile.get(), params, range, file_slots,

                                                          &io_ctx);

            break;

        }

        case TFileFormatType::FORMAT_PARQUET: {

            reader = vectorized::ParquetReader::create_unique(params, range, &io_ctx, nullptr);

            break;

        }

        case TFileFormatType::FORMAT_ORC: {

            reader = vectorized::OrcReader::create_unique(params, range, "", &io_ctx);

            break;

        }

        case TFileFormatType::FORMAT_JSON: {

            std::vector<SlotDescriptor*> file_slots;

            reader = vectorized::NewJsonReader::create_unique(profile.get(), params, range,

                                                              file_slots, &io_ctx);

            break;

        }

        case TFileFormatType::FORMAT_AVRO: {

            // file_slots is no use

            std::vector<SlotDescriptor*> file_slots;

            reader = vectorized::AvroJNIReader::create_unique(profile.get(), params, range,

                                                              file_slots);

            st = ((vectorized::AvroJNIReader*)(reader.get()))->init_fetch_table_schema_reader();

            break;

        }

        default:

            st = Status::InternalError("Not supported file format in fetch table schema: {}",

                                       params.format_type);

            st.to_protobuf(result->mutable_status());

            return;

        }

        if (!st.ok()) {

            LOG(WARNING) << "failed to init reader, errmsg=" << st;

            st.to_protobuf(result->mutable_status());

            return;

        }

        std::vector<std::string> col_names;

        std::vector<TypeDescriptor> col_types;

        st = reader->get_parsed_schema(&col_names, &col_types);

        if (!st.ok()) {

            LOG(WARNING) << "fetch table schema failed, errmsg=" << st;

            st.to_protobuf(result->mutable_status());

            return;

        }

        result->set_column_nums(col_names.size());

        for (size_t idx = 0; idx < col_names.size(); ++idx) {

            result->add_column_names(col_names[idx]);

        }

        for (size_t idx = 0; idx < col_types.size(); ++idx) {

            PTypeDesc* type_desc = result->add_column_types();

            col_types[idx].to_protobuf(type_desc);

        }

        st.to_protobuf(result->mutable_status());

    });

    if (!ret) {

        offer_failed(result, done, _heavy_work_pool);

        return;

    }

}

void PInternalService::fetch_arrow_flight_schema(google::protobuf::RpcController* controller,

                                                 const PFetchArrowFlightSchemaRequest* request,

                                                 PFetchArrowFlightSchemaResult* result,

                                                 google::protobuf::Closure* done) {

    bool ret = _arrow_flight_work_pool.try_offer([request, result, done]() {

        brpc::ClosureGuard closure_guard(done);

        std::shared_ptr<arrow::Schema> schema;

        auto st = ExecEnv::GetInstance()->result_mgr()->find_arrow_schema(

                UniqueId(request->finst_id()).to_thrift(), &schema);

        if (!st.ok()) {

            LOG(WARNING) << "fetch arrow flight schema failed, errmsg=" << st;

            st.to_protobuf(result->mutable_status());

            return;

        }

        std::string schema_str;

        st = serialize_arrow_schema(&schema, &schema_str);

        if (st.ok()) {

            result->set_schema(std::move(schema_str));

            if (!config::public_host.empty()) {

                result->set_be_arrow_flight_ip(config::public_host);

            }

            if (config::arrow_flight_sql_proxy_port != -1) {