diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index 830cb4cc4e2ed4..442bb7b205ae9d 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -34,7 +34,9 @@ #include "common/config.h" #include "common/logging.h" +#include "runtime/exec_env.h" #include "s3_uri.h" +#include "vec/exec/scan/scanner_scheduler.h" namespace doris { @@ -162,7 +164,14 @@ std::shared_ptr S3ClientFactory::create(const S3Conf& s3_conf if (s3_conf.max_connections > 0) { aws_config.maxConnections = s3_conf.max_connections; } else { - aws_config.maxConnections = config::doris_remote_scanner_thread_pool_thread_num; +#ifdef BE_TEST + // the S3Client may shared by many threads. + // So need to set the number of connections large enough. + aws_config.maxConnections = config::doris_scanner_thread_pool_thread_num; +#else + aws_config.maxConnections = + ExecEnv::GetInstance()->scanner_scheduler()->remote_thread_pool_max_size(); +#endif } if (s3_conf.request_timeout_ms > 0) { diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp index e78e8dceffecde..6c40ccc242e135 100644 --- a/be/src/vec/exec/scan/scanner_scheduler.cpp +++ b/be/src/vec/exec/scan/scanner_scheduler.cpp @@ -123,13 +123,13 @@ Status ScannerScheduler::init(ExecEnv* env) { config::doris_scanner_thread_pool_queue_size, "local_scan"); // 3. remote scan thread pool + _remote_thread_pool_max_size = config::doris_max_remote_scanner_thread_pool_thread_num != -1 + ? config::doris_max_remote_scanner_thread_pool_thread_num + : std::max(512, CpuInfo::num_cores() * 10); static_cast( ThreadPoolBuilder("RemoteScanThreadPool") .set_min_threads(config::doris_scanner_thread_pool_thread_num) // 48 default - .set_max_threads( - config::doris_max_remote_scanner_thread_pool_thread_num != -1 - ? config::doris_max_remote_scanner_thread_pool_thread_num - : std::max(512, CpuInfo::num_cores() * 10)) + .set_max_threads(_remote_thread_pool_max_size) .set_max_queue_size(config::doris_scanner_thread_pool_queue_size) .build(&_remote_scan_thread_pool)); @@ -504,4 +504,4 @@ void ScannerScheduler::_deregister_metrics() { DEREGISTER_HOOK_METRIC(group_local_scan_thread_pool_thread_num); } -} // namespace doris::vectorized \ No newline at end of file +} // namespace doris::vectorized diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h index a6d450bc221141..9825085fb1c03f 100644 --- a/be/src/vec/exec/scan/scanner_scheduler.h +++ b/be/src/vec/exec/scan/scanner_scheduler.h @@ -76,6 +76,8 @@ class ScannerScheduler { return _task_group_local_scan_queue.get(); } + int remote_thread_pool_max_size() const { return _remote_thread_pool_max_size; } + private: // scheduling thread function void _schedule_thread(int queue_id); @@ -118,6 +120,7 @@ class ScannerScheduler { // true is the scheduler is closed. std::atomic_bool _is_closed = {false}; bool _is_init = false; + int _remote_thread_pool_max_size; }; struct SimplifiedScanTask {