[cherry-pick](scan)scanner could eos early when reached limit (#36535)

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);

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));

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();

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();

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

@@ -505,8 +505,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);
@@ -597,8 +597,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);

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);
     }