diff --git a/be/src/olap/parallel_scanner_builder.cpp b/be/src/olap/parallel_scanner_builder.cpp index ac61d874cc3e04..ac57448ade7e6d 100644 --- a/be/src/olap/parallel_scanner_builder.cpp +++ b/be/src/olap/parallel_scanner_builder.cpp @@ -199,9 +199,8 @@ Status ParallelScannerBuilder::_load() { std::shared_ptr ParallelScannerBuilder::_build_scanner( BaseTabletSPtr tablet, int64_t version, const std::vector& key_ranges, TabletReader::ReadSource&& read_source) { - NewOlapScanner::Params params { - _state, _scanner_profile.get(), key_ranges, std::move(tablet), - version, std::move(read_source), _limit_per_scanner, _is_preaggregation}; + NewOlapScanner::Params params {_state, _scanner_profile.get(), key_ranges, std::move(tablet), + version, std::move(read_source), _limit, _is_preaggregation}; return NewOlapScanner::create_shared(_parent, std::move(params)); } diff --git a/be/src/olap/parallel_scanner_builder.h b/be/src/olap/parallel_scanner_builder.h index e1aeae8e58ad76..eb25e183df2777 100644 --- a/be/src/olap/parallel_scanner_builder.h +++ b/be/src/olap/parallel_scanner_builder.h @@ -44,11 +44,11 @@ class ParallelScannerBuilder { const std::vector& tablets, const std::shared_ptr& profile, const std::vector& key_ranges, RuntimeState* state, - int64_t limit_per_scanner, bool is_dup_mow_key, bool is_preaggregation) + int64_t limit, bool is_dup_mow_key, bool is_preaggregation) : _parent(parent), _scanner_profile(profile), _state(state), - _limit_per_scanner(limit_per_scanner), + _limit(limit), _is_dup_mow_key(is_dup_mow_key), _is_preaggregation(is_preaggregation), _tablets(tablets.cbegin(), tablets.cend()), @@ -85,7 +85,7 @@ class ParallelScannerBuilder { std::shared_ptr _scanner_profile; RuntimeState* _state; - int64_t _limit_per_scanner; + int64_t _limit; bool _is_dup_mow_key; bool _is_preaggregation; std::vector _tablets; diff --git a/be/src/pipeline/exec/es_scan_operator.cpp b/be/src/pipeline/exec/es_scan_operator.cpp index 9fcfdd999d403f..7dfa73a8cdfee2 100644 --- a/be/src/pipeline/exec/es_scan_operator.cpp +++ b/be/src/pipeline/exec/es_scan_operator.cpp @@ -94,7 +94,7 @@ Status EsScanLocalState::_init_scanners(std::list* sca properties, p._column_names, p._docvalue_context, &doc_value_mode); std::shared_ptr scanner = vectorized::NewEsScanner::create_shared( - RuntimeFilterConsumer::_state, this, p._limit_per_scanner, p._tuple_id, properties, + RuntimeFilterConsumer::_state, this, p._limit, p._tuple_id, properties, p._docvalue_context, doc_value_mode, RuntimeFilterConsumer::_state->runtime_profile()); diff --git a/be/src/pipeline/exec/file_scan_operator.cpp b/be/src/pipeline/exec/file_scan_operator.cpp index d3f424542070c5..9e636d0331f4b3 100644 --- a/be/src/pipeline/exec/file_scan_operator.cpp +++ b/be/src/pipeline/exec/file_scan_operator.cpp @@ -44,8 +44,7 @@ Status FileScanLocalState::_init_scanners(std::list* s _kv_cache.reset(new vectorized::ShardedKVCache(shard_num)); for (int i = 0; i < _max_scanners; ++i) { std::unique_ptr scanner = vectorized::VFileScanner::create_unique( - state(), this, p._limit_per_scanner, _split_source, _scanner_profile.get(), - _kv_cache.get()); + state(), this, p._limit, _split_source, _scanner_profile.get(), _kv_cache.get()); RETURN_IF_ERROR( scanner->prepare(_conjuncts, &_colname_to_value_range, &p._colname_to_slot_id)); scanners->push_back(std::move(scanner)); diff --git a/be/src/pipeline/exec/jdbc_scan_operator.cpp b/be/src/pipeline/exec/jdbc_scan_operator.cpp index ec81b780c707a7..54f76396343944 100644 --- a/be/src/pipeline/exec/jdbc_scan_operator.cpp +++ b/be/src/pipeline/exec/jdbc_scan_operator.cpp @@ -31,7 +31,7 @@ std::string JDBCScanLocalState::name_suffix() const { Status JDBCScanLocalState::_init_scanners(std::list* scanners) { auto& p = _parent->cast(); std::unique_ptr scanner = vectorized::NewJdbcScanner::create_unique( - state(), this, p._limit_per_scanner, p._tuple_id, p._query_string, p._table_type, + state(), this, p._limit, p._tuple_id, p._query_string, p._table_type, _scanner_profile.get()); RETURN_IF_ERROR(scanner->prepare(state(), _conjuncts)); scanners->push_back(std::move(scanner)); diff --git a/be/src/pipeline/exec/meta_scan_operator.cpp b/be/src/pipeline/exec/meta_scan_operator.cpp index e981c6a2dc1b0d..0ff8a7337bb2aa 100644 --- a/be/src/pipeline/exec/meta_scan_operator.cpp +++ b/be/src/pipeline/exec/meta_scan_operator.cpp @@ -30,8 +30,7 @@ Status MetaScanLocalState::_init_scanners(std::list* s for (auto& scan_range : _scan_ranges) { std::shared_ptr scanner = vectorized::VMetaScanner::create_shared( - state(), this, p._tuple_id, scan_range, p._limit_per_scanner, profile(), - p._user_identity); + state(), this, p._tuple_id, scan_range, p._limit, profile(), p._user_identity); RETURN_IF_ERROR(scanner->prepare(state(), _conjuncts)); scanners->push_back(scanner); } diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp index 346540e03eccf3..35587f9fbce127 100644 --- a/be/src/pipeline/exec/olap_scan_operator.cpp +++ b/be/src/pipeline/exec/olap_scan_operator.cpp @@ -306,8 +306,7 @@ Status OlapScanLocalState::_init_scanners(std::list* s } ParallelScannerBuilder scanner_builder(this, tablets, _scanner_profile, key_ranges, state(), - p._limit_per_scanner, true, - p._olap_scan_node.is_preaggregation); + p._limit, true, p._olap_scan_node.is_preaggregation); int max_scanners_count = state()->parallel_scan_max_scanners_count(); @@ -345,7 +344,7 @@ Status OlapScanLocalState::_init_scanners(std::list* s std::move(tablet), version, {}, - p._limit_per_scanner, + p._limit, p._olap_scan_node.is_preaggregation, }); RETURN_IF_ERROR(scanner->prepare(state(), _conjuncts)); diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp index e15ec015908eff..1de8e7a79509fa 100644 --- a/be/src/vec/exec/scan/scanner_context.cpp +++ b/be/src/vec/exec/scan/scanner_context.cpp @@ -80,7 +80,11 @@ ScannerContext::ScannerContext( : state->query_parallel_instance_num()); _max_thread_num = _max_thread_num == 0 ? 1 : _max_thread_num; _max_thread_num = std::min(_max_thread_num, (int32_t)scanners.size()); - + // 1. Calculate max concurrency + // For select * from table limit 10; should just use one thread. + if (_local_state && _local_state->should_run_serial()) { + _max_thread_num = 1; + } // 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.