database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[improve](move-memtable) set brpc streaming params in config (#27442)

Browse Source
This commit is contained in:
Kaijie Chen
2023-11-23 14:14:43 +08:00
committed by GitHub
parent dbbed113cf
commit 7a75f8c380
3 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
be/src/common/config.cpp
View File

@ -746,6 +746,13 @@ DEFINE_Bool(share_delta_writers, "true");
// timeout for open load stream rpc in ms
DEFINE_Int64(open_load_stream_timeout_ms, "500");
// idle timeout for load stream in ms
DEFINE_Int64(load_stream_idle_timeout_ms, "600000");
// brpc streaming max_buf_size in bytes
DEFINE_Int64(load_stream_max_buf_size, "20971520"); // 20MB
// brpc streaming messages_in_batch
DEFINE_Int32(load_stream_messages_in_batch, "128");
// max send batch parallelism for OlapTableSink
// The value set by the user for send_batch_parallelism is not allowed to exceed max_send_batch_parallelism_per_job,
// if exceed, the value of send_batch_parallelism would be max_send_batch_parallelism_per_job

7
be/src/common/config.h
View File

@ -803,6 +803,13 @@ DECLARE_Bool(share_delta_writers);
// timeout for open load stream rpc in ms
DECLARE_Int64(open_load_stream_timeout_ms);
// idle timeout for load stream in ms
DECLARE_Int64(load_stream_idle_timeout_ms);
// brpc streaming max_buf_size in bytes
DECLARE_Int64(load_stream_max_buf_size);
// brpc streaming messages_in_batch
DECLARE_Int32(load_stream_messages_in_batch);
// max send batch parallelism for OlapTableSink
// The value set by the user for send_batch_parallelism is not allowed to exceed max_send_batch_parallelism_per_job,
// if exceed, the value of send_batch_parallelism would be max_send_batch_parallelism_per_job

6
be/src/vec/sink/load_stream_stub.cpp
View File

@ -134,9 +134,9 @@ Status LoadStreamStub::open(BrpcClientCache<PBackendService_Stub>* client_cache,
_handler.set_load_id(_load_id);
std::string host_port = get_host_port(node_info.host, node_info.brpc_port);
brpc::StreamOptions opt;
opt.max_buf_size = 20 << 20; // 20MB
opt.idle_timeout_ms = 30000;
opt.messages_in_batch = 128;
opt.max_buf_size = config::load_stream_max_buf_size;
opt.idle_timeout_ms = config::load_stream_idle_timeout_ms;
opt.messages_in_batch = config::load_stream_messages_in_batch;
opt.handler = &_handler;
brpc::Controller cntl;
if (int ret = StreamCreate(&_stream_id, cntl, &opt)) {