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