[Improvement](external) Increase scanner concurrency

Author: Gabriel39

be/src/pipeline/exec/file_scan_operator.cpp

@@ -31,6 +31,39 @@
 
 namespace doris::pipeline {
 #include "common/compile_check_begin.h"
+
+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(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()
+                                                   : 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::ScannerScheduler* 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;
@@ -44,9 +77,9 @@ Status FileScanLocalState::_init_scanners(std::list<vectorized::ScannerSPtr>* sc
 
     auto& p = _parent->cast<FileScanOperatorX>();
     // There's only one scan range for each backend in batch split mode. Each backend only starts up one ScanNode instance.
-    uint32_t shard_num = std::min(vectorized::ScannerScheduler::get_remote_scan_thread_num() /
-                                          p.query_parallel_instance_num(),
-                                  _max_scanners);
+    uint32_t shard_num = std::min(
+            vectorized::ScannerScheduler::get_remote_scan_thread_num() / p.parallelism(state()),
+            _max_scanners);
     shard_num = std::max(shard_num, 1U);
     _kv_cache.reset(new vectorized::ShardedKVCache(shard_num));
     for (int i = 0; i < _max_scanners; ++i) {
@@ -85,15 +118,15 @@ void FileScanLocalState::set_scan_ranges(RuntimeState* state,
             auto split_source = scan_range.split_source;
             RuntimeProfile::Counter* get_split_timer = ADD_TIMER(custom_profile(), "GetSplitTime");
 
-            _max_scanners = calc_max_scanners(p.query_parallel_instance_num());
+            _max_scanners = calc_max_scanners(p.parallelism(state));
             _split_source = std::make_shared<vectorized::RemoteSplitSourceConnector>(
                     state, get_split_timer, split_source.split_source_id, split_source.num_splits,
                     _max_scanners);
         }
     }
 
     if (!p._batch_split_mode) {
-        _max_scanners = calc_max_scanners(p.query_parallel_instance_num());
+        _max_scanners = calc_max_scanners(p.parallelism(state));
         if (_split_source == nullptr) {
             _split_source = std::make_shared<vectorized::LocalSplitSourceConnector>(scan_ranges,
                                                                                     _max_scanners);

be/src/pipeline/exec/file_scan_operator.h

@@ -54,6 +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) const override;
+    int min_scanners_concurrency(RuntimeState* state) const override;
+    vectorized::ScannerScheduler* scan_scheduler(RuntimeState* state) const override;
 
 private:
     friend class vectorized::FileScanner;
@@ -83,8 +86,8 @@ class FileScanOperatorX final : public ScanOperatorX<FileScanLocalState> {
     bool is_file_scan_operator() const override { return true; }
 
     // There's only one scan range for each backend in batch split mode. Each backend only starts up one ScanNode instance.
-    int query_parallel_instance_num() const override {
-        return _batch_split_mode ? 1 : _query_parallel_instance_num;
+    int parallelism(RuntimeState* state) const override {
+        return _batch_split_mode ? 1 : ScanOperatorX<FileScanLocalState>::parallelism(state);
     }
 
 private:

be/src/pipeline/exec/operator.h

@@ -113,6 +113,9 @@ class OperatorBase {
     [[nodiscard]] virtual Status terminate(RuntimeState* state) = 0;
     [[nodiscard]] virtual Status close(RuntimeState* state);
     [[nodiscard]] virtual int node_id() const = 0;
+    [[nodiscard]] virtual int parallelism(RuntimeState* state) const {
+        return _is_serial_operator ? 1 : state->query_parallel_instance_num();
+    }
 
     [[nodiscard]] virtual Status set_child(OperatorPtr child) {
         if (_child && child != nullptr) {

be/src/pipeline/exec/scan_operator.cpp

@@ -72,6 +72,38 @@ bool ScanLocalState<Derived>::should_run_serial() const {
     return _parent->cast<typename Derived::Parent>()._should_run_serial;
 }
 
+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: 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()
+                                              : 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::ScannerScheduler* 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));
@@ -1052,19 +1084,14 @@ template <typename Derived>
 Status ScanLocalState<Derived>::_start_scanners(
         const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners) {
     auto& p = _parent->cast<typename Derived::Parent>();
-    // If scan operator is serial operator(like topn), its real parallelism is 1.
-    // Otherwise, its real parallelism is query_parallel_instance_num.
-    // query_parallel_instance_num of olap table is usually equal to session var parallel_pipeline_task_num.
-    // for file scan operator, its real parallelism will be 1 if it is in batch mode.
-    // Related pr:
-    // https://github.com/apache/doris/pull/42460
-    // https://github.com/apache/doris/pull/44635
-    const int parallism_of_scan_operator =
-            p.is_serial_operator() ? 1 : p.query_parallel_instance_num();
-
-    _scanner_ctx = vectorized::ScannerContext::create_shared(
-            state(), this, p._output_tuple_desc, p.output_row_descriptor(), scanners, p.limit(),
-            _scan_dependency, parallism_of_scan_operator);
+    _scanner_ctx = vectorized::ScannerContext::create_shared(state(), this, p._output_tuple_desc,
+                                                             p.output_row_descriptor(), scanners,
+                                                             p.limit(), _scan_dependency
+#ifdef BE_TEST
+                                                             ,
+                                                             max_scanners_concurrency(state())
+#endif
+    );
     return Status::OK();
 }
 
@@ -1273,8 +1300,6 @@ Status ScanOperatorX<LocalStateType>::init(const TPlanNode& tnode, RuntimeState*
         }
     }
 
-    _query_parallel_instance_num = state->query_parallel_instance_num();
-
     return Status::OK();
 }
 

be/src/pipeline/exec/scan_operator.h

@@ -83,6 +83,16 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
     virtual TPushAggOp::type get_push_down_agg_type() = 0;
 
     virtual int64_t get_push_down_count() = 0;
+    // If scan operator is serial operator(like topn), its real parallelism is 1.
+    // Otherwise, its real parallelism is query_parallel_instance_num.
+    // query_parallel_instance_num of olap table is usually equal to session var parallel_pipeline_task_num.
+    // for file scan operator, its real parallelism will be 1 if it is in batch mode.
+    // Related pr:
+    // https://github.com/apache/doris/pull/42460
+    // https://github.com/apache/doris/pull/44635
+    [[nodiscard]] virtual int max_scanners_concurrency(RuntimeState* state) const;
+    [[nodiscard]] virtual int min_scanners_concurrency(RuntimeState* state) const;
+    [[nodiscard]] virtual vectorized::ScannerScheduler* scan_scheduler(RuntimeState* state) const;
 
     [[nodiscard]] std::string get_name() { return _parent->get_name(); }
 
@@ -363,10 +373,6 @@ class ScanOperatorX : public OperatorX<LocalStateType> {
 
     [[nodiscard]] virtual bool is_file_scan_operator() const { return false; }
 
-    [[nodiscard]] virtual int query_parallel_instance_num() const {
-        return _query_parallel_instance_num;
-    }
-
     [[nodiscard]] size_t get_reserve_mem_size(RuntimeState* state) override;
 
     const std::vector<TRuntimeFilterDesc>& runtime_filter_descs() override {
@@ -443,8 +449,6 @@ class ScanOperatorX : public OperatorX<LocalStateType> {
     int64_t _push_down_count = -1;
     const int _parallel_tasks = 0;
 
-    int _query_parallel_instance_num = 0;
-
     std::vector<int> _topn_filter_source_node_ids;
 };
 

be/src/runtime/runtime_state.h

@@ -142,8 +142,15 @@ class RuntimeState {
         return _query_options.__isset.execution_timeout ? _query_options.execution_timeout
                                                         : _query_options.query_timeout;
     }
-    int num_scanner_threads() const {
-        return _query_options.__isset.num_scanner_threads ? _query_options.num_scanner_threads : 0;
+    int max_scanners_concurrency() const {
+        return _query_options.__isset.max_scanners_concurrency
+                       ? _query_options.max_scanners_concurrency
+                       : 0;
+    }
+    int max_file_scanners_concurrency() const {
+        return _query_options.__isset.max_file_scanners_concurrency
+                       ? _query_options.max_file_scanners_concurrency
+                       : max_scanners_concurrency();
     }
     int min_scan_concurrency_of_scan_scheduler() const {
         return _query_options.__isset.min_scan_scheduler_concurrency

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

@@ -56,7 +56,12 @@ ScannerContext::ScannerContext(
         RuntimeState* state, pipeline::ScanLocalStateBase* local_state,
         const TupleDescriptor* output_tuple_desc, const RowDescriptor* output_row_descriptor,
         const std::list<std::shared_ptr<vectorized::ScannerDelegate>>& scanners, int64_t limit_,
-        std::shared_ptr<pipeline::Dependency> dependency, int parallism_of_scan_operator)
+        std::shared_ptr<pipeline::Dependency> dependency
+#ifdef BE_TEST
+        ,
+        int num_parallel_instances
+#endif
+        )
         : HasTaskExecutionCtx(state),
           _state(state),
           _local_state(local_state),
@@ -67,9 +72,20 @@ ScannerContext::ScannerContext(
           _batch_size(state->batch_size()),
           limit(limit_),
           _all_scanners(scanners.begin(), scanners.end()),
-          _parallism_of_scan_operator(parallism_of_scan_operator),
-          _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
+          _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
+          _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);
@@ -106,14 +122,6 @@ 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)) {
         std::shared_ptr<TaskExecutor> task_executor = task_executor_scheduler->task_executor();
@@ -132,43 +140,11 @@ 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();
     }
 
-    // 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.
-    _max_scan_concurrency = _state->num_scanner_threads();
-    if (_max_scan_concurrency == 0) {
-        // Why this is safe:
-        /*
-            1. If num cpu cores is less than or equal to 24:
-                _max_concurrency_of_scan_scheduler will be 96. _parallism_of_scan_operator will be 1 or C/2.
-                so _max_scan_concurrency will be 96 or (96 * 2 / C).
-                For a single scan node, most scanner it can submit will be 96 or (96 * 2 / C) * (C / 2) which is 96 too.
-                So a single scan node could make full utilization of the resource without sumbiting all its tasks.
-            2. If num cpu cores greater than 24:
-                _max_concurrency_of_scan_scheduler will be 4 * C. _parallism_of_scan_operator will be 1 or C/2.
-                so _max_scan_concurrency will be 4 * C or (4 * C * 2 / C).
-                For a single scan node, most scanner it can submit will be 4 * C or (4 * C * 2 / C) * (C / 2) which is 4 * C too.
-
-            So, in all situations, when there is only one scan node, it could make full utilization of the resource.
-        */
-        _max_scan_concurrency =
-                _min_scan_concurrency_of_scan_scheduler / _parallism_of_scan_operator;
-        _max_scan_concurrency = _max_scan_concurrency == 0 ? 1 : _max_scan_concurrency;
-    }
-
-    _max_scan_concurrency = std::min(_max_scan_concurrency, (int32_t)_pending_scanners.size());
-
     // when user not specify scan_thread_num, so we can try downgrade _max_thread_num.
     // becaue we found in a table with 5k columns, column reader may ocuppy too much memory.
     // you can refer https://github.com/apache/doris/issues/35340 for details.
@@ -191,15 +167,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);