This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch opt_perf_4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/opt_perf_4.1 by this push:
new 7261a1e48be [opt](scan) Boost scanner concurrency under starvation
7261a1e48be is described below
commit 7261a1e48beb0940bc5c9a016f02bda65fbfe895
Author: happenlee <[email protected]>
AuthorDate: Thu Jul 9 20:07:57 2026 +0800
[opt](scan) Boost scanner concurrency under starvation
---
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 c930696743c..7ec1074bab1 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
@@ -1104,8 +1104,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]
