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 b61927b78c8f327bcf02fb80fb4e94fc2123885e Author: Hu Shenggang <[email protected]> AuthorDate: Tue Jul 14 15:44:05 2026 +0800 Revert "[opt](scan) Boost scanner concurrency under starvation" This reverts commit 7261a1e48beb0940bc5c9a016f02bda65fbfe895. --- 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, 9 insertions(+), 53 deletions(-) diff --git a/be/src/exec/scan/scanner_context.cpp b/be/src/exec/scan/scanner_context.cpp index 3371af5bf8e..f7376fd2946 100644 --- a/be/src/exec/scan/scanner_context.cpp +++ b/be/src/exec/scan/scanner_context.cpp @@ -166,7 +166,6 @@ 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); @@ -258,7 +257,6 @@ 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(); } @@ -318,13 +316,9 @@ 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)); } @@ -339,12 +333,7 @@ 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(); @@ -380,7 +369,6 @@ 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()) { @@ -463,13 +451,8 @@ 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) { - 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 - + // 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 - (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. @@ -491,11 +474,10 @@ int32_t ScannerContext::_get_margin(std::unique_lock<std::mutex>& transfer_lock, VLOG_DEBUG << fmt::format( "[{}|{}] schedule scan task, margin_1: {} = {} - ({} + {}), margin_2: {} = {} - " - "({} + {}), margin: {}, scan starving: {}", - print_id(_query_id), ctx_id, margin_1, target_scan_concurrency, _tasks_queue.size(), + "({} + {}), margin: {}", + print_id(_query_id), ctx_id, margin_1, _min_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, - _scan_starving); + _scanner_scheduler->get_active_threads(), _scanner_scheduler->get_queue_size(), margin); return margin; } diff --git a/be/src/exec/scan/scanner_context.h b/be/src/exec/scan/scanner_context.h index 857dfb736e9..553408ebc96 100644 --- a/be/src/exec/scan/scanner_context.h +++ b/be/src/exec/scan/scanner_context.h @@ -251,8 +251,7 @@ 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; - int32_t _min_scan_concurrency = 1; - bool _scan_starving = false; + MOCK_REMOVE(const) int32_t _min_scan_concurrency = 1; 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 a6d35f3a0b7..c116c354f10 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 = default_local_scan_thread_num(); + : config::min_active_scan_threads = CpuInfo::num_cores() * 2; } 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 8b2097716de..35156865bb3 100644 --- a/be/test/exec/scan/scanner_context_test.cpp +++ b/be/test/exec/scan/scanner_context_test.cpp @@ -423,31 +423,6 @@ 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 34e82825c48..4a7253db60c 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 扫描数据的最大并发，默认为 8", "The max threads to read data of ScanNode, default 8"}) - public int maxScannersConcurrency = 8; + "ScanNode 扫描数据的最大并发，默认为 4", "The max threads to read data of ScanNode, default 4"}) + public int maxScannersConcurrency = 4; @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]
