// 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 <algorithm>

#include <exception>

#include <memory>

#include <set>

#include <sstream>

#include <string>

#include <utility>

#include <vector>

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet_mgr.h"

#include "cloud/config.h"

#include "common/config.h"

#include "common/exception.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "common/metrics/metrics.h"

#include "common/signal_handler.h"

#include "common/status.h"

#include "core/block/block.h"

#include "core/data_type/data_type.h"

#include "exec/common/variant_util.h"

#include "exec/exchange/vdata_stream_mgr.h"

#include "exec/rowid_fetcher.h"

#include "exec/runtime_filter/runtime_filter_mgr.h"

#include "exec/sink/writer/varrow_flight_result_writer.h"

#include "exec/sink/writer/vmysql_result_writer.h"

#include "exprs/function/dictionary_factory.h"

#include "format/arrow/arrow_row_batch.h"

#include "format/csv/csv_reader.h"

#include "format/generic_reader.h"

#include "format/jni/jni_reader.h"

#include "format/json/new_json_reader.h"

#include "format/native/native_reader.h"

#include "format/orc/vorc_reader.h"

#include "format/parquet/vparquet_reader.h"

#include "format/text/text_reader.h"

#include "io/fs/local_file_system.h"

#include "io/fs/stream_load_pipe.h"

#include "io/io_common.h"

#include "load/channel/load_channel_mgr.h"

#include "load/channel/load_stream_mgr.h"

#include "load/delta_writer/delta_writer.h"

#include "load/group_commit/wal/wal_manager.h"

#include "load/routine_load/routine_load_task_executor.h"

#include "load/stream_load/new_load_stream_mgr.h"

#include "load/stream_load/stream_load_context.h"

#include "runtime/cache/result_cache.h"

#include "runtime/cdc_client_mgr.h"

#include "runtime/descriptors.h"

#include "runtime/exec_env.h"

#include "runtime/fold_constant_executor.h"

#include "runtime/fragment_mgr.h"

#include "runtime/query_context.h"

#include "runtime/result_block_buffer.h"

#include "runtime/result_buffer_mgr.h"

#include "runtime/runtime_profile.h"

#include "runtime/thread_context.h"

#include "runtime/workload_group/workload_group.h"

#include "runtime/workload_group/workload_group_manager.h"

#include "service/backend_options.h"

#include "service/point_query_executor.h"

#include "storage/data_dir.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/segment/column_reader.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_fwd.h"

#include "storage/tablet/tablet_manager.h"

#include "storage/tablet/tablet_schema.h"

#include "storage/txn/txn_manager.h"

#include "util/async_io.h"

#include "util/brpc_client_cache.h"

#include "util/brpc_closure.h"

#include "util/jdbc_utils.h"

#include "util/jsonb/serialize.h"

#include "util/md5.h"

#include "util/network_util.h"

#include "util/proto_util.h"

#include "util/stopwatch.hpp"

#include "util/string_util.h"

#include "util/thrift_util.h"

#include "util/time.h"

#include "util/uid_util.h"

namespace google {

namespace protobuf {

class RpcController;

} // namespace protobuf

} // namespace google

namespace doris {

#include "common/compile_check_avoid_begin.h"

using namespace ErrorCode;

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_queue_size, MetricUnit::NOUNIT);

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(peer_fetch_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(peer_fetch_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(peer_fetch_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(peer_fetch_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);

static bvar::LatencyRecorder g_process_remote_fetch_rowsets_latency("process_remote_fetch_rowsets");

bthread_key_t btls_key;

static void thread_context_deleter(void* d) {

    delete static_cast<ThreadContext*>(d);

}

static int32_t resolved_brpc_peer_fetch_pool_threads() {

    return config::brpc_peer_fetch_pool_threads != -1 ? config::brpc_peer_fetch_pool_threads

                                                      : std::max(64, CpuInfo::num_cores() * 2);

}

static int32_t resolved_brpc_peer_fetch_pool_max_queue_size() {

    return config::brpc_peer_fetch_pool_max_queue_size != -1

                   ? config::brpc_peer_fetch_pool_max_queue_size

                   : std::max(4096, CpuInfo::num_cores() * 128);

}

template <typename T>

concept CanCancel = requires(T* response) { response->mutable_status(); };

template <typename T>

void offer_failed(T* response, google::protobuf::Closure* done, const FifoThreadPool& pool) {

    brpc::ClosureGuard closure_guard(done);

    LOG(WARNING) << "fail to offer request to the work pool, pool=" << pool.get_info();

}

Unexecuted instantiation: _ZN5doris12offer_failedINS_25PTabletWriterCancelResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedINS_14PCacheResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedINS_17PFetchCacheResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

template <CanCancel T>

void offer_failed(T* response, google::protobuf::Closure* done, const FifoThreadPool& pool) {

    brpc::ClosureGuard closure_guard(done);

    // Should use status to generate protobuf message, because it will encoding Backend Info

    // into the error message and then we could know which backend's pool is full.

    Status st = Status::Error<TStatusCode::CANCELLED>(

            "fail to offer request to the work pool, pool={}", pool.get_info());

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

    LOG(WARNING) << "cancelled due to fail to offer request to the work pool, pool="

                 << pool.get_info();

}

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PTabletWriterOpenResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PExecPlanFragmentResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23POpenLoadStreamResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_27PTabletWriterAddBlockResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_25PCancelPlanFragmentResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_21PFetchArrowDataResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_26POutfileWriteSuccessResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PFetchTableSchemaResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_29PFetchArrowFlightSchemaResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_24PTabletKeyLookupResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_25PJdbcTestConnectionResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_20PFetchColIdsResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_26PFetchRemoteSchemaResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_12PProxyResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_20PMergeFilterResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PSendFilterSizeResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PSyncFilterSizeResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_22PPublishFilterResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_15PSendDataResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_13PCommitResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_15PRollbackResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_19PConstantExprResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_26PTransmitRecCTEBlockResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_20PRerunFragmentResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_20PResetGlobalRfResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_19PTransmitDataResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_24PCheckRPCChannelResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_24PResetRPCChannelResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_17PMultiGetResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_13PGlobResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_26PGroupCommitInsertResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_24PGetWalQueueSizeResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_22PGetBeResourceResponseEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

Unexecuted instantiation: _ZN5doris12offer_failedITkNS_9CanCancelENS_23PRequestCdcClientResultEEEvPT_PN6google8protobuf7ClosureERKNS_14WorkThreadPoolILb0EEE

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"),

          // peer fetch threadpool isolates fetch_peer_data from heavy load traffic to avoid peer reads starving imports.

          _peer_fetch_pool(resolved_brpc_peer_fetch_pool_threads(),

                           resolved_brpc_peer_fetch_pool_max_queue_size(), "brpc_peer_fetch"),

          // 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(peer_fetch_work_pool_queue_size,

                         [this]() { return _peer_fetch_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(peer_fetch_work_active_threads,

                         [this]() { return _peer_fetch_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(peer_fetch_work_pool_max_queue_size,

                         []() { return resolved_brpc_peer_fetch_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(peer_fetch_work_max_threads,

                         []() { return resolved_brpc_peer_fetch_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(peer_fetch_work_pool_queue_size);

    DEREGISTER_HOOK_METRIC(light_work_pool_queue_size);

    DEREGISTER_HOOK_METRIC(heavy_work_active_threads);

    DEREGISTER_HOOK_METRIC(peer_fetch_work_active_threads);

    DEREGISTER_HOOK_METRIC(light_work_active_threads);

    DEREGISTER_HOOK_METRIC(heavy_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(peer_fetch_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(light_work_pool_max_queue_size);

    DEREGISTER_HOOK_METRIC(heavy_work_max_threads);

    DEREGISTER_HOOK_METRIC(peer_fetch_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::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::SignalTaskIdKeeper keeper(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 (const std::exception& e) {

        st = Status::Error(ErrorCode::INTERNAL_ERROR, e.what());

    } 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::SignalTaskIdKeeper keeper(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();

        std::vector<BaseTabletSPtr> tablets;

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

            tablets.push_back(tablet);

        }

        if (!tablets.empty()) {

            auto* tablet_load_infos = response->mutable_tablet_load_rowset_num_infos();

            for (const auto& tablet : tablets) {

                BaseDeltaWriter::collect_tablet_load_rowset_num_info(tablet.get(),

                                                                     tablet_load_infos);

            }

        }

        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::SignalTaskIdKeeper keeper(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::SignalTaskIdKeeper keeper(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");

    }

    CHECK(version == PFragmentRequestVersion::VERSION_3)

            << "only support version 3, received " << version;

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

        // work for old version frontend

        if (!t_request.__isset.runtime_filter_info) {

            TRuntimeFilterInfo runtime_filter_info;

            auto local_param = fragment_list[0].local_params[0];

            if (local_param.__isset.runtime_filter_params) {

                runtime_filter_info.__set_runtime_filter_params(local_param.runtime_filter_params);

            }

            if (local_param.__isset.topn_filter_descs) {

                runtime_filter_info.__set_topn_filter_descs(local_param.topn_filter_descs);

            }

            t_request.__set_runtime_filter_info(runtime_filter_info);

        }

        for (const TPipelineFragmentParams& fragment : fragment_list) {

            if (cb) {

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

                        fragment, QuerySource::INTERNAL_FRONTEND, cb, t_request));

            } else {

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

                        fragment, QuerySource::INTERNAL_FRONTEND, t_request));

            }

        }

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

        signal::SignalTaskIdKeeper keeper(request->finst_id());

        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<false>(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 ctx = GetResultBatchCtx::create_shared(result, done);

    TUniqueId unique_id = UniqueId(request->finst_id()).to_thrift(); // query_id or instance_id

    std::shared_ptr<MySQLResultBlockBuffer> buffer;

    Status st = ExecEnv::GetInstance()->result_mgr()->find_buffer(unique_id, buffer);

    if (!st.ok()) {

        LOG(WARNING) << "Result buffer not found! finst ID: " << print_id(unique_id);

        return;

    }

    if (st = buffer->get_batch(ctx); !st.ok()) {

        LOG(WARNING) << "fetch_data failed: " << st.to_string();

    }

}

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([request, result, done]() {

        auto ctx = GetArrowResultBatchCtx::create_shared(result, done);

        TUniqueId unique_id = UniqueId(request->finst_id()).to_thrift(); // query_id or instance_id

        std::shared_ptr<ArrowFlightResultBlockBuffer> arrow_buffer;

        auto st = ExecEnv::GetInstance()->result_mgr()->find_buffer(unique_id, arrow_buffer);

        if (!st.ok()) {

            LOG(WARNING) << "Result buffer not found! Query ID: " << print_id(unique_id);

            return;

        }

        if (st = arrow_buffer->get_batch(ctx); !st.ok()) {

            LOG(WARNING) << "fetch_arrow_data failed: " << st.to_string();

        }

    });

    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 file_type_res =

                FileFactory::convert_storage_type(result_file_sink.storage_backend_type);

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

            st = std::move(file_type_res).error();

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

            LOG(WARNING) << "encounter unkonw type=" << result_file_sink.storage_backend_type

                         << ", st=" << st;

            return;

        }

        auto&& res = FileFactory::create_file_writer(file_type_res.value(), 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_ATTACH_TASK(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<GenericReader> reader(nullptr);

        auto io_ctx = std::make_shared<io::IOContext>();

        auto file_cache_statis = std::make_shared<io::FileCacheStatistics>();

        auto file_reader_stats = std::make_shared<io::FileReaderStats>();

        io_ctx->file_cache_stats = file_cache_statis.get();

        io_ctx->file_reader_stats = file_reader_stats.get();

        constexpr size_t fetch_schema_batch_size = 4064;

        // file_slots is no use, but the lifetime should be longer than reader

        std::vector<SlotDescriptor*> file_slots;

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

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

                                              file_slots, fetch_schema_batch_size, io_ctx.get(),

                                              io_ctx);

            break;

        }

        case TFileFormatType::FORMAT_TEXT: {

            reader = TextReader::create_unique(nullptr, profile.get(), nullptr, params, range,

                                               file_slots, fetch_schema_batch_size, io_ctx.get());

            break;

        }

        case TFileFormatType::FORMAT_PARQUET: {

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

            break;

        }

        case TFileFormatType::FORMAT_ORC: {

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

            break;

        }

        case TFileFormatType::FORMAT_NATIVE: {

            reader = NativeReader::create_unique(profile.get(), params, range, io_ctx.get(),

                                                 nullptr);

            break;