Coverage Report

Created: 2025-03-13 18:54

/root/doris/be/src/service/internal_service.cpp
Line
Count
Source (jump to first uncovered line)
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "service/internal_service.h"
19
20
#include <assert.h>
21
#include <brpc/closure_guard.h>
22
#include <brpc/controller.h>
23
#include <bthread/bthread.h>
24
#include <bthread/types.h>
25
#include <butil/errno.h>
26
#include <butil/iobuf.h>
27
#include <fcntl.h>
28
#include <fmt/core.h>
29
#include <gen_cpp/DataSinks_types.h>
30
#include <gen_cpp/MasterService_types.h>
31
#include <gen_cpp/PaloInternalService_types.h>
32
#include <gen_cpp/PlanNodes_types.h>
33
#include <gen_cpp/Status_types.h>
34
#include <gen_cpp/Types_types.h>
35
#include <gen_cpp/internal_service.pb.h>
36
#include <gen_cpp/olap_file.pb.h>
37
#include <gen_cpp/segment_v2.pb.h>
38
#include <gen_cpp/types.pb.h>
39
#include <google/protobuf/stubs/callback.h>
40
#include <stddef.h>
41
#include <stdint.h>
42
#include <sys/stat.h>
43
#include <vec/exec/vjdbc_connector.h>
44
#include <vec/sink/varrow_flight_result_writer.h>
45
46
#include <algorithm>
47
#include <exception>
48
#include <filesystem>
49
#include <memory>
50
#include <set>
51
#include <sstream>
52
#include <string>
53
#include <unordered_map>
54
#include <utility>
55
#include <vector>
56
57
#include "common/config.h"
58
#include "common/consts.h"
59
#include "common/exception.h"
60
#include "common/logging.h"
61
#include "common/signal_handler.h"
62
#include "common/status.h"
63
#include "exec/rowid_fetcher.h"
64
#include "gen_cpp/BackendService.h"
65
#include "gen_cpp/PaloInternalService_types.h"
66
#include "gen_cpp/internal_service.pb.h"
67
#include "gutil/integral_types.h"
68
#include "http/http_client.h"
69
#include "io/fs/local_file_system.h"
70
#include "io/fs/stream_load_pipe.h"
71
#include "io/io_common.h"
72
#include "olap/data_dir.h"
73
#include "olap/olap_common.h"
74
#include "olap/olap_define.h"
75
#include "olap/rowset/beta_rowset.h"
76
#include "olap/rowset/rowset.h"
77
#include "olap/rowset/rowset_factory.h"
78
#include "olap/rowset/rowset_meta.h"
79
#include "olap/rowset/segment_v2/column_reader.h"
80
#include "olap/rowset/segment_v2/common.h"
81
#include "olap/rowset/segment_v2/inverted_index_desc.h"
82
#include "olap/rowset/segment_v2/segment.h"
83
#include "olap/rowset/segment_v2/segment_iterator.h"
84
#include "olap/segment_loader.h"
85
#include "olap/storage_engine.h"
86
#include "olap/tablet.h"
87
#include "olap/tablet_fwd.h"
88
#include "olap/tablet_manager.h"
89
#include "olap/tablet_schema.h"
90
#include "olap/txn_manager.h"
91
#include "olap/utils.h"
92
#include "olap/wal/wal_manager.h"
93
#include "runtime/cache/result_cache.h"
94
#include "runtime/define_primitive_type.h"
95
#include "runtime/descriptors.h"
96
#include "runtime/exec_env.h"
97
#include "runtime/fold_constant_executor.h"
98
#include "runtime/fragment_mgr.h"
99
#include "runtime/load_channel_mgr.h"
100
#include "runtime/load_stream_mgr.h"
101
#include "runtime/result_block_buffer.h"
102
#include "runtime/result_buffer_mgr.h"
103
#include "runtime/routine_load/routine_load_task_executor.h"
104
#include "runtime/stream_load/new_load_stream_mgr.h"
105
#include "runtime/stream_load/stream_load_context.h"
106
#include "runtime/thread_context.h"
107
#include "runtime/types.h"
108
#include "service/backend_options.h"
109
#include "service/point_query_executor.h"
110
#include "util/arrow/row_batch.h"
111
#include "util/async_io.h"
112
#include "util/brpc_client_cache.h"
113
#include "util/brpc_closure.h"
114
#include "util/doris_metrics.h"
115
#include "util/md5.h"
116
#include "util/metrics.h"
117
#include "util/network_util.h"
118
#include "util/proto_util.h"
119
#include "util/runtime_profile.h"
120
#include "util/stopwatch.hpp"
121
#include "util/string_util.h"
122
#include "util/thrift_util.h"
123
#include "util/time.h"
124
#include "util/uid_util.h"
125
#include "vec/columns/column.h"
126
#include "vec/columns/column_string.h"
127
#include "vec/common/schema_util.h"
128
#include "vec/core/block.h"
129
#include "vec/core/column_with_type_and_name.h"
130
#include "vec/data_types/data_type.h"
131
#include "vec/exec/format/avro//avro_jni_reader.h"
132
#include "vec/exec/format/csv/csv_reader.h"
133
#include "vec/exec/format/generic_reader.h"
134
#include "vec/exec/format/json/new_json_reader.h"
135
#include "vec/exec/format/orc/vorc_reader.h"
136
#include "vec/exec/format/parquet/vparquet_reader.h"
137
#include "vec/jsonb/serialize.h"
138
#include "vec/runtime/vdata_stream_mgr.h"
139
#include "vec/sink/vmysql_result_writer.h"
140
141
namespace google {
142
namespace protobuf {
143
class RpcController;
144
} // namespace protobuf
145
} // namespace google
146
147
namespace doris {
148
using namespace ErrorCode;
149
150
const uint32_t DOWNLOAD_FILE_MAX_RETRY = 3;
151
152
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_queue_size, MetricUnit::NOUNIT);
153
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_pool_queue_size, MetricUnit::NOUNIT);
154
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_active_threads, MetricUnit::NOUNIT);
155
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_active_threads, MetricUnit::NOUNIT);
156
157
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_pool_max_queue_size, MetricUnit::NOUNIT);
158
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_pool_max_queue_size, MetricUnit::NOUNIT);
159
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(heavy_work_max_threads, MetricUnit::NOUNIT);
160
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(light_work_max_threads, MetricUnit::NOUNIT);
161
162
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_pool_queue_size, MetricUnit::NOUNIT);
163
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_active_threads, MetricUnit::NOUNIT);
164
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_pool_max_queue_size, MetricUnit::NOUNIT);
165
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(arrow_flight_work_max_threads, MetricUnit::NOUNIT);
166
167
bthread_key_t btls_key;
168
169
0
static void thread_context_deleter(void* d) {
170
0
    delete static_cast<ThreadContext*>(d);
171
0
}
172
173
template <typename T>
174
class NewHttpClosure : public ::google::protobuf::Closure {
175
public:
176
    NewHttpClosure(google::protobuf::Closure* done) : _done(done) {}
177
0
    NewHttpClosure(T* request, google::protobuf::Closure* done) : _request(request), _done(done) {}
Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_28PTabletWriterAddBlockRequestEEC2EPS1_PN6google8protobuf7ClosureE
Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_19PTransmitDataParamsEEC2EPS1_PN6google8protobuf7ClosureE
178
179
0
    void Run() override {
180
0
        if (_request != nullptr) {
181
0
            delete _request;
182
0
            _request = nullptr;
183
0
        }
184
0
        if (_done != nullptr) {
185
0
            _done->Run();
186
0
        }
187
0
        delete this;
188
0
    }
Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_28PTabletWriterAddBlockRequestEE3RunEv
Unexecuted instantiation: _ZN5doris14NewHttpClosureINS_19PTransmitDataParamsEE3RunEv
189
190
private:
191
    T* _request = nullptr;
192
    google::protobuf::Closure* _done = nullptr;
193
};
194
195
PInternalService::PInternalService(ExecEnv* exec_env)
196
        : _exec_env(exec_env),
197
          _heavy_work_pool(config::brpc_heavy_work_pool_threads != -1
198
                                   ? config::brpc_heavy_work_pool_threads
199
                                   : std::max(128, CpuInfo::num_cores() * 4),
200
                           config::brpc_heavy_work_pool_max_queue_size != -1
201
                                   ? config::brpc_heavy_work_pool_max_queue_size
202
                                   : std::max(10240, CpuInfo::num_cores() * 320),
203
                           "brpc_heavy"),
204
          _light_work_pool(config::brpc_light_work_pool_threads != -1
205
                                   ? config::brpc_light_work_pool_threads
206
                                   : std::max(128, CpuInfo::num_cores() * 4),
207
                           config::brpc_light_work_pool_max_queue_size != -1
208
                                   ? config::brpc_light_work_pool_max_queue_size
209
                                   : std::max(10240, CpuInfo::num_cores() * 320),
210
                           "brpc_light"),
211
          _arrow_flight_work_pool(config::brpc_arrow_flight_work_pool_threads != -1
212
                                          ? config::brpc_arrow_flight_work_pool_threads
213
                                          : std::max(512, CpuInfo::num_cores() * 2),
214
                                  config::brpc_arrow_flight_work_pool_max_queue_size != -1
215
                                          ? config::brpc_arrow_flight_work_pool_max_queue_size
216
                                          : std::max(20480, CpuInfo::num_cores() * 640),
217
0
                                  "brpc_arrow_flight") {
218
0
    REGISTER_HOOK_METRIC(heavy_work_pool_queue_size,
219
0
                         [this]() { return _heavy_work_pool.get_queue_size(); });
220
0
    REGISTER_HOOK_METRIC(light_work_pool_queue_size,
221
0
                         [this]() { return _light_work_pool.get_queue_size(); });
222
0
    REGISTER_HOOK_METRIC(heavy_work_active_threads,
223
0
                         [this]() { return _heavy_work_pool.get_active_threads(); });
224
0
    REGISTER_HOOK_METRIC(light_work_active_threads,
225
0
                         [this]() { return _light_work_pool.get_active_threads(); });
226
227
0
    REGISTER_HOOK_METRIC(heavy_work_pool_max_queue_size,
228
0
                         []() { return config::brpc_heavy_work_pool_max_queue_size; });
229
0
    REGISTER_HOOK_METRIC(light_work_pool_max_queue_size,
230
0
                         []() { return config::brpc_light_work_pool_max_queue_size; });
231
0
    REGISTER_HOOK_METRIC(heavy_work_max_threads,
232
0
                         []() { return config::brpc_heavy_work_pool_threads; });
233
0
    REGISTER_HOOK_METRIC(light_work_max_threads,
234
0
                         []() { return config::brpc_light_work_pool_threads; });
235
236
0
    REGISTER_HOOK_METRIC(arrow_flight_work_pool_queue_size,
237
0
                         [this]() { return _arrow_flight_work_pool.get_queue_size(); });
238
0
    REGISTER_HOOK_METRIC(arrow_flight_work_active_threads,
239
0
                         [this]() { return _arrow_flight_work_pool.get_active_threads(); });
240
0
    REGISTER_HOOK_METRIC(arrow_flight_work_pool_max_queue_size,
241
0
                         []() { return config::brpc_arrow_flight_work_pool_max_queue_size; });
242
0
    REGISTER_HOOK_METRIC(arrow_flight_work_max_threads,
243
0
                         []() { return config::brpc_arrow_flight_work_pool_threads; });
244
245
0
    _exec_env->load_stream_mgr()->set_heavy_work_pool(&_heavy_work_pool);
246
0
    _exec_env->load_stream_mgr()->set_light_work_pool(&_light_work_pool);
247
248
0
    CHECK_EQ(0, bthread_key_create(&btls_key, thread_context_deleter));
249
0
    CHECK_EQ(0, bthread_key_create(&AsyncIO::btls_io_ctx_key, AsyncIO::io_ctx_key_deleter));
250
0
}
251
252
PInternalServiceImpl::PInternalServiceImpl(StorageEngine& engine, ExecEnv* exec_env)
253
0
        : PInternalService(exec_env), _engine(engine) {}
254
255
0
PInternalServiceImpl::~PInternalServiceImpl() = default;
256
257
0
PInternalService::~PInternalService() {
258
0
    DEREGISTER_HOOK_METRIC(heavy_work_pool_queue_size);
259
0
    DEREGISTER_HOOK_METRIC(light_work_pool_queue_size);
260
0
    DEREGISTER_HOOK_METRIC(heavy_work_active_threads);
261
0
    DEREGISTER_HOOK_METRIC(light_work_active_threads);
262
263
0
    DEREGISTER_HOOK_METRIC(heavy_work_pool_max_queue_size);
264
0
    DEREGISTER_HOOK_METRIC(light_work_pool_max_queue_size);
265
0
    DEREGISTER_HOOK_METRIC(heavy_work_max_threads);
266
0
    DEREGISTER_HOOK_METRIC(light_work_max_threads);
267
268
0
    DEREGISTER_HOOK_METRIC(arrow_flight_work_pool_queue_size);
269
0
    DEREGISTER_HOOK_METRIC(arrow_flight_work_active_threads);
270
0
    DEREGISTER_HOOK_METRIC(arrow_flight_work_pool_max_queue_size);
271
0
    DEREGISTER_HOOK_METRIC(arrow_flight_work_max_threads);
272
273
0
    CHECK_EQ(0, bthread_key_delete(btls_key));
274
0
    CHECK_EQ(0, bthread_key_delete(AsyncIO::btls_io_ctx_key));
275
0
}
276
277
void PInternalService::tablet_writer_open(google::protobuf::RpcController* controller,
278
                                          const PTabletWriterOpenRequest* request,
279
                                          PTabletWriterOpenResult* response,
280
0
                                          google::protobuf::Closure* done) {
281
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
282
0
        VLOG_RPC << "tablet writer open, id=" << request->id()
283
0
                 << ", index_id=" << request->index_id() << ", txn_id=" << request->txn_id();
284
0
        signal::set_signal_task_id(request->id());
285
0
        brpc::ClosureGuard closure_guard(done);
286
0
        auto st = _exec_env->load_channel_mgr()->open(*request);
287
0
        if (!st.ok()) {
288
0
            LOG(WARNING) << "load channel open failed, message=" << st << ", id=" << request->id()
289
0
                         << ", index_id=" << request->index_id()
290
0
                         << ", txn_id=" << request->txn_id();
291
0
        }
292
0
        st.to_protobuf(response->mutable_status());
293
0
    });
294
0
    if (!ret) {
295
0
        offer_failed(response, done, _light_work_pool);
296
0
        return;
297
0
    }
298
0
}
299
300
void PInternalService::exec_plan_fragment(google::protobuf::RpcController* controller,
301
                                          const PExecPlanFragmentRequest* request,
302
                                          PExecPlanFragmentResult* response,
303
0
                                          google::protobuf::Closure* done) {
304
0
    timeval tv {};
305
0
    gettimeofday(&tv, nullptr);
306
0
    response->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);
307
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
308
0
        _exec_plan_fragment_in_pthread(controller, request, response, done);
309
0
    });
310
0
    if (!ret) {
311
0
        offer_failed(response, done, _light_work_pool);
312
0
        return;
313
0
    }
314
0
}
315
316
void PInternalService::_exec_plan_fragment_in_pthread(google::protobuf::RpcController* controller,
317
                                                      const PExecPlanFragmentRequest* request,
318
                                                      PExecPlanFragmentResult* response,
319
0
                                                      google::protobuf::Closure* done) {
320
0
    timeval tv1 {};
321
0
    gettimeofday(&tv1, nullptr);
322
0
    response->set_execution_time(tv1.tv_sec * 1000LL + tv1.tv_usec / 1000);
323
0
    brpc::ClosureGuard closure_guard(done);
324
0
    auto st = Status::OK();
325
0
    bool compact = request->has_compact() ? request->compact() : false;
326
0
    PFragmentRequestVersion version =
327
0
            request->has_version() ? request->version() : PFragmentRequestVersion::VERSION_1;
328
0
    try {
329
0
        st = _exec_plan_fragment_impl(request->request(), version, compact);
330
0
    } catch (const Exception& e) {
331
0
        st = e.to_status();
332
0
    } catch (...) {
333
0
        st = Status::Error(ErrorCode::INTERNAL_ERROR,
334
0
                           "_exec_plan_fragment_impl meet unknown error");
335
0
    }
336
0
    if (!st.ok()) {
337
0
        LOG(WARNING) << "exec plan fragment failed, errmsg=" << st;
338
0
    }
339
0
    st.to_protobuf(response->mutable_status());
340
0
    timeval tv2 {};
341
0
    gettimeofday(&tv2, nullptr);
342
0
    response->set_execution_done_time(tv2.tv_sec * 1000LL + tv2.tv_usec / 1000);
343
0
}
344
345
void PInternalService::exec_plan_fragment_prepare(google::protobuf::RpcController* controller,
346
                                                  const PExecPlanFragmentRequest* request,
347
                                                  PExecPlanFragmentResult* response,
348
0
                                                  google::protobuf::Closure* done) {
349
0
    timeval tv {};
350
0
    gettimeofday(&tv, nullptr);
351
0
    response->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);
352
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
353
0
        _exec_plan_fragment_in_pthread(controller, request, response, done);
354
0
    });
355
0
    if (!ret) {
356
0
        offer_failed(response, done, _light_work_pool);
357
0
        return;
358
0
    }
359
0
}
360
361
void PInternalService::exec_plan_fragment_start(google::protobuf::RpcController* /*controller*/,
362
                                                const PExecPlanFragmentStartRequest* request,
363
                                                PExecPlanFragmentResult* result,
364
0
                                                google::protobuf::Closure* done) {
365
0
    timeval tv {};
366
0
    gettimeofday(&tv, nullptr);
367
0
    result->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);
368
0
    bool ret = _light_work_pool.try_offer([this, request, result, done]() {
369
0
        timeval tv1 {};
370
0
        gettimeofday(&tv1, nullptr);
371
0
        result->set_execution_time(tv1.tv_sec * 1000LL + tv1.tv_usec / 1000);
372
0
        brpc::ClosureGuard closure_guard(done);
373
0
        auto st = _exec_env->fragment_mgr()->start_query_execution(request);
374
0
        st.to_protobuf(result->mutable_status());
375
0
        timeval tv2 {};
376
0
        gettimeofday(&tv2, nullptr);
377
0
        result->set_execution_done_time(tv2.tv_sec * 1000LL + tv2.tv_usec / 1000);
378
0
    });
379
0
    if (!ret) {
380
0
        offer_failed(result, done, _light_work_pool);
381
0
        return;
382
0
    }
383
0
}
384
385
void PInternalService::open_load_stream(google::protobuf::RpcController* controller,
386
                                        const POpenLoadStreamRequest* request,
387
                                        POpenLoadStreamResponse* response,
388
0
                                        google::protobuf::Closure* done) {
389
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
390
0
        signal::set_signal_task_id(request->load_id());
391
0
        brpc::ClosureGuard done_guard(done);
392
0
        brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
393
0
        brpc::StreamOptions stream_options;
394
395
0
        LOG(INFO) << "open load stream, load_id=" << request->load_id()
396
0
                  << ", src_id=" << request->src_id();
397
398
0
        for (const auto& req : request->tablets()) {
399
0
            BaseTabletSPtr tablet;
400
0
            if (auto res = ExecEnv::get_tablet(req.tablet_id()); !res.has_value()) [[unlikely]] {
401
0
                auto st = std::move(res).error();
402
0
                st.to_protobuf(response->mutable_status());
403
0
                cntl->SetFailed(st.to_string());
404
0
                return;
405
0
            } else {
406
0
                tablet = std::move(res).value();
407
0
            }
408
0
            auto resp = response->add_tablet_schemas();
409
0
            resp->set_index_id(req.index_id());
410
0
            resp->set_enable_unique_key_merge_on_write(tablet->enable_unique_key_merge_on_write());
411
0
            tablet->tablet_schema()->to_schema_pb(resp->mutable_tablet_schema());
412
0
        }
413
414
0
        LoadStream* load_stream = nullptr;
415
0
        auto st = _exec_env->load_stream_mgr()->open_load_stream(request, load_stream);
416
0
        if (!st.ok()) {
417
0
            st.to_protobuf(response->mutable_status());
418
0
            return;
419
0
        }
420
421
0
        stream_options.handler = load_stream;
422
0
        stream_options.idle_timeout_ms = request->idle_timeout_ms();
423
0
        DBUG_EXECUTE_IF("PInternalServiceImpl.open_load_stream.set_idle_timeout",
424
0
                        { stream_options.idle_timeout_ms = 1; });
425
426
0
        StreamId streamid;
427
0
        if (brpc::StreamAccept(&streamid, *cntl, &stream_options) != 0) {
428
0
            st = Status::Cancelled("Fail to accept stream {}", streamid);
429
0
            st.to_protobuf(response->mutable_status());
430
0
            cntl->SetFailed(st.to_string());
431
0
            return;
432
0
        }
433
434
0
        VLOG_DEBUG << "get streamid =" << streamid;
435
0
        st.to_protobuf(response->mutable_status());
436
0
    });
437
0
    if (!ret) {
438
0
        offer_failed(response, done, _light_work_pool);
439
0
    }
440
0
}
441
442
void PInternalService::tablet_writer_add_block_by_http(google::protobuf::RpcController* controller,
443
                                                       const ::doris::PEmptyRequest* request,
444
                                                       PTabletWriterAddBlockResult* response,
445
0
                                                       google::protobuf::Closure* done) {
446
0
    PTabletWriterAddBlockRequest* new_request = new PTabletWriterAddBlockRequest();
447
0
    google::protobuf::Closure* new_done =
448
0
            new NewHttpClosure<PTabletWriterAddBlockRequest>(new_request, done);
449
0
    brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
450
0
    Status st = attachment_extract_request_contain_block<PTabletWriterAddBlockRequest>(new_request,
451
0
                                                                                       cntl);
452
0
    if (st.ok()) {
453
0
        tablet_writer_add_block(controller, new_request, response, new_done);
454
0
    } else {
455
0
        st.to_protobuf(response->mutable_status());
456
0
    }
457
0
}
458
459
void PInternalService::tablet_writer_add_block(google::protobuf::RpcController* controller,
460
                                               const PTabletWriterAddBlockRequest* request,
461
                                               PTabletWriterAddBlockResult* response,
462
0
                                               google::protobuf::Closure* done) {
463
0
    int64_t submit_task_time_ns = MonotonicNanos();
464
0
    bool ret = _heavy_work_pool.try_offer([request, response, done, submit_task_time_ns, this]() {
465
0
        int64_t wait_execution_time_ns = MonotonicNanos() - submit_task_time_ns;
466
0
        brpc::ClosureGuard closure_guard(done);
467
0
        int64_t execution_time_ns = 0;
468
0
        {
469
0
            SCOPED_RAW_TIMER(&execution_time_ns);
470
0
            signal::set_signal_task_id(request->id());
471
0
            auto st = _exec_env->load_channel_mgr()->add_batch(*request, response);
472
0
            if (!st.ok()) {
473
0
                LOG(WARNING) << "tablet writer add block failed, message=" << st
474
0
                             << ", id=" << request->id() << ", index_id=" << request->index_id()
475
0
                             << ", sender_id=" << request->sender_id()
476
0
                             << ", backend id=" << request->backend_id();
477
0
            }
478
0
            st.to_protobuf(response->mutable_status());
479
0
        }
480
0
        response->set_execution_time_us(execution_time_ns / NANOS_PER_MICRO);
481
0
        response->set_wait_execution_time_us(wait_execution_time_ns / NANOS_PER_MICRO);
482
0
    });
483
0
    if (!ret) {
484
0
        offer_failed(response, done, _heavy_work_pool);
485
0
        return;
486
0
    }
487
0
}
488
489
void PInternalService::tablet_writer_cancel(google::protobuf::RpcController* controller,
490
                                            const PTabletWriterCancelRequest* request,
491
                                            PTabletWriterCancelResult* response,
492
0
                                            google::protobuf::Closure* done) {
493
0
    bool ret = _light_work_pool.try_offer([this, request, done]() {
494
0
        VLOG_RPC << "tablet writer cancel, id=" << request->id()
495
0
                 << ", index_id=" << request->index_id() << ", sender_id=" << request->sender_id();
496
0
        signal::set_signal_task_id(request->id());
497
0
        brpc::ClosureGuard closure_guard(done);
498
0
        auto st = _exec_env->load_channel_mgr()->cancel(*request);
499
0
        if (!st.ok()) {
500
0
            LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
501
0
                         << ", index_id=" << request->index_id()
502
0
                         << ", sender_id=" << request->sender_id();
503
0
        }
504
0
    });
505
0
    if (!ret) {
506
0
        offer_failed(response, done, _light_work_pool);
507
0
        return;
508
0
    }
509
0
}
510
511
Status PInternalService::_exec_plan_fragment_impl(
512
        const std::string& ser_request, PFragmentRequestVersion version, bool compact,
513
0
        const std::function<void(RuntimeState*, Status*)>& cb) {
514
    // Sometimes the BE do not receive the first heartbeat message and it receives request from FE
515
    // If BE execute this fragment, it will core when it wants to get some property from master info.
516
0
    if (ExecEnv::GetInstance()->cluster_info() == nullptr) {
517
0
        return Status::InternalError(
518
0
                "Have not receive the first heartbeat message from master, not ready to provide "
519
0
                "service");
520
0
    }
521
0
    if (version == PFragmentRequestVersion::VERSION_1) {
522
        // VERSION_1 should be removed in v1.2
523
0
        TExecPlanFragmentParams t_request;
524
0
        {
525
0
            const uint8_t* buf = (const uint8_t*)ser_request.data();
526
0
            uint32_t len = ser_request.size();
527
0
            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));
528
0
        }
529
0
        if (cb) {
530
0
            return _exec_env->fragment_mgr()->exec_plan_fragment(
531
0
                    t_request, QuerySource::INTERNAL_FRONTEND, cb);
532
0
        } else {
533
0
            return _exec_env->fragment_mgr()->exec_plan_fragment(t_request,
534
0
                                                                 QuerySource::INTERNAL_FRONTEND);
535
0
        }
536
0
    } else if (version == PFragmentRequestVersion::VERSION_2) {
537
0
        TExecPlanFragmentParamsList t_request;
538
0
        {
539
0
            const uint8_t* buf = (const uint8_t*)ser_request.data();
540
0
            uint32_t len = ser_request.size();
541
0
            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));
542
0
        }
543
0
        const auto& fragment_list = t_request.paramsList;
544
0
        MonotonicStopWatch timer;
545
0
        timer.start();
546
547
0
        for (const TExecPlanFragmentParams& params : t_request.paramsList) {
548
0
            if (cb) {
549
0
                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(
550
0
                        params, QuerySource::INTERNAL_FRONTEND, cb));
551
0
            } else {
552
0
                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(
553
0
                        params, QuerySource::INTERNAL_FRONTEND));
554
0
            }
555
0
        }
556
557
0
        timer.stop();
558
0
        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;
559
0
        if (cost_secs > 5) {
560
0
            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",
561
0
                        fragment_list.size(), print_id(fragment_list.front().params.query_id),
562
0
                        cost_secs);
563
0
        }
564
565
0
        return Status::OK();
566
0
    } else if (version == PFragmentRequestVersion::VERSION_3) {
567
0
        TPipelineFragmentParamsList t_request;
568
0
        {
569
0
            const uint8_t* buf = (const uint8_t*)ser_request.data();
570
0
            uint32_t len = ser_request.size();
571
0
            RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, compact, &t_request));
572
0
        }
573
574
0
        const auto& fragment_list = t_request.params_list;
575
0
        if (fragment_list.empty()) {
576
0
            return Status::InternalError("Invalid TPipelineFragmentParamsList!");
577
0
        }
578
0
        MonotonicStopWatch timer;
579
0
        timer.start();
580
0
        for (const TPipelineFragmentParams& fragment : fragment_list) {
581
0
            if (cb) {
582
0
                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(
583
0
                        fragment, QuerySource::INTERNAL_FRONTEND, cb));
584
0
            } else {
585
0
                RETURN_IF_ERROR(_exec_env->fragment_mgr()->exec_plan_fragment(
586
0
                        fragment, QuerySource::INTERNAL_FRONTEND));
587
0
            }
588
0
        }
589
0
        timer.stop();
590
0
        double cost_secs = static_cast<double>(timer.elapsed_time()) / 1000000000ULL;
591
0
        if (cost_secs > 5) {
592
0
            LOG_WARNING("Prepare {} fragments of query {} costs {} seconds, it costs too much",
593
0
                        fragment_list.size(), print_id(fragment_list.front().query_id), cost_secs);
594
0
        }
595
596
0
        return Status::OK();
597
0
    } else {
598
0
        return Status::InternalError("invalid version");
599
0
    }
600
0
}
601
602
void PInternalService::cancel_plan_fragment(google::protobuf::RpcController* /*controller*/,
603
                                            const PCancelPlanFragmentRequest* request,
604
                                            PCancelPlanFragmentResult* result,
605
0
                                            google::protobuf::Closure* done) {
606
0
    bool ret = _light_work_pool.try_offer([this, request, result, done]() {
607
0
        brpc::ClosureGuard closure_guard(done);
608
0
        TUniqueId tid;
609
0
        tid.__set_hi(request->finst_id().hi());
610
0
        tid.__set_lo(request->finst_id().lo());
611
0
        signal::set_signal_task_id(tid);
612
0
        Status st = Status::OK();
613
614
0
        const bool has_cancel_reason = request->has_cancel_reason();
615
0
        const bool has_cancel_status = request->has_cancel_status();
616
        // During upgrade only LIMIT_REACH is used, other reason is changed to internal error
617
0
        Status actual_cancel_status = Status::OK();
618
        // Convert PPlanFragmentCancelReason to Status
619
0
        if (has_cancel_status) {
620
            // If fe set cancel status, then it is new FE now, should use cancel status.
621
0
            actual_cancel_status = Status::create<false>(request->cancel_status());
622
0
        } else if (has_cancel_reason) {
623
            // If fe not set cancel status, but set cancel reason, should convert cancel reason
624
            // to cancel status here.
625
0
            if (request->cancel_reason() == PPlanFragmentCancelReason::LIMIT_REACH) {
626
0
                actual_cancel_status = Status::Error<ErrorCode::LIMIT_REACH>("limit reach");
627
0
            } else {
628
                // Use cancel reason as error message
629
0
                actual_cancel_status = Status::InternalError(
630
0
                        PPlanFragmentCancelReason_Name(request->cancel_reason()));
631
0
            }
632
0
        } else {
633
0
            actual_cancel_status = Status::InternalError("unknown error");
634
0
        }
635
636
0
        TUniqueId query_id;
637
0
        query_id.__set_hi(request->query_id().hi());
638
0
        query_id.__set_lo(request->query_id().lo());
639
0
        LOG(INFO) << fmt::format("Cancel query {}, reason: {}", print_id(query_id),
640
0
                                 actual_cancel_status.to_string());
641
0
        _exec_env->fragment_mgr()->cancel_query(query_id, actual_cancel_status);
642
643
        // TODO: the logic seems useless, cancel only return Status::OK. remove it
644
0
        st.to_protobuf(result->mutable_status());
645
0
    });
646
0
    if (!ret) {
647
0
        offer_failed(result, done, _light_work_pool);
648
0
        return;
649
0
    }
650
0
}
651
652
void PInternalService::fetch_data(google::protobuf::RpcController* controller,
653
                                  const PFetchDataRequest* request, PFetchDataResult* result,
654
0
                                  google::protobuf::Closure* done) {
655
    // fetch_data is a light operation which will put a request rather than wait inplace when there's no data ready.
656
    // when there's data ready, use brpc to send. there's queue in brpc service. won't take it too long.
657
0
    auto ctx = vectorized::GetResultBatchCtx::create_shared(result, done);
658
0
    TUniqueId tid = UniqueId(request->finst_id()).to_thrift();
659
0
    std::shared_ptr<vectorized::MySQLResultBlockBuffer> buffer;
660
0
    Status st = ExecEnv::GetInstance()->result_mgr()->find_buffer(tid, buffer);
661
0
    if (!st.ok()) {
662
0
        LOG(WARNING) << "Result buffer not found! Query ID: " << print_id(tid);
663
0
        return;
664
0
    }
665
0
    if (st = buffer->get_batch(ctx); !st.ok()) {
666
0
        LOG(WARNING) << "fetch_data failed: " << st.to_string();
667
0
    }
668
0
}
669
670
void PInternalService::fetch_arrow_data(google::protobuf::RpcController* controller,
671
                                        const PFetchArrowDataRequest* request,
672
                                        PFetchArrowDataResult* result,
673
0
                                        google::protobuf::Closure* done) {
674
0
    bool ret = _arrow_flight_work_pool.try_offer([request, result, done]() {
675
0
        brpc::ClosureGuard closure_guard(done);
676
0
        auto ctx = vectorized::GetArrowResultBatchCtx::create_shared(result);
677
0
        TUniqueId tid = UniqueId(request->finst_id()).to_thrift();
678
0
        std::shared_ptr<vectorized::ArrowFlightResultBlockBuffer> arrow_buffer;
679
0
        auto st = ExecEnv::GetInstance()->result_mgr()->find_buffer(tid, arrow_buffer);
680
0
        if (!st.ok()) {
681
0
            LOG(WARNING) << "Result buffer not found! Query ID: " << print_id(tid);
682
0
            return;
683
0
        }
684
0
        if (st = arrow_buffer->get_batch(ctx); !st.ok()) {
685
0
            LOG(WARNING) << "fetch_arrow_data failed: " << st.to_string();
686
0
        }
687
0
    });
688
0
    if (!ret) {
689
0
        offer_failed(result, done, _arrow_flight_work_pool);
690
0
        return;
691
0
    }
692
0
}
693
694
void PInternalService::outfile_write_success(google::protobuf::RpcController* controller,
695
                                             const POutfileWriteSuccessRequest* request,
696
                                             POutfileWriteSuccessResult* result,
697
0
                                             google::protobuf::Closure* done) {
698
0
    bool ret = _heavy_work_pool.try_offer([request, result, done]() {
699
0
        VLOG_RPC << "outfile write success file";
700
0
        brpc::ClosureGuard closure_guard(done);
701
0
        TResultFileSink result_file_sink;
702
0
        Status st = Status::OK();
703
0
        {
704
0
            const uint8_t* buf = (const uint8_t*)(request->result_file_sink().data());
705
0
            uint32_t len = request->result_file_sink().size();
706
0
            st = deserialize_thrift_msg(buf, &len, false, &result_file_sink);
707
0
            if (!st.ok()) {
708
0
                LOG(WARNING) << "outfile write success file failed, errmsg = " << st;
709
0
                st.to_protobuf(result->mutable_status());
710
0
                return;
711
0
            }
712
0
        }
713
714
0
        TResultFileSinkOptions file_options = result_file_sink.file_options;
715
0
        std::stringstream ss;
716
0
        ss << file_options.file_path << file_options.success_file_name;
717
0
        std::string file_name = ss.str();
718
0
        if (result_file_sink.storage_backend_type == TStorageBackendType::LOCAL) {
719
            // For local file writer, the file_path is a local dir.
720
            // Here we do a simple security verification by checking whether the file exists.
721
            // Because the file path is currently arbitrarily specified by the user,
722
            // Doris is not responsible for ensuring the correctness of the path.
723
            // This is just to prevent overwriting the existing file.
724
0
            bool exists = true;
725
0
            st = io::global_local_filesystem()->exists(file_name, &exists);
726
0
            if (!st.ok()) {
727
0
                LOG(WARNING) << "outfile write success filefailed, errmsg = " << st;
728
0
                st.to_protobuf(result->mutable_status());
729
0
                return;
730
0
            }
731
0
            if (exists) {
732
0
                st = Status::InternalError("File already exists: {}", file_name);
733
0
            }
734
0
            if (!st.ok()) {
735
0
                LOG(WARNING) << "outfile write success file failed, errmsg = " << st;
736
0
                st.to_protobuf(result->mutable_status());
737
0
                return;
738
0
            }
739
0
        }
740
741
0
        auto&& res = FileFactory::create_file_writer(
742
0
                FileFactory::convert_storage_type(result_file_sink.storage_backend_type),
743
0
                ExecEnv::GetInstance(), file_options.broker_addresses,
744
0
                file_options.broker_properties, file_name,
745
0
                {
746
0
                        .write_file_cache = false,
747
0
                        .sync_file_data = false,
748
0
                });
749
0
        using T = std::decay_t<decltype(res)>;
750
0
        if (!res.has_value()) [[unlikely]] {
751
0
            st = std::forward<T>(res).error();
752
0
            st.to_protobuf(result->mutable_status());
753
0
            return;
754
0
        }
755
756
0
        std::unique_ptr<doris::io::FileWriter> _file_writer_impl = std::forward<T>(res).value();
757
        // must write somthing because s3 file writer can not writer empty file
758
0
        st = _file_writer_impl->append({"success"});
759
0
        if (!st.ok()) {
760
0
            LOG(WARNING) << "outfile write success filefailed, errmsg=" << st;
761
0
            st.to_protobuf(result->mutable_status());
762
0
            return;
763
0
        }
764
0
        st = _file_writer_impl->close();
765
0
        if (!st.ok()) {
766
0
            LOG(WARNING) << "outfile write success filefailed, errmsg=" << st;
767
0
            st.to_protobuf(result->mutable_status());
768
0
            return;
769
0
        }
770
0
    });
771
0
    if (!ret) {
772
0
        offer_failed(result, done, _heavy_work_pool);
773
0
        return;
774
0
    }
775
0
}
776
777
void PInternalService::fetch_table_schema(google::protobuf::RpcController* controller,
778
                                          const PFetchTableSchemaRequest* request,
779
                                          PFetchTableSchemaResult* result,
780
0
                                          google::protobuf::Closure* done) {
781
0
    bool ret = _heavy_work_pool.try_offer([request, result, done]() {
782
0
        VLOG_RPC << "fetch table schema";
783
0
        brpc::ClosureGuard closure_guard(done);
784
0
        TFileScanRange file_scan_range;
785
0
        Status st = Status::OK();
786
0
        {
787
0
            const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
788
0
            uint32_t len = request->file_scan_range().size();
789
0
            st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
790
0
            if (!st.ok()) {
791
0
                LOG(WARNING) << "fetch table schema failed, errmsg=" << st;
792
0
                st.to_protobuf(result->mutable_status());
793
0
                return;
794
0
            }
795
0
        }
796
0
        if (file_scan_range.__isset.ranges == false) {
797
0
            st = Status::InternalError("can not get TFileRangeDesc.");
798
0
            st.to_protobuf(result->mutable_status());
799
0
            return;
800
0
        }
801
0
        if (file_scan_range.__isset.params == false) {
802
0
            st = Status::InternalError("can not get TFileScanRangeParams.");
803
0
            st.to_protobuf(result->mutable_status());
804
0
            return;
805
0
        }
806
0
        const TFileRangeDesc& range = file_scan_range.ranges.at(0);
807
0
        const TFileScanRangeParams& params = file_scan_range.params;
808
809
0
        std::shared_ptr<MemTrackerLimiter> mem_tracker = MemTrackerLimiter::create_shared(
810
0
                MemTrackerLimiter::Type::OTHER,
811
0
                fmt::format("InternalService::fetch_table_schema:{}#{}", params.format_type,
812
0
                            params.file_type));
813
0
        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(mem_tracker);
814
815
        // make sure profile is desctructed after reader cause PrefetchBufferedReader
816
        // might asynchronouslly access the profile
817
0
        std::unique_ptr<RuntimeProfile> profile =
818
0
                std::make_unique<RuntimeProfile>("FetchTableSchema");
819
0
        std::unique_ptr<vectorized::GenericReader> reader(nullptr);
820
0
        io::IOContext io_ctx;
821
0
        io::FileCacheStatistics file_cache_statis;
822
0
        io_ctx.file_cache_stats = &file_cache_statis;
823
0
        switch (params.format_type) {
824
0
        case TFileFormatType::FORMAT_CSV_PLAIN:
825
0
        case TFileFormatType::FORMAT_CSV_GZ:
826
0
        case TFileFormatType::FORMAT_CSV_BZ2:
827
0
        case TFileFormatType::FORMAT_CSV_LZ4FRAME:
828
0
        case TFileFormatType::FORMAT_CSV_LZ4BLOCK:
829
0
        case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
830
0
        case TFileFormatType::FORMAT_CSV_LZOP:
831
0
        case TFileFormatType::FORMAT_CSV_DEFLATE: {
832
            // file_slots is no use
833
0
            std::vector<SlotDescriptor*> file_slots;
834
0
            reader = vectorized::CsvReader::create_unique(profile.get(), params, range, file_slots,
835
0
                                                          &io_ctx);
836
0
            break;
837
0
        }
838
0
        case TFileFormatType::FORMAT_PARQUET: {
839
0
            reader = vectorized::ParquetReader::create_unique(params, range, &io_ctx, nullptr);
840
0
            break;
841
0
        }
842
0
        case TFileFormatType::FORMAT_ORC: {
843
0
            reader = vectorized::OrcReader::create_unique(params, range, "", &io_ctx);
844
0
            break;
845
0
        }
846
0
        case TFileFormatType::FORMAT_JSON: {
847
0
            std::vector<SlotDescriptor*> file_slots;
848
0
            reader = vectorized::NewJsonReader::create_unique(profile.get(), params, range,
849
0
                                                              file_slots, &io_ctx);
850
0
            break;
851
0
        }
852
0
        case TFileFormatType::FORMAT_AVRO: {
853
            // file_slots is no use
854
0
            std::vector<SlotDescriptor*> file_slots;
855
0
            reader = vectorized::AvroJNIReader::create_unique(profile.get(), params, range,
856
0
                                                              file_slots);
857
0
            st = ((vectorized::AvroJNIReader*)(reader.get()))->init_fetch_table_schema_reader();
858
0
            break;
859
0
        }
860
0
        default:
861
0
            st = Status::InternalError("Not supported file format in fetch table schema: {}",
862
0
                                       params.format_type);
863
0
            st.to_protobuf(result->mutable_status());
864
0
            return;
865
0
        }
866
0
        if (!st.ok()) {
867
0
            LOG(WARNING) << "failed to init reader, errmsg=" << st;
868
0
            st.to_protobuf(result->mutable_status());
869
0
            return;
870
0
        }
871
0
        std::vector<std::string> col_names;
872
0
        std::vector<TypeDescriptor> col_types;
873
0
        st = reader->get_parsed_schema(&col_names, &col_types);
874
0
        if (!st.ok()) {
875
0
            LOG(WARNING) << "fetch table schema failed, errmsg=" << st;
876
0
            st.to_protobuf(result->mutable_status());
877
0
            return;
878
0
        }
879
0
        result->set_column_nums(col_names.size());
880
0
        for (size_t idx = 0; idx < col_names.size(); ++idx) {
881
0
            result->add_column_names(col_names[idx]);
882
0
        }
883
0
        for (size_t idx = 0; idx < col_types.size(); ++idx) {
884
0
            PTypeDesc* type_desc = result->add_column_types();
885
0
            col_types[idx].to_protobuf(type_desc);
886
0
        }
887
0
        st.to_protobuf(result->mutable_status());
888
0
    });
889
0
    if (!ret) {
890
0
        offer_failed(result, done, _heavy_work_pool);
891
0
        return;
892
0
    }
893
0
}
894
895
void PInternalService::fetch_arrow_flight_schema(google::protobuf::RpcController* controller,
896
                                                 const PFetchArrowFlightSchemaRequest* request,
897
                                                 PFetchArrowFlightSchemaResult* result,
898
0
                                                 google::protobuf::Closure* done) {
899
0
    bool ret = _light_work_pool.try_offer([request, result, done]() {
900
0
        brpc::ClosureGuard closure_guard(done);
901
0
        std::shared_ptr<arrow::Schema> schema;
902
0
        std::shared_ptr<vectorized::ArrowFlightResultBlockBuffer> buffer;
903
0
        auto st = ExecEnv::GetInstance()->result_mgr()->find_buffer(
904
0
                UniqueId(request->finst_id()).to_thrift(), buffer);
905
0
        if (!st.ok()) {
906
0
            LOG(WARNING) << "fetch arrow flight schema failed, errmsg=" << st;
907
0
            st.to_protobuf(result->mutable_status());
908
0
            return;
909
0
        }
910
0
        st = buffer->get_schema(&schema);
911
0
        if (!st.ok()) {
912
0
            LOG(WARNING) << "fetch arrow flight schema failed, errmsg=" << st;
913
0
            st.to_protobuf(result->mutable_status());
914
0
            return;
915
0
        }
916
917
0
        std::string schema_str;
918
0
        st = serialize_arrow_schema(&schema, &schema_str);
919
0
        if (st.ok()) {
920
0
            result->set_schema(std::move(schema_str));
921
0
            if (!config::public_host.empty()) {
922
0
                result->set_be_arrow_flight_ip(config::public_host);
923
0
            }
924
0
            if (config::arrow_flight_sql_proxy_port != -1) {
925
0
                result->set_be_arrow_flight_port(config::arrow_flight_sql_proxy_port);
926
0
            }
927
0
        }
928
0
        st.to_protobuf(result->mutable_status());
929
0
    });
930
0
    if (!ret) {
931
0
        offer_failed(result, done, _heavy_work_pool);
932
0
        return;
933
0
    }
934
0
}
935
936
Status PInternalService::_tablet_fetch_data(const PTabletKeyLookupRequest* request,
937
0
                                            PTabletKeyLookupResponse* response) {
938
0
    PointQueryExecutor executor;
939
0
    RETURN_IF_ERROR(executor.init(request, response));
940
0
    RETURN_IF_ERROR(executor.lookup_up());
941
0
    executor.print_profile();
942
0
    return Status::OK();
943
0
}
944
945
void PInternalService::tablet_fetch_data(google::protobuf::RpcController* controller,
946
                                         const PTabletKeyLookupRequest* request,
947
                                         PTabletKeyLookupResponse* response,
948
0
                                         google::protobuf::Closure* done) {
949
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
950
0
        [[maybe_unused]] auto* cntl = static_cast<brpc::Controller*>(controller);
951
0
        brpc::ClosureGuard guard(done);
952
0
        Status st = _tablet_fetch_data(request, response);
953
0
        st.to_protobuf(response->mutable_status());
954
0
    });
955
0
    if (!ret) {
956
0
        offer_failed(response, done, _light_work_pool);
957
0
        return;
958
0
    }
959
0
}
960
961
void PInternalService::test_jdbc_connection(google::protobuf::RpcController* controller,
962
                                            const PJdbcTestConnectionRequest* request,
963
                                            PJdbcTestConnectionResult* result,
964
0
                                            google::protobuf::Closure* done) {
965
0
    bool ret = _heavy_work_pool.try_offer([request, result, done]() {
966
0
        VLOG_RPC << "test jdbc connection";
967
0
        brpc::ClosureGuard closure_guard(done);
968
0
        TTableDescriptor table_desc;
969
0
        vectorized::JdbcConnectorParam jdbc_param;
970
0
        Status st = Status::OK();
971
0
        {
972
0
            const uint8_t* buf = (const uint8_t*)request->jdbc_table().data();
973
0
            uint32_t len = request->jdbc_table().size();
974
0
            st = deserialize_thrift_msg(buf, &len, false, &table_desc);
975
0
            if (!st.ok()) {
976
0
                LOG(WARNING) << "test jdbc connection failed, errmsg=" << st;
977
0
                st.to_protobuf(result->mutable_status());
978
0
                return;
979
0
            }
980
0
        }
981
0
        TJdbcTable jdbc_table = (table_desc.jdbcTable);
982
0
        jdbc_param.catalog_id = jdbc_table.catalog_id;
983
0
        jdbc_param.driver_class = jdbc_table.jdbc_driver_class;
984
0
        jdbc_param.driver_path = jdbc_table.jdbc_driver_url;
985
0
        jdbc_param.driver_checksum = jdbc_table.jdbc_driver_checksum;
986
0
        jdbc_param.jdbc_url = jdbc_table.jdbc_url;
987
0
        jdbc_param.user = jdbc_table.jdbc_user;
988
0
        jdbc_param.passwd = jdbc_table.jdbc_password;
989
0
        jdbc_param.query_string = request->query_str();
990
0
        jdbc_param.table_type = static_cast<TOdbcTableType::type>(request->jdbc_table_type());
991
0
        jdbc_param.use_transaction = false;
992
0
        jdbc_param.connection_pool_min_size = jdbc_table.connection_pool_min_size;
993
0
        jdbc_param.connection_pool_max_size = jdbc_table.connection_pool_max_size;
994
0
        jdbc_param.connection_pool_max_life_time = jdbc_table.connection_pool_max_life_time;
995
0
        jdbc_param.connection_pool_max_wait_time = jdbc_table.connection_pool_max_wait_time;
996
0
        jdbc_param.connection_pool_keep_alive = jdbc_table.connection_pool_keep_alive;
997
998
0
        std::unique_ptr<vectorized::JdbcConnector> jdbc_connector;
999
0
        jdbc_connector.reset(new (std::nothrow) vectorized::JdbcConnector(jdbc_param));
1000
1001
0
        st = jdbc_connector->test_connection();
1002
0
        st.to_protobuf(result->mutable_status());
1003
1004
0
        Status clean_st = jdbc_connector->clean_datasource();
1005
0
        if (!clean_st.ok()) {
1006
0
            LOG(WARNING) << "Failed to clean JDBC datasource: " << clean_st.msg();
1007
0
        }
1008
0
        Status close_st = jdbc_connector->close();
1009
0
        if (!close_st.ok()) {
1010
0
            LOG(WARNING) << "Failed to close JDBC connector: " << close_st.msg();
1011
0
        }
1012
0
    });
1013
1014
0
    if (!ret) {
1015
0
        offer_failed(result, done, _heavy_work_pool);
1016
0
        return;
1017
0
    }
1018
0
}
1019
1020
void PInternalServiceImpl::get_column_ids_by_tablet_ids(google::protobuf::RpcController* controller,
1021
                                                        const PFetchColIdsRequest* request,
1022
                                                        PFetchColIdsResponse* response,
1023
0
                                                        google::protobuf::Closure* done) {
1024
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
1025
0
        _get_column_ids_by_tablet_ids(controller, request, response, done);
1026
0
    });
1027
0
    if (!ret) {
1028
0
        offer_failed(response, done, _light_work_pool);
1029
0
        return;
1030
0
    }
1031
0
}
1032
1033
void PInternalServiceImpl::_get_column_ids_by_tablet_ids(
1034
        google::protobuf::RpcController* controller, const PFetchColIdsRequest* request,
1035
0
        PFetchColIdsResponse* response, google::protobuf::Closure* done) {
1036
0
    brpc::ClosureGuard guard(done);
1037
0
    [[maybe_unused]] auto* cntl = static_cast<brpc::Controller*>(controller);
1038
0
    TabletManager* tablet_mgr = _engine.tablet_manager();
1039
0
    const auto& params = request->params();
1040
0
    for (const auto& param : params) {
1041
0
        int64_t index_id = param.indexid();
1042
0
        const auto& tablet_ids = param.tablet_ids();
1043
0
        std::set<std::set<int32_t>> filter_set;
1044
0
        std::map<int32_t, const TabletColumn*> id_to_column;
1045
0
        for (const int64_t tablet_id : tablet_ids) {
1046
0
            TabletSharedPtr tablet = tablet_mgr->get_tablet(tablet_id);
1047
0
            if (tablet == nullptr) {
1048
0
                std::stringstream ss;
1049
0
                ss << "cannot get tablet by id:" << tablet_id;
1050
0
                LOG(WARNING) << ss.str();
1051
0
                response->mutable_status()->set_status_code(TStatusCode::ILLEGAL_STATE);
1052
0
                response->mutable_status()->add_error_msgs(ss.str());
1053
0
                return;
1054
0
            }
1055
            // check schema consistency, column ids should be the same
1056
0
            const auto& columns = tablet->tablet_schema()->columns();
1057
1058
0
            std::set<int32_t> column_ids;
1059
0
            for (const auto& col : columns) {
1060
0
                column_ids.insert(col->unique_id());
1061
0
            }
1062
0
            filter_set.insert(std::move(column_ids));
1063
1064
0
            if (id_to_column.empty()) {
1065
0
                for (const auto& col : columns) {
1066
0
                    id_to_column.insert(std::pair {col->unique_id(), col.get()});
1067
0
                }
1068
0
            } else {
1069
0
                for (const auto& col : columns) {
1070
0
                    auto it = id_to_column.find(col->unique_id());
1071
0
                    if (it == id_to_column.end() || *(it->second) != *col) {
1072
0
                        ColumnPB prev_col_pb;
1073
0
                        ColumnPB curr_col_pb;
1074
0
                        if (it != id_to_column.end()) {
1075
0
                            it->second->to_schema_pb(&prev_col_pb);
1076
0
                        }
1077
0
                        col->to_schema_pb(&curr_col_pb);
1078
0
                        std::stringstream ss;
1079
0
                        ss << "consistency check failed: index{ " << index_id << " }"
1080
0
                           << " got inconsistent schema, prev column: " << prev_col_pb.DebugString()
1081
0
                           << " current column: " << curr_col_pb.DebugString();
1082
0
                        LOG(WARNING) << ss.str();
1083
0
                        response->mutable_status()->set_status_code(TStatusCode::ILLEGAL_STATE);
1084
0
                        response->mutable_status()->add_error_msgs(ss.str());
1085
0
                        return;
1086
0
                    }
1087
0
                }
1088
0
            }
1089
0
        }
1090
1091
0
        if (filter_set.size() > 1) {
1092
            // consistecy check failed
1093
0
            std::stringstream ss;
1094
0
            ss << "consistency check failed: index{" << index_id << "}"
1095
0
               << "got inconsistent schema";
1096
0
            LOG(WARNING) << ss.str();
1097
0
            response->mutable_status()->set_status_code(TStatusCode::ILLEGAL_STATE);
1098
0
            response->mutable_status()->add_error_msgs(ss.str());
1099
0
            return;
1100
0
        }
1101
        // consistency check passed, use the first tablet to be the representative
1102
0
        TabletSharedPtr tablet = tablet_mgr->get_tablet(tablet_ids[0]);
1103
0
        const auto& columns = tablet->tablet_schema()->columns();
1104
0
        auto entry = response->add_entries();
1105
0
        entry->set_index_id(index_id);
1106
0
        auto col_name_to_id = entry->mutable_col_name_to_id();
1107
0
        for (const auto& column : columns) {
1108
0
            (*col_name_to_id)[column->name()] = column->unique_id();
1109
0
        }
1110
0
    }
1111
0
    response->mutable_status()->set_status_code(TStatusCode::OK);
1112
0
}
1113
1114
template <class RPCResponse>
1115
struct AsyncRPCContext {
1116
    RPCResponse response;
1117
    brpc::Controller cntl;
1118
    brpc::CallId cid;
1119
};
1120
1121
void PInternalService::fetch_remote_tablet_schema(google::protobuf::RpcController* controller,
1122
                                                  const PFetchRemoteSchemaRequest* request,
1123
                                                  PFetchRemoteSchemaResponse* response,
1124
0
                                                  google::protobuf::Closure* done) {
1125
0
    bool ret = _heavy_work_pool.try_offer([request, response, done]() {
1126
0
        brpc::ClosureGuard closure_guard(done);
1127
0
        Status st = Status::OK();
1128
0
        if (request->is_coordinator()) {
1129
            // Spawn rpc request to none coordinator nodes, and finally merge them all
1130
0
            PFetchRemoteSchemaRequest remote_request(*request);
1131
            // set it none coordinator to get merged schema
1132
0
            remote_request.set_is_coordinator(false);
1133
0
            using PFetchRemoteTabletSchemaRpcContext = AsyncRPCContext<PFetchRemoteSchemaResponse>;
1134
0
            std::vector<PFetchRemoteTabletSchemaRpcContext> rpc_contexts(
1135
0
                    request->tablet_location_size());
1136
0
            for (int i = 0; i < request->tablet_location_size(); ++i) {
1137
0
                std::string host = request->tablet_location(i).host();
1138
0
                int32_t brpc_port = request->tablet_location(i).brpc_port();
1139
0
                std::shared_ptr<PBackendService_Stub> stub(
1140
0
                        ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(
1141
0
                                host, brpc_port));
1142
0
                if (stub == nullptr) {
1143
0
                    LOG(WARNING) << "Failed to init rpc to " << host << ":" << brpc_port;
1144
0
                    st = Status::InternalError("Failed to init rpc to {}:{}", host, brpc_port);
1145
0
                    continue;
1146
0
                }
1147
0
                rpc_contexts[i].cid = rpc_contexts[i].cntl.call_id();
1148
0
                rpc_contexts[i].cntl.set_timeout_ms(config::fetch_remote_schema_rpc_timeout_ms);
1149
0
                stub->fetch_remote_tablet_schema(&rpc_contexts[i].cntl, &remote_request,
1150
0
                                                 &rpc_contexts[i].response, brpc::DoNothing());
1151
0
            }
1152
0
            std::vector<TabletSchemaSPtr> schemas;
1153
0
            for (auto& rpc_context : rpc_contexts) {
1154
0
                brpc::Join(rpc_context.cid);
1155
0
                if (!st.ok()) {
1156
                    // make sure all flying rpc request is joined
1157
0
                    continue;
1158
0
                }
1159
0
                if (rpc_context.cntl.Failed()) {
1160
0
                    LOG(WARNING) << "fetch_remote_tablet_schema rpc err:"
1161
0
                                 << rpc_context.cntl.ErrorText();
1162
0
                    ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(
1163
0
                            rpc_context.cntl.remote_side());
1164
0
                    st = Status::InternalError("fetch_remote_tablet_schema rpc err: {}",
1165
0
                                               rpc_context.cntl.ErrorText());
1166
0
                }
1167
0
                if (rpc_context.response.status().status_code() != 0) {
1168
0
                    st = Status::create(rpc_context.response.status());
1169
0
                }
1170
0
                if (rpc_context.response.has_merged_schema()) {
1171
0
                    TabletSchemaSPtr schema = std::make_shared<TabletSchema>();
1172
0
                    schema->init_from_pb(rpc_context.response.merged_schema());
1173
0
                    schemas.push_back(schema);
1174
0
                }
1175
0
            }
1176
0
            if (!schemas.empty() && st.ok()) {
1177
                // merge all
1178
0
                TabletSchemaSPtr merged_schema;
1179
0
                static_cast<void>(vectorized::schema_util::get_least_common_schema(schemas, nullptr,
1180
0
                                                                                   merged_schema));
1181
0
                VLOG_DEBUG << "dump schema:" << merged_schema->dump_structure();
1182
0
                merged_schema->reserve_extracted_columns();
1183
0
                merged_schema->to_schema_pb(response->mutable_merged_schema());
1184
0
            }
1185
0
            st.to_protobuf(response->mutable_status());
1186
0
            return;
1187
0
        } else {
1188
            // This is not a coordinator, get it's tablet and merge schema
1189
0
            std::vector<int64_t> target_tablets;
1190
0
            for (int i = 0; i < request->tablet_location_size(); ++i) {
1191
0
                const auto& location = request->tablet_location(i);
1192
0
                auto backend = BackendOptions::get_local_backend();
1193
                // If this is the target backend
1194
0
                if (backend.host == location.host() && config::brpc_port == location.brpc_port()) {
1195
0
                    target_tablets.assign(location.tablet_id().begin(), location.tablet_id().end());
1196
0
                    break;
1197
0
                }
1198
0
            }
1199
0
            if (!target_tablets.empty()) {
1200
0
                std::vector<TabletSchemaSPtr> tablet_schemas;
1201
0
                for (int64_t tablet_id : target_tablets) {
1202
0
                    auto res = ExecEnv::get_tablet(tablet_id);
1203
0
                    if (!res.has_value()) {
1204
                        // just ignore
1205
0
                        LOG(WARNING) << "tablet does not exist, tablet id is " << tablet_id;
1206
0
                        continue;
1207
0
                    }
1208
0
                    auto schema = res.value()->merged_tablet_schema();
1209
0
                    if (schema != nullptr) {
1210
0
                        tablet_schemas.push_back(schema);
1211
0
                    }
1212
0
                }
1213
0
                if (!tablet_schemas.empty()) {
1214
                    // merge all
1215
0
                    TabletSchemaSPtr merged_schema;
1216
0
                    static_cast<void>(vectorized::schema_util::get_least_common_schema(
1217
0
                            tablet_schemas, nullptr, merged_schema));
1218
0
                    merged_schema->to_schema_pb(response->mutable_merged_schema());
1219
0
                    VLOG_DEBUG << "dump schema:" << merged_schema->dump_structure();
1220
0
                }
1221
0
            }
1222
0
            st.to_protobuf(response->mutable_status());
1223
0
        }
1224
0
    });
1225
0
    if (!ret) {
1226
0
        offer_failed(response, done, _heavy_work_pool);
1227
0
    }
1228
0
}
1229
1230
void PInternalService::report_stream_load_status(google::protobuf::RpcController* controller,
1231
                                                 const PReportStreamLoadStatusRequest* request,
1232
                                                 PReportStreamLoadStatusResponse* response,
1233
0
                                                 google::protobuf::Closure* done) {
1234
0
    TUniqueId load_id;
1235
0
    load_id.__set_hi(request->load_id().hi());
1236
0
    load_id.__set_lo(request->load_id().lo());
1237
0
    Status st = Status::OK();
1238
0
    auto stream_load_ctx = _exec_env->new_load_stream_mgr()->get(load_id);
1239
0
    if (!stream_load_ctx) {
1240
0
        st = Status::InternalError("unknown stream load id: {}", UniqueId(load_id).to_string());
1241
0
    }
1242
0
    stream_load_ctx->promise.set_value(st);
1243
0
    st.to_protobuf(response->mutable_status());
1244
0
}
1245
1246
void PInternalService::get_info(google::protobuf::RpcController* controller,
1247
                                const PProxyRequest* request, PProxyResult* response,
1248
0
                                google::protobuf::Closure* done) {
1249
0
    bool ret = _exec_env->routine_load_task_executor()->get_thread_pool().submit_func([this,
1250
0
                                                                                       request,
1251
0
                                                                                       response,
1252
0
                                                                                       done]() {
1253
0
        brpc::ClosureGuard closure_guard(done);
1254
        // PProxyRequest is defined in gensrc/proto/internal_service.proto
1255
        // Currently it supports 2 kinds of requests:
1256
        // 1. get all kafka partition ids for given topic
1257
        // 2. get all kafka partition offsets for given topic and timestamp.
1258
0
        int timeout_ms = request->has_timeout_secs() ? request->timeout_secs() * 1000 : 60 * 1000;
1259
0
        if (request->has_kafka_meta_request()) {
1260
0
            const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
1261
0
            if (!kafka_request.offset_flags().empty()) {
1262
0
                std::vector<PIntegerPair> partition_offsets;
1263
0
                Status st = _exec_env->routine_load_task_executor()
1264
0
                                    ->get_kafka_real_offsets_for_partitions(
1265
0
                                            request->kafka_meta_request(), &partition_offsets,
1266
0
                                            timeout_ms);
1267
0
                if (st.ok()) {
1268
0
                    PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
1269
0
                    for (const auto& entry : partition_offsets) {
1270
0
                        PIntegerPair* res = part_offsets->add_offset_times();
1271
0
                        res->set_key(entry.key());
1272
0
                        res->set_val(entry.val());
1273
0
                    }
1274
0
                }
1275
0
                st.to_protobuf(response->mutable_status());
1276
0
                return;
1277
0
            } else if (!kafka_request.partition_id_for_latest_offsets().empty()) {
1278
                // get latest offsets for specified partition ids
1279
0
                std::vector<PIntegerPair> partition_offsets;
1280
0
                Status st = _exec_env->routine_load_task_executor()
1281
0
                                    ->get_kafka_latest_offsets_for_partitions(
1282
0
                                            request->kafka_meta_request(), &partition_offsets,
1283
0
                                            timeout_ms);
1284
0
                if (st.ok()) {
1285
0
                    PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
1286
0
                    for (const auto& entry : partition_offsets) {
1287
0
                        PIntegerPair* res = part_offsets->add_offset_times();
1288
0
                        res->set_key(entry.key());
1289
0
                        res->set_val(entry.val());
1290
0
                    }
1291
0
                }
1292
0
                st.to_protobuf(response->mutable_status());
1293
0
                return;
1294
0
            } else if (!kafka_request.offset_times().empty()) {
1295
                // if offset_times() has elements, which means this request is to get offset by timestamp.
1296
0
                std::vector<PIntegerPair> partition_offsets;
1297
0
                Status st = _exec_env->routine_load_task_executor()
1298
0
                                    ->get_kafka_partition_offsets_for_times(
1299
0
                                            request->kafka_meta_request(), &partition_offsets,
1300
0
                                            timeout_ms);
1301
0
                if (st.ok()) {
1302
0
                    PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
1303
0
                    for (const auto& entry : partition_offsets) {
1304
0
                        PIntegerPair* res = part_offsets->add_offset_times();
1305
0
                        res->set_key(entry.key());
1306
0
                        res->set_val(entry.val());
1307
0
                    }
1308
0
                }
1309
0
                st.to_protobuf(response->mutable_status());
1310
0
                return;
1311
0
            } else {
1312
                // get partition ids of topic
1313
0
                std::vector<int32_t> partition_ids;
1314
0
                Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
1315
0
                        request->kafka_meta_request(), &partition_ids);
1316
0
                if (st.ok()) {
1317
0
                    PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
1318
0
                    for (int32_t id : partition_ids) {
1319
0
                        kafka_result->add_partition_ids(id);
1320
0
                    }
1321
0
                }
1322
0
                st.to_protobuf(response->mutable_status());
1323
0
                return;
1324
0
            }
1325
0
        }
1326
0
        Status::OK().to_protobuf(response->mutable_status());
1327
0
    });
1328
0
    if (!ret) {
1329
0
        offer_failed(response, done, _heavy_work_pool);
1330
0
        return;
1331
0
    }
1332
0
}
1333
1334
void PInternalService::update_cache(google::protobuf::RpcController* controller,
1335
                                    const PUpdateCacheRequest* request, PCacheResponse* response,
1336
0
                                    google::protobuf::Closure* done) {
1337
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1338
0
        brpc::ClosureGuard closure_guard(done);
1339
0
        _exec_env->result_cache()->update(request, response);
1340
0
    });
1341
0
    if (!ret) {
1342
0
        offer_failed(response, done, _light_work_pool);
1343
0
        return;
1344
0
    }
1345
0
}
1346
1347
void PInternalService::fetch_cache(google::protobuf::RpcController* controller,
1348
                                   const PFetchCacheRequest* request, PFetchCacheResult* result,
1349
0
                                   google::protobuf::Closure* done) {
1350
0
    bool ret = _heavy_work_pool.try_offer([this, request, result, done]() {
1351
0
        brpc::ClosureGuard closure_guard(done);
1352
0
        _exec_env->result_cache()->fetch(request, result);
1353
0
    });
1354
0
    if (!ret) {
1355
0
        offer_failed(result, done, _heavy_work_pool);
1356
0
        return;
1357
0
    }
1358
0
}
1359
1360
void PInternalService::clear_cache(google::protobuf::RpcController* controller,
1361
                                   const PClearCacheRequest* request, PCacheResponse* response,
1362
0
                                   google::protobuf::Closure* done) {
1363
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1364
0
        brpc::ClosureGuard closure_guard(done);
1365
0
        _exec_env->result_cache()->clear(request, response);
1366
0
    });
1367
0
    if (!ret) {
1368
0
        offer_failed(response, done, _light_work_pool);
1369
0
        return;
1370
0
    }
1371
0
}
1372
1373
void PInternalService::merge_filter(::google::protobuf::RpcController* controller,
1374
                                    const ::doris::PMergeFilterRequest* request,
1375
                                    ::doris::PMergeFilterResponse* response,
1376
0
                                    ::google::protobuf::Closure* done) {
1377
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
1378
0
        brpc::ClosureGuard closure_guard(done);
1379
0
        auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
1380
0
        butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
1381
0
        Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
1382
0
        st.to_protobuf(response->mutable_status());
1383
0
    });
1384
0
    if (!ret) {
1385
0
        offer_failed(response, done, _light_work_pool);
1386
0
        return;
1387
0
    }
1388
0
}
1389
1390
void PInternalService::send_filter_size(::google::protobuf::RpcController* controller,
1391
                                        const ::doris::PSendFilterSizeRequest* request,
1392
                                        ::doris::PSendFilterSizeResponse* response,
1393
0
                                        ::google::protobuf::Closure* done) {
1394
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1395
0
        brpc::ClosureGuard closure_guard(done);
1396
0
        Status st = _exec_env->fragment_mgr()->send_filter_size(request);
1397
0
        st.to_protobuf(response->mutable_status());
1398
0
    });
1399
0
    if (!ret) {
1400
0
        offer_failed(response, done, _light_work_pool);
1401
0
        return;
1402
0
    }
1403
0
}
1404
1405
void PInternalService::sync_filter_size(::google::protobuf::RpcController* controller,
1406
                                        const ::doris::PSyncFilterSizeRequest* request,
1407
                                        ::doris::PSyncFilterSizeResponse* response,
1408
0
                                        ::google::protobuf::Closure* done) {
1409
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1410
0
        brpc::ClosureGuard closure_guard(done);
1411
0
        Status st = _exec_env->fragment_mgr()->sync_filter_size(request);
1412
0
        st.to_protobuf(response->mutable_status());
1413
0
    });
1414
0
    if (!ret) {
1415
0
        offer_failed(response, done, _light_work_pool);
1416
0
        return;
1417
0
    }
1418
0
}
1419
1420
void PInternalService::apply_filterv2(::google::protobuf::RpcController* controller,
1421
                                      const ::doris::PPublishFilterRequestV2* request,
1422
                                      ::doris::PPublishFilterResponse* response,
1423
0
                                      ::google::protobuf::Closure* done) {
1424
0
    bool ret = _light_work_pool.try_offer([this, controller, request, response, done]() {
1425
0
        brpc::ClosureGuard closure_guard(done);
1426
0
        auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
1427
0
        butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
1428
0
        UniqueId unique_id(request->query_id());
1429
0
        VLOG_NOTICE << "rpc apply_filterv2 recv";
1430
0
        Status st = _exec_env->fragment_mgr()->apply_filterv2(request, &zero_copy_input_stream);
1431
0
        if (!st.ok()) {
1432
0
            LOG(WARNING) << "apply filter meet error: " << st.to_string();
1433
0
        }
1434
0
        st.to_protobuf(response->mutable_status());
1435
0
    });
1436
0
    if (!ret) {
1437
0
        offer_failed(response, done, _light_work_pool);
1438
0
        return;
1439
0
    }
1440
0
}
1441
1442
void PInternalService::send_data(google::protobuf::RpcController* controller,
1443
                                 const PSendDataRequest* request, PSendDataResult* response,
1444
0
                                 google::protobuf::Closure* done) {
1445
0
    bool ret = _heavy_work_pool.try_offer([this, request, response, done]() {
1446
0
        brpc::ClosureGuard closure_guard(done);
1447
0
        TUniqueId load_id;
1448
0
        load_id.hi = request->load_id().hi();
1449
0
        load_id.lo = request->load_id().lo();
1450
        // On 1.2.3 we add load id to send data request and using load id to get pipe
1451
0
        auto stream_load_ctx = _exec_env->new_load_stream_mgr()->get(load_id);
1452
0
        if (stream_load_ctx == nullptr) {
1453
0
            response->mutable_status()->set_status_code(1);
1454
0
            response->mutable_status()->add_error_msgs("could not find stream load context");
1455
0
        } else {
1456
0
            auto pipe = stream_load_ctx->pipe;
1457
0
            for (int i = 0; i < request->data_size(); ++i) {
1458
0
                std::unique_ptr<PDataRow> row(new PDataRow());
1459
0
                row->CopyFrom(request->data(i));
1460
0
                Status s = pipe->append(std::move(row));
1461
0
                if (!s.ok()) {
1462
0
                    response->mutable_status()->set_status_code(1);
1463
0
                    response->mutable_status()->add_error_msgs(s.to_string());
1464
0
                    return;
1465
0
                }
1466
0
            }
1467
0
            response->mutable_status()->set_status_code(0);
1468
0
        }
1469
0
    });
1470
0
    if (!ret) {
1471
0
        offer_failed(response, done, _heavy_work_pool);
1472
0
        return;
1473
0
    }
1474
0
}
1475
1476
void PInternalService::commit(google::protobuf::RpcController* controller,
1477
                              const PCommitRequest* request, PCommitResult* response,
1478
0
                              google::protobuf::Closure* done) {
1479
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1480
0
        brpc::ClosureGuard closure_guard(done);
1481
0
        TUniqueId load_id;
1482
0
        load_id.hi = request->load_id().hi();
1483
0
        load_id.lo = request->load_id().lo();
1484
1485
0
        auto stream_load_ctx = _exec_env->new_load_stream_mgr()->get(load_id);
1486
0
        if (stream_load_ctx == nullptr) {
1487
0
            response->mutable_status()->set_status_code(1);
1488
0
            response->mutable_status()->add_error_msgs("could not find stream load context");
1489
0
        } else {
1490
0
            static_cast<void>(stream_load_ctx->pipe->finish());
1491
0
            response->mutable_status()->set_status_code(0);
1492
0
        }
1493
0
    });
1494
0
    if (!ret) {
1495
0
        offer_failed(response, done, _light_work_pool);
1496
0
        return;
1497
0
    }
1498
0
}
1499
1500
void PInternalService::rollback(google::protobuf::RpcController* controller,
1501
                                const PRollbackRequest* request, PRollbackResult* response,
1502
0
                                google::protobuf::Closure* done) {
1503
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1504
0
        brpc::ClosureGuard closure_guard(done);
1505
0
        TUniqueId load_id;
1506
0
        load_id.hi = request->load_id().hi();
1507
0
        load_id.lo = request->load_id().lo();
1508
0
        auto stream_load_ctx = _exec_env->new_load_stream_mgr()->get(load_id);
1509
0
        if (stream_load_ctx == nullptr) {
1510
0
            response->mutable_status()->set_status_code(1);
1511
0
            response->mutable_status()->add_error_msgs("could not find stream load context");
1512
0
        } else {
1513
0
            stream_load_ctx->pipe->cancel("rollback");
1514
0
            response->mutable_status()->set_status_code(0);
1515
0
        }
1516
0
    });
1517
0
    if (!ret) {
1518
0
        offer_failed(response, done, _light_work_pool);
1519
0
        return;
1520
0
    }
1521
0
}
1522
1523
void PInternalService::fold_constant_expr(google::protobuf::RpcController* controller,
1524
                                          const PConstantExprRequest* request,
1525
                                          PConstantExprResult* response,
1526
0
                                          google::protobuf::Closure* done) {
1527
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
1528
0
        brpc::ClosureGuard closure_guard(done);
1529
0
        TFoldConstantParams t_request;
1530
0
        Status st = Status::OK();
1531
0
        {
1532
0
            const uint8_t* buf = (const uint8_t*)request->request().data();
1533
0
            uint32_t len = request->request().size();
1534
0
            st = deserialize_thrift_msg(buf, &len, false, &t_request);
1535
0
        }
1536
0
        if (!st.ok()) {
1537
0
            LOG(WARNING) << "exec fold constant expr failed, errmsg=" << st
1538
0
                         << " .and query_id_is: " << t_request.query_id;
1539
0
        }
1540
0
        st = _fold_constant_expr(request->request(), response);
1541
0
        if (!st.ok()) {
1542
0
            LOG(WARNING) << "exec fold constant expr failed, errmsg=" << st
1543
0
                         << " .and query_id_is: " << t_request.query_id;
1544
0
        }
1545
0
        st.to_protobuf(response->mutable_status());
1546
0
    });
1547
0
    if (!ret) {
1548
0
        offer_failed(response, done, _light_work_pool);
1549
0
        return;
1550
0
    }
1551
0
}
1552
1553
Status PInternalService::_fold_constant_expr(const std::string& ser_request,
1554
0
                                             PConstantExprResult* response) {
1555
0
    TFoldConstantParams t_request;
1556
0
    {
1557
0
        const uint8_t* buf = (const uint8_t*)ser_request.data();
1558
0
        uint32_t len = ser_request.size();
1559
0
        RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, false, &t_request));
1560
0
    }
1561
0
    std::unique_ptr<FoldConstantExecutor> fold_executor = std::make_unique<FoldConstantExecutor>();
1562
0
    RETURN_IF_ERROR_OR_CATCH_EXCEPTION(fold_executor->fold_constant_vexpr(t_request, response));
1563
0
    return Status::OK();
1564
0
}
1565
1566
void PInternalService::transmit_block(google::protobuf::RpcController* controller,
1567
                                      const PTransmitDataParams* request,
1568
                                      PTransmitDataResult* response,
1569
0
                                      google::protobuf::Closure* done) {
1570
0
    int64_t receive_time = GetCurrentTimeNanos();
1571
0
    if (config::enable_bthread_transmit_block) {
1572
0
        response->set_receive_time(receive_time);
1573
        // under high concurrency, thread pool will have a lot of lock contention.
1574
        // May offer failed to the thread pool, so that we should avoid using thread
1575
        // pool here.
1576
0
        _transmit_block(controller, request, response, done, Status::OK(), 0);
1577
0
    } else {
1578
0
        bool ret = _light_work_pool.try_offer([this, controller, request, response, done,
1579
0
                                               receive_time]() {
1580
0
            response->set_receive_time(receive_time);
1581
            // Sometimes transmit block function is the last owner of PlanFragmentExecutor
1582
            // It will release the object. And the object maybe a JNIContext.
1583
            // JNIContext will hold some TLS object. It could not work correctly under bthread
1584
            // Context. So that put the logic into pthread.
1585
            // But this is rarely happens, so this config is disabled by default.
1586
0
            _transmit_block(controller, request, response, done, Status::OK(),
1587
0
                            GetCurrentTimeNanos() - receive_time);
1588
0
        });
1589
0
        if (!ret) {
1590
0
            offer_failed(response, done, _light_work_pool);
1591
0
            return;
1592
0
        }
1593
0
    }
1594
0
}
1595
1596
void PInternalService::transmit_block_by_http(google::protobuf::RpcController* controller,
1597
                                              const PEmptyRequest* request,
1598
                                              PTransmitDataResult* response,
1599
0
                                              google::protobuf::Closure* done) {
1600
0
    int64_t receive_time = GetCurrentTimeNanos();
1601
0
    bool ret = _heavy_work_pool.try_offer([this, controller, response, done, receive_time]() {
1602
0
        PTransmitDataParams* new_request = new PTransmitDataParams();
1603
0
        google::protobuf::Closure* new_done =
1604
0
                new NewHttpClosure<PTransmitDataParams>(new_request, done);
1605
0
        brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
1606
0
        Status st =
1607
0
                attachment_extract_request_contain_block<PTransmitDataParams>(new_request, cntl);
1608
0
        _transmit_block(controller, new_request, response, new_done, st,
1609
0
                        GetCurrentTimeNanos() - receive_time);
1610
0
    });
1611
0
    if (!ret) {
1612
0
        offer_failed(response, done, _heavy_work_pool);
1613
0
        return;
1614
0
    }
1615
0
}
1616
1617
void PInternalService::_transmit_block(google::protobuf::RpcController* controller,
1618
                                       const PTransmitDataParams* request,
1619
                                       PTransmitDataResult* response,
1620
                                       google::protobuf::Closure* done, const Status& extract_st,
1621
0
                                       const int64_t wait_for_worker) {
1622
0
    if (request->has_query_id()) {
1623
0
        VLOG_ROW << "transmit block: fragment_instance_id=" << print_id(request->finst_id())
1624
0
                 << " query_id=" << print_id(request->query_id()) << " node=" << request->node_id();
1625
0
    }
1626
1627
    // The response is accessed when done->Run is called in transmit_block(),
1628
    // give response a default value to avoid null pointers in high concurrency.
1629
0
    Status st;
1630
0
    if (extract_st.ok()) {
1631
0
        st = _exec_env->vstream_mgr()->transmit_block(request, &done, wait_for_worker);
1632
0
        if (!st.ok() && !st.is<END_OF_FILE>()) {
1633
0
            LOG(WARNING) << "transmit_block failed, message=" << st
1634
0
                         << ", fragment_instance_id=" << print_id(request->finst_id())
1635
0
                         << ", node=" << request->node_id()
1636
0
                         << ", from sender_id: " << request->sender_id()
1637
0
                         << ", be_number: " << request->be_number()
1638
0
                         << ", packet_seq: " << request->packet_seq();
1639
0
        }
1640
0
    } else {
1641
0
        st = extract_st;
1642
0
    }
1643
0
    if (done != nullptr) {
1644
0
        st.to_protobuf(response->mutable_status());
1645
0
        done->Run();
1646
0
    }
1647
0
}
1648
1649
void PInternalService::check_rpc_channel(google::protobuf::RpcController* controller,
1650
                                         const PCheckRPCChannelRequest* request,
1651
                                         PCheckRPCChannelResponse* response,
1652
0
                                         google::protobuf::Closure* done) {
1653
0
    bool ret = _light_work_pool.try_offer([request, response, done]() {
1654
0
        brpc::ClosureGuard closure_guard(done);
1655
0
        response->mutable_status()->set_status_code(0);
1656
0
        if (request->data().size() != request->size()) {
1657
0
            std::stringstream ss;
1658
0
            ss << "data size not same, expected: " << request->size()
1659
0
               << ", actual: " << request->data().size();
1660
0
            response->mutable_status()->add_error_msgs(ss.str());
1661
0
            response->mutable_status()->set_status_code(1);
1662
1663
0
        } else {
1664
0
            Md5Digest digest;
1665
0
            digest.update(static_cast<const void*>(request->data().c_str()),
1666
0
                          request->data().size());
1667
0
            digest.digest();
1668
0
            if (!iequal(digest.hex(), request->md5())) {
1669
0
                std::stringstream ss;
1670
0
                ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
1671
0
                response->mutable_status()->add_error_msgs(ss.str());
1672
0
                response->mutable_status()->set_status_code(1);
1673
0
            }
1674
0
        }
1675
0
    });
1676
0
    if (!ret) {
1677
0
        offer_failed(response, done, _light_work_pool);
1678
0
        return;
1679
0
    }
1680
0
}
1681
1682
void PInternalService::reset_rpc_channel(google::protobuf::RpcController* controller,
1683
                                         const PResetRPCChannelRequest* request,
1684
                                         PResetRPCChannelResponse* response,
1685
0
                                         google::protobuf::Closure* done) {
1686
0
    bool ret = _light_work_pool.try_offer([request, response, done]() {
1687
0
        brpc::ClosureGuard closure_guard(done);
1688
0
        response->mutable_status()->set_status_code(0);
1689
0
        if (request->all()) {
1690
0
            int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
1691
0
            if (size > 0) {
1692
0
                std::vector<std::string> endpoints;
1693
0
                ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
1694
0
                ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
1695
0
                *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
1696
0
            }
1697
0
        } else {
1698
0
            for (const std::string& endpoint : request->endpoints()) {
1699
0
                if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
1700
0
                    response->mutable_status()->add_error_msgs(endpoint + ": not found.");
1701
0
                    continue;
1702
0
                }
1703
1704
0
                if (ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
1705
0
                    response->add_channels(endpoint);
1706
0
                } else {
1707
0
                    response->mutable_status()->add_error_msgs(endpoint + ": reset failed.");
1708
0
                }
1709
0
            }
1710
0
            if (request->endpoints_size() != response->channels_size()) {
1711
0
                response->mutable_status()->set_status_code(1);
1712
0
            }
1713
0
        }
1714
0
    });
1715
0
    if (!ret) {
1716
0
        offer_failed(response, done, _light_work_pool);
1717
0
        return;
1718
0
    }
1719
0
}
1720
1721
void PInternalService::hand_shake(google::protobuf::RpcController* controller,
1722
                                  const PHandShakeRequest* request, PHandShakeResponse* response,
1723
0
                                  google::protobuf::Closure* done) {
1724
    // The light pool may be full. Handshake is used to check the connection state of brpc.
1725
    // Should not be interfered by the thread pool logic.
1726
0
    brpc::ClosureGuard closure_guard(done);
1727
0
    if (request->has_hello()) {
1728
0
        response->set_hello(request->hello());
1729
0
    }
1730
0
    response->mutable_status()->set_status_code(0);
1731
0
}
1732
1733
constexpr char HttpProtocol[] = "http://";
1734
constexpr char DownloadApiPath[] = "/api/_tablet/_download?token=";
1735
constexpr char FileParam[] = "&file=";
1736
1737
static std::string construct_url(const std::string& host_port, const std::string& token,
1738
0
                                 const std::string& path) {
1739
0
    return fmt::format("{}{}{}{}{}{}", HttpProtocol, host_port, DownloadApiPath, token, FileParam,
1740
0
                       path);
1741
0
}
1742
1743
static Status download_file_action(std::string& remote_file_url, std::string& local_file_path,
1744
0
                                   uint64_t estimate_timeout, uint64_t file_size) {
1745
0
    auto download_cb = [remote_file_url, estimate_timeout, local_file_path,
1746
0
                        file_size](HttpClient* client) {
1747
0
        RETURN_IF_ERROR(client->init(remote_file_url));
1748
0
        client->set_timeout_ms(estimate_timeout * 1000);
1749
0
        RETURN_IF_ERROR(client->download(local_file_path));
1750
1751
0
        if (file_size > 0) {
1752
            // Check file length
1753
0
            uint64_t local_file_size = std::filesystem::file_size(local_file_path);
1754
0
            if (local_file_size != file_size) {
1755
0
                LOG(WARNING) << "failed to pull rowset for slave replica. download file "
1756
0
                                "length error"
1757
0
                             << ", remote_path=" << remote_file_url << ", file_size=" << file_size
1758
0
                             << ", local_file_size=" << local_file_size;
1759
0
                return Status::InternalError("downloaded file size is not equal");
1760
0
            }
1761
0
        }
1762
1763
0
        return io::global_local_filesystem()->permission(local_file_path,
1764
0
                                                         io::LocalFileSystem::PERMS_OWNER_RW);
1765
0
    };
1766
0
    return HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, download_cb);
1767
0
}
1768
1769
void PInternalServiceImpl::request_slave_tablet_pull_rowset(
1770
        google::protobuf::RpcController* controller, const PTabletWriteSlaveRequest* request,
1771
0
        PTabletWriteSlaveResult* response, google::protobuf::Closure* done) {
1772
0
    brpc::ClosureGuard closure_guard(done);
1773
0
    const RowsetMetaPB& rowset_meta_pb = request->rowset_meta();
1774
0
    const std::string& rowset_path = request->rowset_path();
1775
0
    google::protobuf::Map<int64, int64> segments_size = request->segments_size();
1776
0
    google::protobuf::Map<int64, PTabletWriteSlaveRequest_IndexSizeMap> indices_size =
1777
0
            request->inverted_indices_size();
1778
0
    std::string host = request->host();
1779
0
    int64_t http_port = request->http_port();
1780
0
    int64_t brpc_port = request->brpc_port();
1781
0
    std::string token = request->token();
1782
0
    int64_t node_id = request->node_id();
1783
0
    bool ret = _heavy_work_pool.try_offer([rowset_meta_pb, host, brpc_port, node_id, segments_size,
1784
0
                                           indices_size, http_port, token, rowset_path, this]() {
1785
0
        TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(
1786
0
                rowset_meta_pb.tablet_id(), rowset_meta_pb.tablet_schema_hash());
1787
0
        if (tablet == nullptr) {
1788
0
            LOG(WARNING) << "failed to pull rowset for slave replica. tablet ["
1789
0
                         << rowset_meta_pb.tablet_id()
1790
0
                         << "] is not exist. txn_id=" << rowset_meta_pb.txn_id();
1791
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta_pb.txn_id(),
1792
0
                                        rowset_meta_pb.tablet_id(), node_id, false);
1793
0
            return;
1794
0
        }
1795
1796
0
        RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
1797
0
        std::string rowset_meta_str;
1798
0
        bool ret = rowset_meta_pb.SerializeToString(&rowset_meta_str);
1799
0
        if (!ret) {
1800
0
            LOG(WARNING) << "failed to pull rowset for slave replica. serialize rowset meta "
1801
0
                            "failed. rowset_id="
1802
0
                         << rowset_meta_pb.rowset_id()
1803
0
                         << ", tablet_id=" << rowset_meta_pb.tablet_id()
1804
0
                         << ", txn_id=" << rowset_meta_pb.txn_id();
1805
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta_pb.txn_id(),
1806
0
                                        rowset_meta_pb.tablet_id(), node_id, false);
1807
0
            return;
1808
0
        }
1809
0
        bool parsed = rowset_meta->init(rowset_meta_str);
1810
0
        if (!parsed) {
1811
0
            LOG(WARNING) << "failed to pull rowset for slave replica. parse rowset meta string "
1812
0
                            "failed. rowset_id="
1813
0
                         << rowset_meta_pb.rowset_id()
1814
0
                         << ", tablet_id=" << rowset_meta_pb.tablet_id()
1815
0
                         << ", txn_id=" << rowset_meta_pb.txn_id();
1816
            // return false will break meta iterator, return true to skip this error
1817
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1818
0
                                        rowset_meta->tablet_id(), node_id, false);
1819
0
            return;
1820
0
        }
1821
0
        RowsetId remote_rowset_id = rowset_meta->rowset_id();
1822
        // change rowset id because it maybe same as other local rowset
1823
0
        RowsetId new_rowset_id = _engine.next_rowset_id();
1824
0
        auto pending_rs_guard = _engine.pending_local_rowsets().add(new_rowset_id);
1825
0
        rowset_meta->set_rowset_id(new_rowset_id);
1826
0
        rowset_meta->set_tablet_uid(tablet->tablet_uid());
1827
0
        VLOG_CRITICAL << "succeed to init rowset meta for slave replica. rowset_id="
1828
0
                      << rowset_meta->rowset_id() << ", tablet_id=" << rowset_meta->tablet_id()
1829
0
                      << ", txn_id=" << rowset_meta->txn_id();
1830
1831
0
        auto tablet_scheme = rowset_meta->tablet_schema();
1832
0
        for (const auto& segment : segments_size) {
1833
0
            uint64_t file_size = segment.second;
1834
0
            uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024;
1835
0
            if (estimate_timeout < config::download_low_speed_time) {
1836
0
                estimate_timeout = config::download_low_speed_time;
1837
0
            }
1838
1839
0
            std::string remote_file_path =
1840
0
                    local_segment_path(rowset_path, remote_rowset_id.to_string(), segment.first);
1841
0
            std::string remote_file_url =
1842
0
                    construct_url(get_host_port(host, http_port), token, remote_file_path);
1843
1844
0
            std::string local_file_path = local_segment_path(
1845
0
                    tablet->tablet_path(), rowset_meta->rowset_id().to_string(), segment.first);
1846
1847
0
            auto st = download_file_action(remote_file_url, local_file_path, estimate_timeout,
1848
0
                                           file_size);
1849
0
            if (!st.ok()) {
1850
0
                LOG(WARNING) << "failed to pull rowset for slave replica. failed to download "
1851
0
                                "file. url="
1852
0
                             << remote_file_url << ", local_path=" << local_file_path
1853
0
                             << ", txn_id=" << rowset_meta->txn_id();
1854
0
                _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1855
0
                                            rowset_meta->tablet_id(), node_id, false);
1856
0
                return;
1857
0
            }
1858
0
            VLOG_CRITICAL << "succeed to download file for slave replica. url=" << remote_file_url
1859
0
                          << ", local_path=" << local_file_path
1860
0
                          << ", txn_id=" << rowset_meta->txn_id();
1861
0
            if (indices_size.find(segment.first) != indices_size.end()) {
1862
0
                PTabletWriteSlaveRequest_IndexSizeMap segment_indices_size =
1863
0
                        indices_size.at(segment.first);
1864
1865
0
                for (auto index_size : segment_indices_size.index_sizes()) {
1866
0
                    auto index_id = index_size.indexid();
1867
0
                    auto size = index_size.size();
1868
0
                    auto suffix_path = index_size.suffix_path();
1869
0
                    std::string remote_inverted_index_file;
1870
0
                    std::string local_inverted_index_file;
1871
0
                    std::string remote_inverted_index_file_url;
1872
0
                    if (tablet_scheme->get_inverted_index_storage_format() ==
1873
0
                        InvertedIndexStorageFormatPB::V1) {
1874
0
                        remote_inverted_index_file =
1875
0
                                InvertedIndexDescriptor::get_index_file_path_v1(
1876
0
                                        InvertedIndexDescriptor::get_index_file_path_prefix(
1877
0
                                                remote_file_path),
1878
0
                                        index_id, suffix_path);
1879
0
                        remote_inverted_index_file_url = construct_url(
1880
0
                                get_host_port(host, http_port), token, remote_inverted_index_file);
1881
1882
0
                        local_inverted_index_file = InvertedIndexDescriptor::get_index_file_path_v1(
1883
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(
1884
0
                                        local_file_path),
1885
0
                                index_id, suffix_path);
1886
0
                    } else {
1887
0
                        remote_inverted_index_file =
1888
0
                                InvertedIndexDescriptor::get_index_file_path_v2(
1889
0
                                        InvertedIndexDescriptor::get_index_file_path_prefix(
1890
0
                                                remote_file_path));
1891
0
                        remote_inverted_index_file_url = construct_url(
1892
0
                                get_host_port(host, http_port), token, remote_inverted_index_file);
1893
1894
0
                        local_inverted_index_file = InvertedIndexDescriptor::get_index_file_path_v2(
1895
0
                                InvertedIndexDescriptor::get_index_file_path_prefix(
1896
0
                                        local_file_path));
1897
0
                    }
1898
0
                    st = download_file_action(remote_inverted_index_file_url,
1899
0
                                              local_inverted_index_file, estimate_timeout, size);
1900
0
                    if (!st.ok()) {
1901
0
                        LOG(WARNING) << "failed to pull rowset for slave replica. failed to "
1902
0
                                        "download "
1903
0
                                        "file. url="
1904
0
                                     << remote_inverted_index_file_url
1905
0
                                     << ", local_path=" << local_inverted_index_file
1906
0
                                     << ", txn_id=" << rowset_meta->txn_id();
1907
0
                        _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1908
0
                                                    rowset_meta->tablet_id(), node_id, false);
1909
0
                        return;
1910
0
                    }
1911
1912
0
                    VLOG_CRITICAL
1913
0
                            << "succeed to download inverted index file for slave replica. url="
1914
0
                            << remote_inverted_index_file_url
1915
0
                            << ", local_path=" << local_inverted_index_file
1916
0
                            << ", txn_id=" << rowset_meta->txn_id();
1917
0
                }
1918
0
            }
1919
0
        }
1920
1921
0
        RowsetSharedPtr rowset;
1922
0
        Status create_status = RowsetFactory::create_rowset(
1923
0
                tablet->tablet_schema(), tablet->tablet_path(), rowset_meta, &rowset);
1924
0
        if (!create_status) {
1925
0
            LOG(WARNING) << "failed to create rowset from rowset meta for slave replica"
1926
0
                         << ". rowset_id: " << rowset_meta->rowset_id()
1927
0
                         << ", rowset_type: " << rowset_meta->rowset_type()
1928
0
                         << ", rowset_state: " << rowset_meta->rowset_state()
1929
0
                         << ", tablet_id=" << rowset_meta->tablet_id()
1930
0
                         << ", txn_id=" << rowset_meta->txn_id();
1931
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1932
0
                                        rowset_meta->tablet_id(), node_id, false);
1933
0
            return;
1934
0
        }
1935
0
        if (rowset_meta->rowset_state() != RowsetStatePB::COMMITTED) {
1936
0
            LOG(WARNING) << "could not commit txn for slave replica because master rowset state is "
1937
0
                            "not committed, rowset_state="
1938
0
                         << rowset_meta->rowset_state()
1939
0
                         << ", tablet_id=" << rowset_meta->tablet_id()
1940
0
                         << ", txn_id=" << rowset_meta->txn_id();
1941
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1942
0
                                        rowset_meta->tablet_id(), node_id, false);
1943
0
            return;
1944
0
        }
1945
0
        Status commit_txn_status = _engine.txn_manager()->commit_txn(
1946
0
                tablet->data_dir()->get_meta(), rowset_meta->partition_id(), rowset_meta->txn_id(),
1947
0
                rowset_meta->tablet_id(), tablet->tablet_uid(), rowset_meta->load_id(), rowset,
1948
0
                std::move(pending_rs_guard), false);
1949
0
        if (!commit_txn_status && !commit_txn_status.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {
1950
0
            LOG(WARNING) << "failed to add committed rowset for slave replica. rowset_id="
1951
0
                         << rowset_meta->rowset_id() << ", tablet_id=" << rowset_meta->tablet_id()
1952
0
                         << ", txn_id=" << rowset_meta->txn_id();
1953
0
            _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1954
0
                                        rowset_meta->tablet_id(), node_id, false);
1955
0
            return;
1956
0
        }
1957
0
        VLOG_CRITICAL << "succeed to pull rowset for slave replica. successfully to add committed "
1958
0
                         "rowset: "
1959
0
                      << rowset_meta->rowset_id()
1960
0
                      << " to tablet, tablet_id=" << rowset_meta->tablet_id()
1961
0
                      << ", schema_hash=" << rowset_meta->tablet_schema_hash()
1962
0
                      << ", txn_id=" << rowset_meta->txn_id();
1963
0
        _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
1964
0
                                    rowset_meta->tablet_id(), node_id, true);
1965
0
    });
1966
0
    if (!ret) {
1967
0
        offer_failed(response, closure_guard.release(), _heavy_work_pool);
1968
0
        return;
1969
0
    }
1970
0
    Status::OK().to_protobuf(response->mutable_status());
1971
0
}
1972
1973
void PInternalServiceImpl::_response_pull_slave_rowset(const std::string& remote_host,
1974
                                                       int64_t brpc_port, int64_t txn_id,
1975
                                                       int64_t tablet_id, int64_t node_id,
1976
0
                                                       bool is_succeed) {
1977
0
    std::shared_ptr<PBackendService_Stub> stub =
1978
0
            ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(remote_host,
1979
0
                                                                             brpc_port);
1980
0
    if (stub == nullptr) {
1981
0
        LOG(WARNING) << "failed to response result of slave replica to master replica. get rpc "
1982
0
                        "stub failed, master host="
1983
0
                     << remote_host << ", port=" << brpc_port << ", tablet_id=" << tablet_id
1984
0
                     << ", txn_id=" << txn_id;
1985
0
        return;
1986
0
    }
1987
1988
0
    auto request = std::make_shared<PTabletWriteSlaveDoneRequest>();
1989
0
    request->set_txn_id(txn_id);
1990
0
    request->set_tablet_id(tablet_id);
1991
0
    request->set_node_id(node_id);
1992
0
    request->set_is_succeed(is_succeed);
1993
0
    auto pull_rowset_callback = DummyBrpcCallback<PTabletWriteSlaveDoneResult>::create_shared();
1994
0
    auto closure = AutoReleaseClosure<
1995
0
            PTabletWriteSlaveDoneRequest,
1996
0
            DummyBrpcCallback<PTabletWriteSlaveDoneResult>>::create_unique(request,
1997
0
                                                                           pull_rowset_callback);
1998
0
    closure->cntl_->set_timeout_ms(config::slave_replica_writer_rpc_timeout_sec * 1000);
1999
0
    closure->cntl_->ignore_eovercrowded();
2000
0
    stub->response_slave_tablet_pull_rowset(closure->cntl_.get(), closure->request_.get(),
2001
0
                                            closure->response_.get(), closure.get());
2002
0
    closure.release();
2003
2004
0
    pull_rowset_callback->join();
2005
0
    if (pull_rowset_callback->cntl_->Failed()) {
2006
0
        LOG(WARNING) << "failed to response result of slave replica to master replica, error="
2007
0
                     << berror(pull_rowset_callback->cntl_->ErrorCode())
2008
0
                     << ", error_text=" << pull_rowset_callback->cntl_->ErrorText()
2009
0
                     << ", master host: " << remote_host << ", tablet_id=" << tablet_id
2010
0
                     << ", txn_id=" << txn_id;
2011
0
    }
2012
0
    VLOG_CRITICAL << "succeed to response the result of slave replica pull rowset to master "
2013
0
                     "replica. master host: "
2014
0
                  << remote_host << ". is_succeed=" << is_succeed << ", tablet_id=" << tablet_id
2015
0
                  << ", slave server=" << node_id << ", txn_id=" << txn_id;
2016
0
}
2017
2018
void PInternalServiceImpl::response_slave_tablet_pull_rowset(
2019
        google::protobuf::RpcController* controller, const PTabletWriteSlaveDoneRequest* request,
2020
0
        PTabletWriteSlaveDoneResult* response, google::protobuf::Closure* done) {
2021
0
    bool ret = _light_work_pool.try_offer([txn_mgr = _engine.txn_manager(), request, response,
2022
0
                                           done]() {
2023
0
        brpc::ClosureGuard closure_guard(done);
2024
0
        VLOG_CRITICAL << "receive the result of slave replica pull rowset from slave replica. "
2025
0
                         "slave server="
2026
0
                      << request->node_id() << ", is_succeed=" << request->is_succeed()
2027
0
                      << ", tablet_id=" << request->tablet_id() << ", txn_id=" << request->txn_id();
2028
0
        txn_mgr->finish_slave_tablet_pull_rowset(request->txn_id(), request->tablet_id(),
2029
0
                                                 request->node_id(), request->is_succeed());
2030
0
        Status::OK().to_protobuf(response->mutable_status());
2031
0
    });
2032
0
    if (!ret) {
2033
0
        offer_failed(response, done, _heavy_work_pool);
2034
0
        return;
2035
0
    }
2036
0
}
2037
2038
void PInternalService::multiget_data(google::protobuf::RpcController* controller,
2039
                                     const PMultiGetRequest* request, PMultiGetResponse* response,
2040
0
                                     google::protobuf::Closure* done) {
2041
0
    bool ret = _light_work_pool.try_offer([request, response, done]() {
2042
0
        signal::set_signal_task_id(request->query_id());
2043
        // multi get data by rowid
2044
0
        MonotonicStopWatch watch;
2045
0
        watch.start();
2046
0
        brpc::ClosureGuard closure_guard(done);
2047
0
        response->mutable_status()->set_status_code(0);
2048
0
        SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
2049
0
        Status st = RowIdStorageReader::read_by_rowids(*request, response);
2050
0
        st.to_protobuf(response->mutable_status());
2051
0
        LOG(INFO) << "multiget_data finished, cost(us):" << watch.elapsed_time() / 1000;
2052
0
    });
2053
0
    if (!ret) {
2054
0
        offer_failed(response, done, _heavy_work_pool);
2055
0
        return;
2056
0
    }
2057
0
}
2058
2059
void PInternalServiceImpl::get_tablet_rowset_versions(google::protobuf::RpcController* cntl_base,
2060
                                                      const PGetTabletVersionsRequest* request,
2061
                                                      PGetTabletVersionsResponse* response,
2062
0
                                                      google::protobuf::Closure* done) {
2063
0
    brpc::ClosureGuard closure_guard(done);
2064
0
    VLOG_DEBUG << "receive get tablet versions request: " << request->DebugString();
2065
0
    _engine.get_tablet_rowset_versions(request, response);
2066
0
}
2067
2068
void PInternalService::glob(google::protobuf::RpcController* controller,
2069
                            const PGlobRequest* request, PGlobResponse* response,
2070
0
                            google::protobuf::Closure* done) {
2071
0
    bool ret = _heavy_work_pool.try_offer([request, response, done]() {
2072
0
        brpc::ClosureGuard closure_guard(done);
2073
0
        std::vector<io::FileInfo> files;
2074
0
        Status st = io::global_local_filesystem()->safe_glob(request->pattern(), &files);
2075
0
        if (st.ok()) {
2076
0
            for (auto& file : files) {
2077
0
                PGlobResponse_PFileInfo* pfile = response->add_files();
2078
0
                pfile->set_file(file.file_name);
2079
0
                pfile->set_size(file.file_size);
2080
0
            }
2081
0
        }
2082
0
        st.to_protobuf(response->mutable_status());
2083
0
    });
2084
0
    if (!ret) {
2085
0
        offer_failed(response, done, _heavy_work_pool);
2086
0
        return;
2087
0
    }
2088
0
}
2089
2090
void PInternalService::group_commit_insert(google::protobuf::RpcController* controller,
2091
                                           const PGroupCommitInsertRequest* request,
2092
                                           PGroupCommitInsertResponse* response,
2093
0
                                           google::protobuf::Closure* done) {
2094
0
    TUniqueId load_id;
2095
0
    load_id.__set_hi(request->load_id().hi());
2096
0
    load_id.__set_lo(request->load_id().lo());
2097
0
    std::shared_ptr<std::mutex> lock = std::make_shared<std::mutex>();
2098
0
    std::shared_ptr<bool> is_done = std::make_shared<bool>(false);
2099
0
    bool ret = _light_work_pool.try_offer([this, request, response, done, load_id, lock,
2100
0
                                           is_done]() {
2101
0
        brpc::ClosureGuard closure_guard(done);
2102
0
        std::shared_ptr<StreamLoadContext> ctx = std::make_shared<StreamLoadContext>(_exec_env);
2103
0
        auto pipe = std::make_shared<io::StreamLoadPipe>(
2104
0
                io::kMaxPipeBufferedBytes /* max_buffered_bytes */, 64 * 1024 /* min_chunk_size */,
2105
0
                -1 /* total_length */, true /* use_proto */);
2106
0
        ctx->pipe = pipe;
2107
0
        Status st = _exec_env->new_load_stream_mgr()->put(load_id, ctx);
2108
0
        if (st.ok()) {
2109
0
            try {
2110
0
                st = _exec_plan_fragment_impl(
2111
0
                        request->exec_plan_fragment_request().request(),
2112
0
                        request->exec_plan_fragment_request().version(),
2113
0
                        request->exec_plan_fragment_request().compact(),
2114
0
                        [&, response, done, load_id, lock, is_done](RuntimeState* state,
2115
0
                                                                    Status* status) {
2116
0
                            std::lock_guard<std::mutex> lock1(*lock);
2117
0
                            if (*is_done) {
2118
0
                                return;
2119
0
                            }
2120
0
                            *is_done = true;
2121
0
                            brpc::ClosureGuard cb_closure_guard(done);
2122
0
                            response->set_label(state->import_label());
2123
0
                            response->set_txn_id(state->wal_id());
2124
0
                            response->set_loaded_rows(state->num_rows_load_success());
2125
0
                            response->set_filtered_rows(state->num_rows_load_filtered());
2126
0
                            status->to_protobuf(response->mutable_status());
2127
0
                            if (!state->get_error_log_file_path().empty()) {
2128
0
                                response->set_error_url(
2129
0
                                        to_load_error_http_path(state->get_error_log_file_path()));
2130
0
                            }
2131
0
                            _exec_env->new_load_stream_mgr()->remove(load_id);
2132
0
                        });
2133
0
            } catch (const Exception& e) {
2134
0
                st = e.to_status();
2135
0
            } catch (...) {
2136
0
                st = Status::Error(ErrorCode::INTERNAL_ERROR,
2137
0
                                   "_exec_plan_fragment_impl meet unknown error");
2138
0
            }
2139
0
            if (!st.ok()) {
2140
0
                LOG(WARNING) << "exec plan fragment failed, load_id=" << print_id(load_id)
2141
0
                             << ", errmsg=" << st;
2142
0
                std::lock_guard<std::mutex> lock1(*lock);
2143
0
                if (*is_done) {
2144
0
                    closure_guard.release();
2145
0
                } else {
2146
0
                    *is_done = true;
2147
0
                    st.to_protobuf(response->mutable_status());
2148
0
                    _exec_env->new_load_stream_mgr()->remove(load_id);
2149
0
                }
2150
0
            } else {
2151
0
                closure_guard.release();
2152
0
                for (int i = 0; i < request->data().size(); ++i) {
2153
0
                    std::unique_ptr<PDataRow> row(new PDataRow());
2154
0
                    row->CopyFrom(request->data(i));
2155
0
                    st = pipe->append(std::move(row));
2156
0
                    if (!st.ok()) {
2157
0
                        break;
2158
0
                    }
2159
0
                }
2160
0
                if (st.ok()) {
2161
0
                    static_cast<void>(pipe->finish());
2162
0
                }
2163
0
            }
2164
0
        }
2165
0
    });
2166
0
    if (!ret) {
2167
0
        _exec_env->new_load_stream_mgr()->remove(load_id);
2168
0
        offer_failed(response, done, _light_work_pool);
2169
0
        return;
2170
0
    }
2171
0
};
2172
2173
void PInternalService::get_wal_queue_size(google::protobuf::RpcController* controller,
2174
                                          const PGetWalQueueSizeRequest* request,
2175
                                          PGetWalQueueSizeResponse* response,
2176
0
                                          google::protobuf::Closure* done) {
2177
0
    bool ret = _light_work_pool.try_offer([this, request, response, done]() {
2178
0
        brpc::ClosureGuard closure_guard(done);
2179
0
        Status st = Status::OK();
2180
0
        auto table_id = request->table_id();
2181
0
        auto count = _exec_env->wal_mgr()->get_wal_queue_size(table_id);
2182
0
        response->set_size(count);
2183
0
        response->mutable_status()->set_status_code(st.code());
2184
0
    });
2185
0
    if (!ret) {
2186
0
        offer_failed(response, done, _light_work_pool);
2187
0
    }
2188
0
}
2189
2190
void PInternalService::get_be_resource(google::protobuf::RpcController* controller,
2191
                                       const PGetBeResourceRequest* request,
2192
                                       PGetBeResourceResponse* response,
2193
0
                                       google::protobuf::Closure* done) {
2194
0
    bool ret = _light_work_pool.try_offer([response, done]() {
2195
0
        brpc::ClosureGuard closure_guard(done);
2196
0
        int64_t mem_limit = MemInfo::mem_limit();
2197
0
        int64_t mem_usage = PerfCounters::get_vm_rss();
2198
2199
0
        PGlobalResourceUsage* global_resource_usage = response->mutable_global_be_resource_usage();
2200
0
        global_resource_usage->set_mem_limit(mem_limit);
2201
0
        global_resource_usage->set_mem_usage(mem_usage);
2202
2203
0
        Status st = Status::OK();
2204
0
        response->mutable_status()->set_status_code(st.code());
2205
0
    });
2206
0
    if (!ret) {
2207
0
        offer_failed(response, done, _light_work_pool);
2208
0
    }
2209
0
}
2210
2211
} // namespace doris