[Improvement](scanner) Include open() time in PerScannerRunningTime (#61042)

Use NVI (Non-Virtual Interface) pattern for Scanner::open(), similar to
Scanner::get_block() / _get_block_impl(). Add SCOPED_RAW_TIMER in the
base class open() method to include scanner initialization time
(ReaderInitTime, BlockReaderBuildHeapInitTimer, etc.) in
PerScannerRunningTime profile counter.

This fixes the issue where PerScannerRunningTime showed only
microseconds while ReaderInitTime showed several seconds, making the
profile misleading.

Changes:
- scanner.h: Change open() to non-virtual NVI, add _open_impl() virtual
- 5 subclasses: Rename open() to _open_impl() and call
base::_open_impl() (OlapScanner, FileScanner, MetaScanner, JdbcScanner,
EsScanner)

Author: HappenLee

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

@@ -78,7 +78,7 @@ Status EsScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts)
     return Status::OK();
 }
 
-Status EsScanner::open(RuntimeState* state) {
+Status EsScanner::_open_impl(RuntimeState* state) {
     VLOG_CRITICAL << NEW_SCANNER_TYPE << "::open";
 
     if (nullptr == state) {
@@ -90,7 +90,7 @@ Status EsScanner::open(RuntimeState* state) {
     }
 
     RETURN_IF_CANCELLED(state);
-    RETURN_IF_ERROR(Scanner::open(state));
+    RETURN_IF_ERROR(Scanner::_open_impl(state));
 
     RETURN_IF_ERROR(_es_reader->open());
 

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

@@ -54,7 +54,7 @@ class EsScanner : public Scanner {
               const std::map<std::string, std::string>& docvalue_context, bool doc_value_mode,
               RuntimeProfile* profile);
 
-    Status open(RuntimeState* state) override;
+    Status _open_impl(RuntimeState* state) override;
     Status close(RuntimeState* state) override;
     Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;
 

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

@@ -379,9 +379,9 @@ void FileScanner::_get_slot_ids(VExpr* expr, std::vector<int>* slot_ids) {
     }
 }
 
-Status FileScanner::open(RuntimeState* state) {
+Status FileScanner::_open_impl(RuntimeState* state) {
     RETURN_IF_CANCELLED(state);
-    RETURN_IF_ERROR(Scanner::open(state));
+    RETURN_IF_ERROR(Scanner::_open_impl(state));
     RETURN_IF_ERROR(_split_source->get_next(&_first_scan_range, &_current_range));
     if (_first_scan_range) {
         RETURN_IF_ERROR(_init_expr_ctxes());

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

@@ -70,7 +70,7 @@ class FileScanner : public Scanner {
                 RuntimeProfile* profile, ShardedKVCache* kv_cache,
                 const std::unordered_map<std::string, int>* colname_to_slot_id);
 
-    Status open(RuntimeState* state) override;
+    Status _open_impl(RuntimeState* state) override;
 
     Status close(RuntimeState* state) override;
 

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

@@ -101,7 +101,7 @@ Status JdbcScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts
     return Status::OK();
 }
 
-Status JdbcScanner::open(RuntimeState* state) {
+Status JdbcScanner::_open_impl(RuntimeState* state) {
     VLOG_CRITICAL << "JdbcScanner::open";
     if (state == nullptr) {
         return Status::InternalError("input pointer is NULL of VJdbcScanNode::open.");
@@ -111,7 +111,7 @@ Status JdbcScanner::open(RuntimeState* state) {
         return Status::InternalError("used before initialize of VJdbcScanNode::open.");
     }
     RETURN_IF_CANCELLED(state);
-    RETURN_IF_ERROR(Scanner::open(state));
+    RETURN_IF_ERROR(Scanner::_open_impl(state));
     RETURN_IF_ERROR(_jdbc_connector->open(state, true));
     RETURN_IF_ERROR(_jdbc_connector->query());
     return Status::OK();

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

@@ -48,7 +48,7 @@ class JdbcScanner : public Scanner {
     JdbcScanner(RuntimeState* state, doris::pipeline::JDBCScanLocalState* parent, int64_t limit,
                 const TupleId& tuple_id, const std::string& query_string,
                 TOdbcTableType::type table_type, bool is_tvf, RuntimeProfile* profile);
-    Status open(RuntimeState* state) override;
+    Status _open_impl(RuntimeState* state) override;
     Status close(RuntimeState* state) override;
 
     Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;

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

@@ -64,9 +64,9 @@ MetaScanner::MetaScanner(RuntimeState* state, pipeline::ScanLocalStateBase* loca
           _user_identity(user_identity),
           _scan_range(scan_range.scan_range) {}
 
-Status MetaScanner::open(RuntimeState* state) {
+Status MetaScanner::_open_impl(RuntimeState* state) {
     VLOG_CRITICAL << "MetaScanner::open";
-    RETURN_IF_ERROR(Scanner::open(state));
+    RETURN_IF_ERROR(Scanner::_open_impl(state));
     if (_scan_range.meta_scan_range.metadata_type == TMetadataType::ICEBERG) {
         // TODO: refactor this code
         auto reader = IcebergSysTableJniReader::create_unique(_tuple_desc->slots(), state, _profile,

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

@@ -54,7 +54,7 @@ class MetaScanner : public Scanner {
                 const TScanRangeParams& scan_range, int64_t limit, RuntimeProfile* profile,
                 TUserIdentity user_identity);
 
-    Status open(RuntimeState* state) override;
+    Status _open_impl(RuntimeState* state) override;
     Status close(RuntimeState* state) override;
     Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;
 

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

@@ -306,8 +306,8 @@ Status OlapScanner::prepare() {
     return Status::OK();
 }
 
-Status OlapScanner::open(RuntimeState* state) {
-    RETURN_IF_ERROR(Scanner::open(state));
+Status OlapScanner::_open_impl(RuntimeState* state) {
+    RETURN_IF_ERROR(Scanner::_open_impl(state));
     SCOPED_TIMER(_local_state->cast<pipeline::OlapScanLocalState>()._reader_init_timer);
 
     auto res = _tablet_reader->init(_tablet_reader_params);

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

@@ -75,7 +75,7 @@ class OlapScanner : public Scanner {
 
     Status prepare() override;
 
-    Status open(RuntimeState* state) override;
+    Status _open_impl(RuntimeState* state) override;
 
     Status close(RuntimeState* state) override;