update

Author: Gabriel39

be/src/pipeline/exec/file_scan_operator.cpp

@@ -32,21 +32,39 @@
 namespace doris::pipeline {
 #include "common/compile_check_begin.h"
 
-int FileScanLocalState::max_scanners_concurrency(RuntimeState* state) {
+int FileScanLocalState::max_scanners_concurrency(RuntimeState* state) const {
+    // For select * from table limit 10; should just use one thread.
+    if (should_run_serial()) {
+        return 1;
+    }
     /*
      * The max concurrency of file scanners for each FileScanLocalState is determined by:
      * 1. User specified max_file_scanners_concurrency which is set through session variable.
-     * 2. Default: std::max((std::max(512, CpuInfo::num_cores() * 10)), config::doris_scanner_thread_pool_thread_num) * 2 / operator parallelism
+     * 2. Default: std::max(512, CpuInfo::num_cores() * 10)
      *
      * If this is a serial operator, the max concurrency should multiply by the number of parallel instances of the operator.
      */
     return (state->max_file_scanners_concurrency()
                     ? state->max_file_scanners_concurrency()
-                    : state->get_query_ctx()->get_remote_scan_scheduler()->get_max_threads() * 2 /
-                              _parent->parallelism(state)) *
+                    : std::max(512, CpuInfo::num_cores() * 10)) *
            (state->query_parallel_instance_num() / _parent->parallelism(state));
 }
 
+int FileScanLocalState::min_scanners_concurrency(RuntimeState* state) const {
+    /*
+     * The min concurrency of scanners for each ScanLocalStateBase is determined by:
+     * 1. User specified min_scan_concurrency_of_scan_scheduler which is set through session variable.
+     * 2. Default: 2 * state->get_query_ctx()->get_remote_scan_scheduler()->get_max_threads()
+     */
+    return state->min_scan_concurrency_of_scan_scheduler()
+                   ? state->min_scan_concurrency_of_scan_scheduler()
+                   : 2 * state->get_query_ctx()->get_remote_scan_scheduler()->get_max_threads();
+}
+
+vectorized::SimplifiedScanScheduler* FileScanLocalState::scan_scheduler(RuntimeState* state) const {
+    return state->get_query_ctx()->get_remote_scan_scheduler();
+}
+
 Status FileScanLocalState::_init_scanners(std::list<vectorized::ScannerSPtr>* scanners) {
     if (_split_source->num_scan_ranges() == 0) {
         _eos = true;

be/src/pipeline/exec/file_scan_operator.h

@@ -54,7 +54,9 @@ class FileScanLocalState final : public ScanLocalState<FileScanLocalState> {
                          const std::vector<TScanRangeParams>& scan_ranges) override;
     int parent_id() { return _parent->node_id(); }
     std::string name_suffix() const override;
-    int max_scanners_concurrency(RuntimeState* state) override;
+    int max_scanners_concurrency(RuntimeState* state) const override;
+    int min_scanners_concurrency(RuntimeState* state) const override;
+    vectorized::SimplifiedScanScheduler* scan_scheduler(RuntimeState* state) const override;
 
 private:
     friend class vectorized::FileScanner;

be/src/pipeline/exec/scan_operator.cpp

@@ -72,21 +72,39 @@ bool ScanLocalState<Derived>::should_run_serial() const {
     return _parent->cast<typename Derived::Parent>()._should_run_serial;
 }
 
-int ScanLocalStateBase::max_scanners_concurrency(RuntimeState* state) {
+int ScanLocalStateBase::max_scanners_concurrency(RuntimeState* state) const {
+    // For select * from table limit 10; should just use one thread.
+    if (should_run_serial()) {
+        return 1;
+    }
     /*
      * The max concurrency of scanners for each ScanLocalStateBase is determined by:
      * 1. User specified max_scanners_concurrency which is set through session variable.
-     * 2. Default: config::doris_scanner_thread_pool_thread_num / operator parallelism
+     * 2. Default: std::max(48, CpuInfo::num_cores() * 2))
      *
      * If this is a serial operator, the max concurrency should multiply by the number of parallel instances of the operator.
      */
     return (state->max_scanners_concurrency()
                     ? state->max_scanners_concurrency()
-                    : _state->get_query_ctx()->get_scan_scheduler()->get_max_threads() * 2 /
-                              _parent->parallelism(state)) *
+                    : std::max(48, CpuInfo::num_cores() * 2)) *
            (state->query_parallel_instance_num() / _parent->parallelism(state));
 }
 
+int ScanLocalStateBase::min_scanners_concurrency(RuntimeState* state) const {
+    /*
+     * The min concurrency of scanners for each ScanLocalStateBase is determined by:
+     * 1. User specified min_scan_concurrency_of_scan_scheduler which is set through session variable.
+     * 2. Default: 2 * state->get_query_ctx()->get_scan_scheduler()->get_max_threads()
+     */
+    return state->min_scan_concurrency_of_scan_scheduler()
+                   ? state->min_scan_concurrency_of_scan_scheduler()
+                   : 2 * state->get_query_ctx()->get_scan_scheduler()->get_max_threads();
+}
+
+vectorized::SimplifiedScanScheduler* ScanLocalStateBase::scan_scheduler(RuntimeState* state) const {
+    return state->get_query_ctx()->get_scan_scheduler();
+}
+
 template <typename Derived>
 Status ScanLocalState<Derived>::init(RuntimeState* state, LocalStateInfo& info) {
     RETURN_IF_ERROR(PipelineXLocalState<>::init(state, info));

be/src/pipeline/exec/scan_operator.h

@@ -90,7 +90,9 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
     // Related pr:
     // https://github.com/apache/doris/pull/42460
     // https://github.com/apache/doris/pull/44635
-    [[nodiscard]] virtual int max_scanners_concurrency(RuntimeState* state);
+    [[nodiscard]] virtual int max_scanners_concurrency(RuntimeState* state) const;
+    [[nodiscard]] virtual int min_scanners_concurrency(RuntimeState* state) const;
+    [[nodiscard]] virtual vectorized::SimplifiedScanScheduler* scan_scheduler(RuntimeState* state) const;
 
     [[nodiscard]] std::string get_name() { return _parent->get_name(); }
 

be/src/vec/exec/scan/scanner_context.cpp

@@ -73,14 +73,16 @@ ScannerContext::ScannerContext(
           limit(limit_),
           _scanner_scheduler_global(state->exec_env()->scanner_scheduler()),
           _all_scanners(scanners.begin(), scanners.end()),
-          _min_scan_concurrency_of_scan_scheduler(_state->min_scan_concurrency_of_scan_scheduler()),
-          _min_scan_concurrency(_state->min_scan_concurrency_of_scanner()) {
 #ifndef BE_TEST
-    _max_scan_concurrency =
-            std::min(local_state->max_scanners_concurrency(state), cast_set<int>(scanners.size()));
+          _scanner_scheduler(local_state->scan_scheduler(state)),
+          _min_scan_concurrency_of_scan_scheduler(local_state->min_scanners_concurrency(state)),
+          _max_scan_concurrency(std::min(local_state->max_scanners_concurrency(state), cast_set<int>(scanners.size()))),
 #else
-    _max_scan_concurrency = num_parallel_instances;
+          _scanner_scheduler(state->get_query_ctx()->get_scan_scheduler()),
+          _min_scan_concurrency_of_scan_scheduler(0),
+          _max_scan_concurrency(num_parallel_instances),
 #endif
+          _min_scan_concurrency(local_state->should_run_serial() ? 1 : std::min(_state->min_scan_concurrency_of_scanner(), _max_scan_concurrency)) {
     DCHECK(_state != nullptr);
     DCHECK(_output_row_descriptor == nullptr ||
            _output_row_descriptor->tuple_descriptors().size() == 1);
@@ -117,16 +119,8 @@ Status ScannerContext::init() {
     auto scanner = _all_scanners.front().lock();
     DCHECK(scanner != nullptr);
 
-    // TODO: Maybe need refactor.
-    // A query could have remote scan task and local scan task at the same time.
-    // So we need to compute the _scanner_scheduler in each scan operator instead of query context.
-    if (scanner->_scanner->get_storage_type() == TabletStorageType::STORAGE_TYPE_LOCAL) {
-        _scanner_scheduler = _state->get_query_ctx()->get_scan_scheduler();
-    } else {
-        _scanner_scheduler = _state->get_query_ctx()->get_remote_scan_scheduler();
-    }
-    if (auto* task_executor_scheduler =
-                dynamic_cast<TaskExecutorSimplifiedScanScheduler*>(_scanner_scheduler)) {
+    if (const auto* task_executor_scheduler =
+                dynamic_cast<const TaskExecutorSimplifiedScanScheduler*>(_scanner_scheduler)) {
         std::shared_ptr<TaskExecutor> task_executor = task_executor_scheduler->task_executor();
         vectorized::TaskId task_id(fmt::format("{}-{}", print_id(_state->query_id()), ctx_id));
         _task_handle = DORIS_TRY(task_executor->create_task(
@@ -143,11 +137,6 @@ Status ScannerContext::init() {
     // Provide more memory for wide tables, increase proportionally by multiples of 300
     _max_bytes_in_queue *= _output_tuple_desc->slots().size() / 300 + 1;
 
-    if (_min_scan_concurrency_of_scan_scheduler == 0) {
-        // _scanner_scheduler->get_max_threads() is setted by workload group.
-        _min_scan_concurrency_of_scan_scheduler = 2 * _scanner_scheduler->get_max_threads();
-    }
-
     if (_all_scanners.empty()) {
         _is_finished = true;
         _set_scanner_done();
@@ -175,15 +164,6 @@ Status ScannerContext::init() {
         }
     }
 
-    // For select * from table limit 10; should just use one thread.
-    if (_local_state->should_run_serial()) {
-        _max_scan_concurrency = 1;
-        _min_scan_concurrency = 1;
-    }
-
-    // Avoid corner case.
-    _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);
 

be/src/vec/exec/scan/scanner_context.h

@@ -169,7 +169,7 @@ class ScannerContext : public std::enable_shared_from_this<ScannerContext>,
 
     RuntimeState* state() { return _state; }
 
-    SimplifiedScanScheduler* get_scan_scheduler() { return _scanner_scheduler; }
+    SimplifiedScanScheduler* get_scan_scheduler() const { return _scanner_scheduler; }
 
     void stop_scanners(RuntimeState* state);
 
@@ -234,7 +234,6 @@ class ScannerContext : public std::enable_shared_from_this<ScannerContext>,
 
     int64_t _max_bytes_in_queue = 0;
     doris::vectorized::ScannerScheduler* _scanner_scheduler_global = nullptr;
-    SimplifiedScanScheduler* _scanner_scheduler = nullptr;
     // Using stack so that we can resubmit scanner in a LIFO order, maybe more cache friendly
     std::stack<std::shared_ptr<ScanTask>> _pending_scanners;
     // Scanner that is submitted to the scheduler.
@@ -256,13 +255,14 @@ class ScannerContext : public std::enable_shared_from_this<ScannerContext>,
 
     // adaptive scan concurrency related
 
-    int32_t _min_scan_concurrency_of_scan_scheduler = 0;
-    int32_t _min_scan_concurrency = 1;
+    SimplifiedScanScheduler* _scanner_scheduler = nullptr;
+    const int32_t _min_scan_concurrency_of_scan_scheduler = 0;
     // The overall target of our system is to make full utilization of the resources.
     // At the same time, we dont want too many tasks are queued by scheduler, that is not necessary.
     // 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;
+    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);

be/src/vec/exec/scan/scanner_scheduler.cpp

@@ -90,7 +90,7 @@ Status ScannerScheduler::submit(std::shared_ptr<ScannerContext> ctx,
     scanner_delegate->_scanner->start_wait_worker_timer();
     TabletStorageType type = scanner_delegate->_scanner->get_storage_type();
     auto sumbit_task = [&]() {
-        SimplifiedScanScheduler* scan_sched = ctx->get_scan_scheduler();
+        auto* scan_sched = ctx->get_scan_scheduler();
         auto work_func = [scanner_ref = scan_task, ctx]() {
             auto status = [&] {
                 RETURN_IF_CATCH_EXCEPTION(_scanner_scan(ctx, scanner_ref));