This is an automated email from the ASF dual-hosted git repository. mrhhsg pushed a commit to branch opt_perf_4.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 3d74daa948890c228b37faab7c6464cfa56037c2 Author: Hu Shenggang <[email protected]> AuthorDate: Fri Jul 17 16:30:17 2026 +0800 Reapply "[opt](scan) Boost scanner concurrency under starvation" This reverts commit b61927b78c8f327bcf02fb80fb4e94fc2123885e. --- be/src/exec/scan/scanner_context.cpp | 28 ++++++++++++++++++---- be/src/exec/scan/scanner_context.h | 3 ++- be/src/exec/scan/scanner_scheduler.cpp | 2 +- be/test/exec/scan/scanner_context_test.cpp | 25 +++++++++++++++++++ .../java/org/apache/doris/qe/SessionVariable.java | 4 ++-- 5 files changed, 53 insertions(+), 9 deletions(-) diff --git a/be/src/exec/scan/scanner_context.cpp b/be/src/exec/scan/scanner_context.cpp index f7376fd2946..3371af5bf8e 100644 --- a/be/src/exec/scan/scanner_context.cpp +++ b/be/src/exec/scan/scanner_context.cpp @@ -166,6 +166,7 @@ Status ScannerContext::init() { } } } + _min_scan_concurrency = std::min(_min_scan_concurrency, _max_scan_concurrency); COUNTER_SET(_local_state->_max_scan_concurrency, (int64_t)_max_scan_concurrency); COUNTER_SET(_local_state->_min_scan_concurrency, (int64_t)_min_scan_concurrency); @@ -257,6 +258,7 @@ void ScannerContext::push_back_scan_task(std::shared_ptr<ScanTask> scan_task) { } _tasks_queue.push_back(scan_task); _num_scheduled_scanners--; + _scan_starving = false; _dependency->set_ready(); } @@ -316,9 +318,13 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, Block* block, b if (scan_task->is_eos()) { // 1. if eos, record a finished scanner. _num_finished_scanners++; + _scan_starving = _tasks_queue.empty() && + _num_finished_scanners < cast_set<int32_t>(_all_scanners.size()) && + (_num_scheduled_scanners > 0 || !_pending_scanners.empty()); RETURN_IF_ERROR( _scanner_scheduler->schedule_scan_task(shared_from_this(), nullptr, l)); } else { + _scan_starving = _tasks_queue.empty(); RETURN_IF_ERROR( _scanner_scheduler->schedule_scan_task(shared_from_this(), scan_task, l)); } @@ -333,7 +339,12 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, Block* block, b *eos = done(); if (_tasks_queue.empty()) { + _scan_starving = !done() && + _num_finished_scanners < cast_set<int32_t>(_all_scanners.size()) && + (_num_scheduled_scanners > 0 || !_pending_scanners.empty()); _dependency->block(); + } else { + _scan_starving = false; } return Status::OK(); @@ -369,6 +380,7 @@ void ScannerContext::stop_scanners(RuntimeState* state) { return; } _should_stop = true; + _scan_starving = false; _set_scanner_done(); for (const std::weak_ptr<ScannerDelegate>& scanner : _all_scanners) { if (std::shared_ptr<ScannerDelegate> sc = scanner.lock()) { @@ -451,8 +463,13 @@ void ScannerContext::update_peak_running_scanner(int num) { int32_t ScannerContext::_get_margin(std::unique_lock<std::mutex>& transfer_lock, std::unique_lock<std::shared_mutex>& scheduler_lock) { - // margin_1 is used to ensure each scan operator could have at least _min_scan_concurrency scan tasks. - int32_t margin_1 = _min_scan_concurrency - + int32_t target_scan_concurrency = _min_scan_concurrency; + if (_scan_starving && _tasks_queue.empty()) { + target_scan_concurrency = _max_scan_concurrency; + } + + // margin_1 is used to ensure each scan operator could have enough scan tasks. + int32_t margin_1 = target_scan_concurrency - (cast_set<int32_t>(_tasks_queue.size()) + _num_scheduled_scanners); // margin_2 is used to ensure the scan scheduler could have at least _min_scan_concurrency_of_scan_scheduler scan tasks. @@ -474,10 +491,11 @@ int32_t ScannerContext::_get_margin(std::unique_lock<std::mutex>& transfer_lock, VLOG_DEBUG << fmt::format( "[{}|{}] schedule scan task, margin_1: {} = {} - ({} + {}), margin_2: {} = {} - " - "({} + {}), margin: {}", - print_id(_query_id), ctx_id, margin_1, _min_scan_concurrency, _tasks_queue.size(), + "({} + {}), margin: {}, scan starving: {}", + print_id(_query_id), ctx_id, margin_1, target_scan_concurrency, _tasks_queue.size(), _num_scheduled_scanners, margin_2, _min_scan_concurrency_of_scan_scheduler, - _scanner_scheduler->get_active_threads(), _scanner_scheduler->get_queue_size(), margin); + _scanner_scheduler->get_active_threads(), _scanner_scheduler->get_queue_size(), margin, + _scan_starving); return margin; } diff --git a/be/src/exec/scan/scanner_context.h b/be/src/exec/scan/scanner_context.h index 553408ebc96..857dfb736e9 100644 --- a/be/src/exec/scan/scanner_context.h +++ b/be/src/exec/scan/scanner_context.h @@ -251,7 +251,8 @@ protected: // Each scan operator can submit _max_scan_concurrency scanner to scheduelr if scheduler has enough resource. // So that for a single query, we can make sure it could make full utilization of the resource. int32_t _max_scan_concurrency = 0; - MOCK_REMOVE(const) int32_t _min_scan_concurrency = 1; + int32_t _min_scan_concurrency = 1; + bool _scan_starving = false; std::shared_ptr<ScanTask> _pull_next_scan_task(std::shared_ptr<ScanTask> current_scan_task, int32_t current_concurrency); diff --git a/be/src/exec/scan/scanner_scheduler.cpp b/be/src/exec/scan/scanner_scheduler.cpp index c116c354f10..a6d35f3a0b7 100644 --- a/be/src/exec/scan/scanner_scheduler.cpp +++ b/be/src/exec/scan/scanner_scheduler.cpp @@ -397,7 +397,7 @@ int ScannerScheduler::get_remote_scan_thread_queue_size() { int ScannerScheduler::default_min_active_scan_threads() { return config::min_active_scan_threads > 0 ? config::min_active_scan_threads - : config::min_active_scan_threads = CpuInfo::num_cores() * 2; + : config::min_active_scan_threads = default_local_scan_thread_num(); } int ScannerScheduler::default_min_active_file_scan_threads() { diff --git a/be/test/exec/scan/scanner_context_test.cpp b/be/test/exec/scan/scanner_context_test.cpp index 35156865bb3..8b2097716de 100644 --- a/be/test/exec/scan/scanner_context_test.cpp +++ b/be/test/exec/scan/scanner_context_test.cpp @@ -423,6 +423,31 @@ TEST_F(ScannerContextTest, get_margin) { scanner_context->_num_scheduled_scanners = 20; margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); ASSERT_EQ(margin, 0); + + // Downstream is waiting for scan data while scheduler is busy. The scan operator should + // refill up to max scan concurrency instead of being limited by min scan concurrency. + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(50)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(10)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_min_scan_concurrency_of_scan_scheduler = 20; + scanner_context->_num_scheduled_scanners = 0; + scanner_context->_min_scan_concurrency = 1; + scanner_context->_max_scan_concurrency = 8; + scanner_context->_scan_starving = true; + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, scanner_context->_max_scan_concurrency); + + // If there is already a produced task waiting in the queue, downstream is not scan-starved. + scheduler = std::make_unique<MockSimplifiedScanScheduler>(cgroup_cpu_ctl); + EXPECT_CALL(*scheduler, get_active_threads()).WillOnce(testing::Return(50)); + EXPECT_CALL(*scheduler, get_queue_size()).WillOnce(testing::Return(10)); + scanner_context->_scanner_scheduler = scheduler.get(); + scanner_context->_tasks_queue.push_back( + std::make_shared<ScanTask>(std::make_shared<ScannerDelegate>(scanner))); + margin = scanner_context->_get_margin(transfer_lock, scheduler_lock); + ASSERT_EQ(margin, 0); + scanner_context->_tasks_queue.clear(); } TEST_F(ScannerContextTest, pull_next_scan_task) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 4a7253db60c..34e82825c48 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -1112,8 +1112,8 @@ public class SessionVariable implements Serializable, Writable { public long maxScanQueueMemByte = 2147483648L / 20; @VariableMgr.VarAttr(name = MAX_SCANNERS_CONCURRENCY, needForward = true, description = { - "ScanNode 扫描数据的最大并发，默认为 4", "The max threads to read data of ScanNode, default 4"}) - public int maxScannersConcurrency = 4; + "ScanNode 扫描数据的最大并发，默认为 8", "The max threads to read data of ScanNode, default 8"}) + public int maxScannersConcurrency = 8; @VariableMgr.VarAttr(name = MAX_FILE_SCANNERS_CONCURRENCY, needForward = true, description = { "FileScanNode 扫描数据的最大并发，默认为 16", "The max threads to read data of FileScanNode, default 16"}) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
