This is an automated email from the ASF dual-hosted git repository.
zhangstar333 pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 22bf7548dc5 [cherry-pick](scan)scanner could eos early when reached
limit (#36535) (#36738)
22bf7548dc5 is described below
commit 22bf7548dc5477b6d37a501e468bbbb71382819b
Author: zhangstar333 <[email protected]>
AuthorDate: Wed Jun 26 14:14:52 2024 +0800
[cherry-pick](scan)scanner could eos early when reached limit (#36535)
(#36738)
## Proposed changes
cherry-pick from master #36535
---
be/src/vec/exec/scan/new_es_scan_node.cpp | 4 ++--
be/src/vec/exec/scan/new_file_scan_node.cpp | 7 +++----
be/src/vec/exec/scan/new_jdbc_scan_node.cpp | 5 ++---
be/src/vec/exec/scan/new_odbc_scan_node.cpp | 2 +-
be/src/vec/exec/scan/new_olap_scan_node.cpp | 8 ++++----
be/src/vec/exec/scan/vmeta_scan_node.cpp | 5 ++---
6 files changed, 14 insertions(+), 17 deletions(-)
diff --git a/be/src/vec/exec/scan/new_es_scan_node.cpp
b/be/src/vec/exec/scan/new_es_scan_node.cpp
index 4704f5eb8c6..ab0b565a27a 100644
--- a/be/src/vec/exec/scan/new_es_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_es_scan_node.cpp
@@ -165,8 +165,8 @@ Status
NewEsScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
properties, _column_names, _docvalue_context, &doc_value_mode);
std::shared_ptr<NewEsScanner> scanner = NewEsScanner::create_shared(
- _state, this, _limit_per_scanner, _tuple_id, properties,
_docvalue_context,
- doc_value_mode, _state->runtime_profile());
+ _state, this, _limit, _tuple_id, properties,
_docvalue_context, doc_value_mode,
+ _state->runtime_profile());
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanners->push_back(scanner);
diff --git a/be/src/vec/exec/scan/new_file_scan_node.cpp
b/be/src/vec/exec/scan/new_file_scan_node.cpp
index da33538b8c3..2cc34f65f18 100644
--- a/be/src/vec/exec/scan/new_file_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_file_scan_node.cpp
@@ -121,10 +121,9 @@ Status
NewFileScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
std::min<size_t>(config::doris_scanner_thread_pool_thread_num,
_scan_ranges.size());
_kv_cache.reset(new ShardedKVCache(shard_num));
for (auto& scan_range : _scan_ranges) {
- std::unique_ptr<VFileScanner> scanner =
- VFileScanner::create_unique(_state, this, _limit_per_scanner,
-
scan_range.scan_range.ext_scan_range.file_scan_range,
- runtime_profile(),
_kv_cache.get());
+ std::unique_ptr<VFileScanner> scanner = VFileScanner::create_unique(
+ _state, this, _limit,
scan_range.scan_range.ext_scan_range.file_scan_range,
+ runtime_profile(), _kv_cache.get());
RETURN_IF_ERROR(
scanner->prepare(_conjuncts, &_colname_to_value_range,
&_colname_to_slot_id));
scanners->push_back(std::move(scanner));
diff --git a/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
b/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
index f8219b4337e..deb3f636814 100644
--- a/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_jdbc_scan_node.cpp
@@ -65,9 +65,8 @@ Status
NewJdbcScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
if (_eos == true) {
return Status::OK();
}
- std::unique_ptr<NewJdbcScanner> scanner =
- NewJdbcScanner::create_unique(_state, this, _limit_per_scanner,
_tuple_id,
- _query_string, _table_type,
_state->runtime_profile());
+ std::unique_ptr<NewJdbcScanner> scanner = NewJdbcScanner::create_unique(
+ _state, this, _limit, _tuple_id, _query_string, _table_type,
_state->runtime_profile());
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanners->push_back(std::move(scanner));
return Status::OK();
diff --git a/be/src/vec/exec/scan/new_odbc_scan_node.cpp
b/be/src/vec/exec/scan/new_odbc_scan_node.cpp
index eafad165968..c6361b49bab 100644
--- a/be/src/vec/exec/scan/new_odbc_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_odbc_scan_node.cpp
@@ -66,7 +66,7 @@ Status
NewOdbcScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
return Status::OK();
}
std::shared_ptr<NewOdbcScanner> scanner = NewOdbcScanner::create_shared(
- _state, this, _limit_per_scanner, _odbc_scan_node,
_state->runtime_profile());
+ _state, this, _limit, _odbc_scan_node, _state->runtime_profile());
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanners->push_back(scanner);
return Status::OK();
diff --git a/be/src/vec/exec/scan/new_olap_scan_node.cpp
b/be/src/vec/exec/scan/new_olap_scan_node.cpp
index 3e55823f0ec..61147cc77a8 100644
--- a/be/src/vec/exec/scan/new_olap_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_olap_scan_node.cpp
@@ -513,8 +513,8 @@ Status
NewOlapScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
const std::vector<OlapScanRange*>&
key_ranges,
TabletReader::ReadSource read_source) {
std::shared_ptr<NewOlapScanner> scanner =
NewOlapScanner::create_shared(
- _state, this, _limit_per_scanner,
_olap_scan_node.is_preaggregation, scan_range,
- key_ranges, std::move(read_source),
_scanner_profile.get());
+ _state, this, _limit, _olap_scan_node.is_preaggregation,
scan_range, key_ranges,
+ std::move(read_source), _scanner_profile.get());
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanner->set_compound_filters(_compound_filters);
@@ -605,8 +605,8 @@ Status
NewOlapScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
auto build_new_scanner = [&](const TPaloScanRange& scan_range,
const std::vector<OlapScanRange*>&
key_ranges) {
std::shared_ptr<NewOlapScanner> scanner =
NewOlapScanner::create_shared(
- _state, this, _limit_per_scanner,
_olap_scan_node.is_preaggregation, scan_range,
- key_ranges, _scanner_profile.get());
+ _state, this, _limit, _olap_scan_node.is_preaggregation,
scan_range, key_ranges,
+ _scanner_profile.get());
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanner->set_compound_filters(_compound_filters);
diff --git a/be/src/vec/exec/scan/vmeta_scan_node.cpp
b/be/src/vec/exec/scan/vmeta_scan_node.cpp
index 5ba559466ef..23aecf447af 100644
--- a/be/src/vec/exec/scan/vmeta_scan_node.cpp
+++ b/be/src/vec/exec/scan/vmeta_scan_node.cpp
@@ -68,9 +68,8 @@ Status VMetaScanNode::_init_scanners(std::list<VScannerSPtr>*
scanners) {
}
for (auto& scan_range : _scan_ranges) {
- std::shared_ptr<VMetaScanner> scanner =
- VMetaScanner::create_shared(_state, this, _tuple_id,
scan_range, _limit_per_scanner,
- runtime_profile(), _user_identity);
+ std::shared_ptr<VMetaScanner> scanner = VMetaScanner::create_shared(
+ _state, this, _tuple_id, scan_range, _limit,
runtime_profile(), _user_identity);
RETURN_IF_ERROR(scanner->prepare(_state, _conjuncts));
scanners->push_back(scanner);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
