Coverage Report

Created: 2026-06-05 04:37

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exec/scan/olap_scanner.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "exec/scan/olap_scanner.h"
19
20
#include <gen_cpp/Descriptors_types.h>
21
#include <gen_cpp/PlanNodes_types.h>
22
#include <gen_cpp/Types_types.h>
23
#include <glog/logging.h>
24
#include <stdlib.h>
25
#include <thrift/protocol/TDebugProtocol.h>
26
27
#include <algorithm>
28
#include <atomic>
29
#include <iterator>
30
#include <ostream>
31
#include <set>
32
33
#include "cloud/cloud_storage_engine.h"
34
#include "cloud/cloud_tablet_hotspot.h"
35
#include "cloud/config.h"
36
#include "common/config.h"
37
#include "common/consts.h"
38
#include "common/logging.h"
39
#include "common/metrics/doris_metrics.h"
40
#include "core/block/block.h"
41
#include "exec/common/variant_util.h"
42
#include "exec/operator/olap_scan_operator.h"
43
#include "exec/scan/scan_node.h"
44
#include "exprs/function_filter.h"
45
#include "exprs/vexpr.h"
46
#include "exprs/vexpr_context.h"
47
#include "io/cache/block_file_cache_profile.h"
48
#include "io/io_common.h"
49
#include "runtime/descriptors.h"
50
#include "runtime/exec_env.h"
51
#include "runtime/runtime_profile.h"
52
#include "runtime/runtime_state.h"
53
#include "service/backend_options.h"
54
#include "storage/id_manager.h"
55
#include "storage/index/inverted/inverted_index_profile.h"
56
#include "storage/iterator/block_reader.h"
57
#include "storage/olap_common.h"
58
#include "storage/olap_tuple.h"
59
#include "storage/olap_utils.h"
60
#include "storage/storage_engine.h"
61
#include "storage/tablet/tablet_schema.h"
62
#ifndef NDEBUG
63
#include "util/debug_points.h"
64
#endif
65
#include "util/json/path_in_data.h"
66
67
namespace doris {
68
#include "common/compile_check_avoid_begin.h"
69
70
using ReadSource = TabletReadSource;
71
72
OlapScanner::OlapScanner(ScanLocalStateBase* parent, OlapScanner::Params&& params)
73
875k
        : Scanner(params.state, parent, params.limit, params.profile),
74
875k
          _key_ranges(std::move(params.key_ranges)),
75
875k
          _tablet_reader_params({.tablet = std::move(params.tablet),
76
875k
                                 .tablet_schema {},
77
875k
                                 .reader_type = params.read_row_binlog ? ReaderType::READER_BINLOG
78
875k
                                                                       : ReaderType::READER_QUERY,
79
875k
                                 .aggregation = params.aggregation,
80
875k
                                 .version = {0, params.version},
81
875k
                                 .start_key {},
82
875k
                                 .end_key {},
83
875k
                                 .predicates {},
84
875k
                                 .function_filters {},
85
875k
                                 .delete_predicates {},
86
875k
                                 .target_cast_type_for_variants {},
87
875k
                                 .all_access_paths {},
88
875k
                                 .predicate_access_paths {},
89
875k
                                 .rs_splits {},
90
875k
                                 .return_columns {},
91
875k
                                 .output_columns {},
92
875k
                                 .common_expr_ctxs_push_down {},
93
875k
                                 .topn_filter_source_node_ids {},
94
875k
                                 .key_group_cluster_key_idxes {},
95
875k
                                 .virtual_column_exprs {},
96
875k
                                 .vir_cid_to_idx_in_block {},
97
875k
                                 .vir_col_idx_to_type {},
98
875k
                                 .score_runtime {},
99
875k
                                 .collection_statistics {},
100
875k
                                 .ann_topn_runtime {},
101
875k
                                 .condition_cache_digest = parent->get_condition_cache_digest()}) {
102
875k
    _tablet_reader_params.set_read_source(std::move(params.read_source),
103
875k
                                          _state->skip_delete_bitmap());
104
875k
    _has_prepared = false;
105
875k
    _vector_search_params = params.state->get_vector_search_params();
106
875k
}
107
108
static std::string read_columns_to_string(TabletSchemaSPtr tablet_schema,
109
3.25k
                                          const std::vector<uint32_t>& read_columns) {
110
    // avoid too long for one line,
111
    // it is hard to display in `show profile` stmt if one line is too long.
112
3.25k
    const int col_per_line = 10;
113
3.25k
    int i = 0;
114
3.25k
    std::string read_columns_string;
115
3.25k
    read_columns_string += "[";
116
17.3k
    for (auto it = read_columns.cbegin(); it != read_columns.cend(); it++) {
117
14.1k
        if (it != read_columns.cbegin()) {
118
10.9k
            read_columns_string += ", ";
119
10.9k
        }
120
14.1k
        read_columns_string += tablet_schema->columns().at(*it)->name();
121
14.1k
        if (i >= col_per_line) {
122
13
            read_columns_string += "\n";
123
13
            i = 0;
124
14.0k
        } else {
125
14.0k
            ++i;
126
14.0k
        }
127
14.1k
    }
128
3.25k
    read_columns_string += "]";
129
3.25k
    return read_columns_string;
130
3.25k
}
131
132
1.79M
static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) {
133
1.79M
    return stats.num_local_io_total != 0 || stats.num_remote_io_total != 0 ||
134
1.79M
           stats.num_peer_io_total != 0 || stats.local_io_timer != 0 ||
135
1.79M
           stats.bytes_read_from_local != 0 || stats.bytes_read_from_remote != 0 ||
136
1.79M
           stats.bytes_read_from_peer != 0 || stats.remote_io_timer != 0 ||
137
1.79M
           stats.peer_io_timer != 0 || stats.remote_wait_timer != 0 ||
138
1.79M
           stats.write_cache_io_timer != 0 || stats.bytes_write_into_cache != 0 ||
139
1.79M
           stats.num_skip_cache_io_total != 0 || stats.read_cache_file_directly_timer != 0 ||
140
1.79M
           stats.cache_get_or_set_timer != 0 || stats.lock_wait_timer != 0 ||
141
1.79M
           stats.get_timer != 0 || stats.set_timer != 0 ||
142
1.79M
           stats.inverted_index_num_local_io_total != 0 ||
143
1.79M
           stats.inverted_index_num_remote_io_total != 0 ||
144
1.79M
           stats.inverted_index_num_peer_io_total != 0 ||
145
1.79M
           stats.inverted_index_bytes_read_from_local != 0 ||
146
1.79M
           stats.inverted_index_bytes_read_from_remote != 0 ||
147
1.79M
           stats.inverted_index_bytes_read_from_peer != 0 ||
148
1.79M
           stats.inverted_index_local_io_timer != 0 || stats.inverted_index_remote_io_timer != 0 ||
149
1.79M
           stats.inverted_index_peer_io_timer != 0 || stats.inverted_index_io_timer != 0;
150
1.79M
}
151
152
875k
Status OlapScanner::_prepare_impl() {
153
875k
    auto* local_state = static_cast<OlapScanLocalState*>(_local_state);
154
875k
    auto& tablet = _tablet_reader_params.tablet;
155
875k
    auto& tablet_schema = _tablet_reader_params.tablet_schema;
156
875k
    DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", {
157
875k
        LOG_WARNING("CloudTablet.capture_rs_readers.return e-230 init")
158
875k
                .tag("tablet_id", tablet->tablet_id());
159
875k
        return Status::Error<false>(-230, "injected error");
160
875k
    });
161
162
875k
    for (auto& ctx : local_state->_common_expr_ctxs_push_down) {
163
22.8k
        VExprContextSPtr context;
164
22.8k
        RETURN_IF_ERROR(ctx->clone(_state, context));
165
22.8k
        _common_expr_ctxs_push_down.emplace_back(context);
166
22.8k
        context->prepare_ann_range_search(_vector_search_params);
167
22.8k
    }
168
169
875k
    for (auto pair : local_state->_slot_id_to_virtual_column_expr) {
170
        // Scanner will be executed in a different thread, so we need to clone the context.
171
482
        VExprContextSPtr context;
172
482
        RETURN_IF_ERROR(pair.second->clone(_state, context));
173
482
        _slot_id_to_virtual_column_expr[pair.first] = context;
174
482
    }
175
176
875k
    _slot_id_to_index_in_block = local_state->_slot_id_to_index_in_block;
177
875k
    _slot_id_to_col_type = local_state->_slot_id_to_col_type;
178
875k
    _score_runtime = local_state->_score_runtime;
179
    // All scanners share the same ann_topn_runtime.
180
875k
    _ann_topn_runtime = local_state->_ann_topn_runtime;
181
182
    // set limit to reduce end of rowset and segment mem use
183
875k
    _tablet_reader = std::make_unique<BlockReader>();
184
    // batch size is passed down to segment iterator, use _state->batch_size()
185
    // instead of _parent->limit(), because if _parent->limit() is a very small
186
    // value (e.g. select a from t where a .. and b ... limit 1),
187
    // it will be very slow when reading data in segment iterator
188
875k
    _tablet_reader->set_batch_size(_state->batch_size());
189
    // Adaptive batch size: pass byte-budget settings to the storage reader.
190
    // The reader still uses batch_size() as the row ceiling.
191
875k
    _tablet_reader->set_preferred_block_size_bytes(_state->preferred_block_size_bytes());
192
875k
    {
193
875k
        TOlapScanNode& olap_scan_node = local_state->olap_scan_node();
194
875k
        TabletSchemaSPtr source_tablet_schema =
195
875k
                _tablet_reader_params.reader_type == ReaderType::READER_BINLOG
196
875k
                        ? tablet->row_binlog_tablet_schema()
197
875k
                        : tablet->tablet_schema();
198
199
875k
        tablet_schema = std::make_shared<TabletSchema>();
200
875k
        tablet_schema->copy_from(*source_tablet_schema);
201
875k
        if (olap_scan_node.__isset.columns_desc && !olap_scan_node.columns_desc.empty() &&
202
875k
            olap_scan_node.columns_desc[0].col_unique_id >= 0) {
203
874k
            tablet_schema->clear_columns();
204
13.7M
            for (const auto& column_desc : olap_scan_node.columns_desc) {
205
13.7M
                tablet_schema->append_column(TabletColumn(column_desc));
206
13.7M
            }
207
875k
            if (olap_scan_node.__isset.schema_version) {
208
875k
                tablet_schema->set_schema_version(olap_scan_node.schema_version);
209
875k
            }
210
874k
        }
211
875k
        if (olap_scan_node.__isset.indexes_desc) {
212
874k
            tablet_schema->update_indexes_from_thrift(olap_scan_node.indexes_desc);
213
874k
        }
214
215
875k
        if (_tablet_reader_params.rs_splits.empty()) {
216
            // Non-pipeline mode, Tablet : Scanner = 1 : 1
217
            // acquire tablet rowset readers at the beginning of the scan node
218
            // to prevent this case: when there are lots of olap scanners to run for example 10000
219
            // the rowsets maybe compacted when the last olap scanner starts
220
0
            ReadSource read_source;
221
222
0
            if (config::is_cloud_mode()) {
223
                // FIXME(plat1ko): Avoid pointer cast
224
0
                ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(*tablet);
225
0
            }
226
227
0
            auto maybe_read_source = tablet->capture_read_source(
228
0
                    _tablet_reader_params.version,
229
0
                    {
230
0
                            .skip_missing_versions = _state->skip_missing_version(),
231
0
                            .enable_fetch_rowsets_from_peers =
232
0
                                    config::enable_fetch_rowsets_from_peer_replicas,
233
0
                            .capture_row_binlog =
234
0
                                    _tablet_reader_params.reader_type == ReaderType::READER_BINLOG,
235
0
                            .enable_prefer_cached_rowset =
236
0
                                    config::is_cloud_mode() ? _state->enable_prefer_cached_rowset()
237
0
                                                            : false,
238
0
                            .query_freshness_tolerance_ms =
239
0
                                    config::is_cloud_mode() ? _state->query_freshness_tolerance_ms()
240
0
                                                            : -1,
241
0
                    });
242
0
            if (!maybe_read_source) {
243
0
                LOG(WARNING) << "fail to init reader. res=" << maybe_read_source.error();
244
0
                return maybe_read_source.error();
245
0
            }
246
0
            read_source = std::move(maybe_read_source.value());
247
248
0
            if (config::enable_mow_verbose_log && tablet->enable_unique_key_merge_on_write()) {
249
0
                LOG_INFO("finish capture_rs_readers for tablet={}, query_id={}",
250
0
                         tablet->tablet_id(), print_id(_state->query_id()));
251
0
            }
252
253
0
            if (!_state->skip_delete_predicate()) {
254
0
                read_source.fill_delete_predicates();
255
0
            }
256
0
            _tablet_reader_params.set_read_source(std::move(read_source));
257
0
        }
258
259
        // Initialize tablet_reader_params
260
875k
        RETURN_IF_ERROR(_init_tablet_reader_params(
261
875k
                local_state->_parent->cast<OlapScanOperatorX>()._slot_id_to_slot_desc, _key_ranges,
262
875k
                local_state->_slot_id_to_predicates, local_state->_push_down_functions));
263
875k
    }
264
265
    // add read columns in profile
266
875k
    if (_state->enable_profile()) {
267
3.23k
        _profile->add_info_string("ReadColumns",
268
3.23k
                                  read_columns_to_string(tablet_schema, _return_columns));
269
3.23k
    }
270
271
875k
    if (_tablet_reader_params.score_runtime) {
272
14
        SCOPED_TIMER(local_state->_statistics_collect_timer);
273
14
        _tablet_reader_params.collection_statistics = std::make_shared<CollectionStatistics>();
274
275
14
        io::IOContext io_ctx {
276
14
                .reader_type = _tablet_reader_params.reader_type,
277
14
                .expiration_time = tablet->ttl_seconds(),
278
14
                .query_id = &_state->query_id(),
279
14
                .file_cache_stats = &_tablet_reader->mutable_stats()->file_cache_stats,
280
14
                .is_inverted_index = true,
281
14
        };
282
283
14
        RETURN_IF_ERROR(_tablet_reader_params.collection_statistics->collect(
284
14
                _state, _tablet_reader_params.rs_splits, _tablet_reader_params.tablet_schema,
285
14
                _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx));
286
14
    }
287
288
875k
    _has_prepared = true;
289
875k
    return Status::OK();
290
875k
}
291
292
873k
Status OlapScanner::_open_impl(RuntimeState* state) {
293
873k
    RETURN_IF_ERROR(Scanner::_open_impl(state));
294
873k
    SCOPED_TIMER(_local_state->cast<OlapScanLocalState>()._reader_init_timer);
295
296
873k
    auto res = _tablet_reader->init(_tablet_reader_params);
297
873k
    if (!res.ok()) {
298
50
        res.append("failed to initialize storage reader. tablet=" +
299
50
                   std::to_string(_tablet_reader_params.tablet->tablet_id()) +
300
50
                   ", backend=" + BackendOptions::get_localhost());
301
50
        return res;
302
50
    }
303
304
    // Do not hold rs_splits any more to release memory.
305
873k
    _tablet_reader_params.rs_splits.clear();
306
307
873k
    return Status::OK();
308
873k
}
309
310
// it will be called under tablet read lock because capture rs readers need
311
Status OlapScanner::_init_tablet_reader_params(
312
        const phmap::flat_hash_map<int, SlotDescriptor*>& slot_id_to_slot_desc,
313
        const std::vector<OlapScanRange*>& key_ranges,
314
        const phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
315
                slot_to_predicates,
316
874k
        const std::vector<FunctionFilter>& function_filters) {
317
    // if the table with rowset [0-x] or [0-1] [2-y], and [0-1] is empty
318
874k
    const bool single_version = _tablet_reader_params.has_single_version();
319
320
874k
    auto* olap_local_state = static_cast<OlapScanLocalState*>(_local_state);
321
874k
    bool read_mor_as_dup = olap_local_state->olap_scan_node().__isset.read_mor_as_dup &&
322
874k
                           olap_local_state->olap_scan_node().read_mor_as_dup;
323
874k
    if (_state->skip_storage_engine_merge() || read_mor_as_dup) {
324
49
        _tablet_reader_params.direct_mode = true;
325
49
        _tablet_reader_params.aggregation = true;
326
874k
    } else {
327
874k
        auto push_down_agg_type = _local_state->get_push_down_agg_type();
328
874k
        _tablet_reader_params.direct_mode = _tablet_reader_params.aggregation || single_version ||
329
874k
                                            (push_down_agg_type != TPushAggOp::NONE &&
330
10.1k
                                             push_down_agg_type != TPushAggOp::COUNT_ON_INDEX);
331
874k
    }
332
333
874k
    RETURN_IF_ERROR(_init_variant_columns());
334
874k
    RETURN_IF_ERROR(_init_return_columns());
335
336
874k
    _tablet_reader_params.push_down_agg_type_opt = _local_state->get_push_down_agg_type();
337
338
874k
    _tablet_reader_params.common_expr_ctxs_push_down = _common_expr_ctxs_push_down;
339
874k
    _tablet_reader_params.virtual_column_exprs = _virtual_column_exprs;
340
874k
    _tablet_reader_params.vir_cid_to_idx_in_block = _vir_cid_to_idx_in_block;
341
874k
    _tablet_reader_params.vir_col_idx_to_type = _vir_col_idx_to_type;
342
874k
    _tablet_reader_params.score_runtime = _score_runtime;
343
874k
    _tablet_reader_params.output_columns = ((OlapScanLocalState*)_local_state)->_output_column_ids;
344
874k
    _tablet_reader_params.ann_topn_runtime = _ann_topn_runtime;
345
874k
    for (const auto& ele : ((OlapScanLocalState*)_local_state)->_cast_types_for_variants) {
346
1.72k
        _tablet_reader_params.target_cast_type_for_variants[ele.first] = ele.second;
347
1.72k
    };
348
874k
    auto& tablet_schema = _tablet_reader_params.tablet_schema;
349
6.03M
    for (auto& predicates : slot_to_predicates) {
350
6.03M
        const int sid = predicates.first;
351
6.03M
        DCHECK(slot_id_to_slot_desc.contains(sid));
352
6.03M
        int32_t index =
353
6.03M
                tablet_schema->field_index(slot_id_to_slot_desc.find(sid)->second->col_name());
354
6.03M
        if (index < 0) {
355
0
            throw Exception(
356
0
                    Status::InternalError("Column {} not found in tablet schema",
357
0
                                          slot_id_to_slot_desc.find(sid)->second->col_name()));
358
0
        }
359
6.03M
        for (auto& predicate : predicates.second) {
360
642k
            _tablet_reader_params.predicates.push_back(predicate->clone(index));
361
642k
        }
362
6.03M
    }
363
364
874k
    std::copy(function_filters.cbegin(), function_filters.cend(),
365
874k
              std::inserter(_tablet_reader_params.function_filters,
366
874k
                            _tablet_reader_params.function_filters.begin()));
367
368
    // Merge the columns in delete predicate that not in latest schema in to current tablet schema
369
874k
    for (auto& del_pred : _tablet_reader_params.delete_predicates) {
370
7.10k
        tablet_schema->merge_dropped_columns(*del_pred->tablet_schema());
371
7.10k
    }
372
373
    // Push key ranges to the tablet reader.
374
    // Skip the "full scan" placeholder (has_lower_bound == false) — when no key
375
    // predicates exist, start_key/end_key remain empty and the reader does a full scan.
376
1.39M
    for (auto* key_range : key_ranges) {
377
1.39M
        if (!key_range->has_lower_bound) {
378
133k
            continue;
379
133k
        }
380
381
1.26M
        _tablet_reader_params.start_key_include = key_range->begin_include;
382
1.26M
        _tablet_reader_params.end_key_include = key_range->end_include;
383
384
1.26M
        _tablet_reader_params.start_key.push_back(key_range->begin_scan_range);
385
1.26M
        _tablet_reader_params.end_key.push_back(key_range->end_scan_range);
386
1.26M
    }
387
388
874k
    _tablet_reader_params.profile = _local_state->custom_profile();
389
874k
    _tablet_reader_params.runtime_state = _state;
390
391
874k
    _tablet_reader_params.origin_return_columns = &_return_columns;
392
874k
    _tablet_reader_params.tablet_columns_convert_to_null_set = &_tablet_columns_convert_to_null_set;
393
394
874k
    if (_tablet_reader_params.direct_mode) {
395
862k
        _tablet_reader_params.return_columns = _return_columns;
396
862k
    } else {
397
        // we need to fetch all key columns to do the right aggregation on storage engine side.
398
38.3k
        for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) {
399
26.3k
            _tablet_reader_params.return_columns.push_back(i);
400
26.3k
        }
401
48.0k
        for (auto index : _return_columns) {
402
48.0k
            if (tablet_schema->column(index).is_key()) {
403
17.2k
                continue;
404
17.2k
            }
405
30.7k
            _tablet_reader_params.return_columns.push_back(index);
406
30.7k
        }
407
        // expand the sequence column
408
12.0k
        if (tablet_schema->has_sequence_col() || tablet_schema->has_seq_map()) {
409
40
            bool has_replace_col = false;
410
90
            for (auto col : _return_columns) {
411
90
                if (tablet_schema->column(col).aggregation() ==
412
90
                    FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE) {
413
39
                    has_replace_col = true;
414
39
                    break;
415
39
                }
416
90
            }
417
40
            if (auto sequence_col_idx = tablet_schema->sequence_col_idx();
418
40
                has_replace_col && tablet_schema->has_sequence_col() &&
419
40
                std::find(_return_columns.begin(), _return_columns.end(), sequence_col_idx) ==
420
27
                        _return_columns.end()) {
421
16
                _tablet_reader_params.return_columns.push_back(sequence_col_idx);
422
16
            }
423
40
            if (has_replace_col) {
424
39
                const auto& val_to_seq = tablet_schema->value_col_idx_to_seq_col_idx();
425
39
                std::set<uint32_t> return_seq_columns;
426
427
240
                for (auto col : _tablet_reader_params.return_columns) {
428
                    // we need to add the necessary sequence column in _return_columns, and
429
                    // Avoid adding the same seq column twice
430
240
                    const auto val_iter = val_to_seq.find(col);
431
240
                    if (val_iter != val_to_seq.end()) {
432
42
                        auto seq = val_iter->second;
433
42
                        if (std::find(_tablet_reader_params.return_columns.begin(),
434
42
                                      _tablet_reader_params.return_columns.end(),
435
42
                                      seq) == _tablet_reader_params.return_columns.end()) {
436
4
                            return_seq_columns.insert(seq);
437
4
                        }
438
42
                    }
439
240
                }
440
39
                _tablet_reader_params.return_columns.insert(
441
39
                        std::end(_tablet_reader_params.return_columns),
442
39
                        std::begin(return_seq_columns), std::end(return_seq_columns));
443
39
            }
444
40
        }
445
12.0k
    }
446
447
874k
    _tablet_reader_params.use_page_cache = _state->enable_page_cache();
448
449
874k
    DBUG_EXECUTE_IF("NewOlapScanner::_init_tablet_reader_params.block", DBUG_BLOCK);
450
451
874k
    if (!_state->skip_storage_engine_merge()) {
452
872k
        auto* olap_scan_local_state = (OlapScanLocalState*)_local_state;
453
872k
        TOlapScanNode& olap_scan_node = olap_scan_local_state->olap_scan_node();
454
455
        // Set MOR value predicate pushdown flag
456
872k
        if (olap_scan_node.__isset.enable_mor_value_predicate_pushdown &&
457
872k
            olap_scan_node.enable_mor_value_predicate_pushdown) {
458
24
            _tablet_reader_params.enable_mor_value_predicate_pushdown = true;
459
24
        }
460
461
872k
        const bool has_key_topn =
462
872k
                olap_scan_node.__isset.sort_info && !olap_scan_node.sort_info.is_asc_order.empty();
463
872k
        if (has_key_topn) {
464
2.24k
            _limit = _local_state->limit_per_scanner();
465
2.24k
        }
466
467
872k
        const bool no_runtime_filters = _total_rf_num == 0;
468
872k
        const bool segment_limit_enabled = _state->enable_segment_limit_pushdown();
469
872k
        const bool storage_no_merge = olap_scan_local_state->_storage_no_merge();
470
471
872k
        if (_limit > 0 && no_runtime_filters && segment_limit_enabled && storage_no_merge) {
472
3.98k
            for (const auto& conjunct : _conjuncts) {
473
0
                DORIS_CHECK(!olap_scan_local_state->_check_expr_storage_filter(
474
0
                        conjunct->root(), OlapScanLocalState::ExprStorageFilterCheckMode::
475
0
                                                  HAS_SEGMENT_EVALUABLE_EXPR));
476
0
            }
477
3.98k
        }
478
479
        // Segment LIMIT has only two legal states: completely disabled, or enabled after every
480
        // row-filtering conjunct has become a storage predicate or SegmentIterator common expr.
481
872k
        const bool can_push_down_segment_limit = _limit > 0 && no_runtime_filters &&
482
872k
                                                 _conjuncts.empty() && segment_limit_enabled &&
483
872k
                                                 storage_no_merge;
484
872k
        if (can_push_down_segment_limit) {
485
3.99k
            if (has_key_topn) {
486
2.13k
                _tablet_reader_params.read_orderby_key = true;
487
2.13k
                if (!olap_scan_node.sort_info.is_asc_order[0]) {
488
384
                    _tablet_reader_params.read_orderby_key_reverse = true;
489
384
                }
490
2.13k
                _tablet_reader_params.read_orderby_key_num_prefix_columns =
491
2.13k
                        olap_scan_node.sort_info.is_asc_order.size();
492
2.13k
                _tablet_reader_params.read_orderby_key_limit = _limit;
493
2.13k
            } else {
494
1.85k
                _tablet_reader_params.general_read_limit = _limit;
495
1.85k
            }
496
3.99k
        }
497
498
872k
        if (_tablet_reader_params.read_orderby_key_limit > 0 ||
499
872k
            _tablet_reader_params.general_read_limit > 0) {
500
3.99k
            DORIS_CHECK(can_push_down_segment_limit);
501
3.99k
            DORIS_CHECK(_conjuncts.empty());
502
3.99k
        }
503
504
        // A key TopN scan cannot share the plain LIMIT early-stop counter. If
505
        // storage TopN is pushed down, each scanner must produce its full local
506
        // candidates. If it is not pushed down for any reason, the upper TopN
507
        // still needs all rows from the scan.
508
872k
        if (has_key_topn) {
509
2.24k
            _shared_scan_limit = nullptr;
510
2.24k
            if (_tablet_reader_params.read_orderby_key_limit == 0) {
511
106
                _limit = -1;
512
106
            }
513
2.24k
        }
514
        // Note: _shared_scan_limit is intentionally not pushed into the
515
        // storage layer. SegmentIterator's _process_eof() is irreversible,
516
        // so a concurrently-decremented atomic could reach 0 while a segment
517
        // still has data needed by other scanners.
518
519
        // set push down topn filter
520
872k
        _tablet_reader_params.topn_filter_source_node_ids =
521
872k
                olap_scan_local_state->get_topn_filter_source_node_ids(_state, true);
522
872k
        if (!_tablet_reader_params.topn_filter_source_node_ids.empty()) {
523
5.27k
            _tablet_reader_params.topn_filter_target_node_id =
524
5.27k
                    olap_scan_local_state->parent()->node_id();
525
5.27k
        }
526
872k
    }
527
528
    // If this is a Two-Phase read query, and we need to delay the release of Rowset
529
    // by rowset->update_delayed_expired_timestamp().This could expand the lifespan of Rowset
530
874k
    if (tablet_schema->field_index(BeConsts::ROWID_COL) >= 0) {
531
0
        constexpr static int delayed_s = 60;
532
0
        for (auto rs_reader : _tablet_reader_params.rs_splits) {
533
0
            uint64_t delayed_expired_timestamp =
534
0
                    UnixSeconds() + _tablet_reader_params.runtime_state->execution_timeout() +
535
0
                    delayed_s;
536
0
            rs_reader.rs_reader->rowset()->update_delayed_expired_timestamp(
537
0
                    delayed_expired_timestamp);
538
0
            ExecEnv::GetInstance()->storage_engine().add_quering_rowset(
539
0
                    rs_reader.rs_reader->rowset());
540
0
        }
541
0
    }
542
543
874k
    if (tablet_schema->has_global_row_id()) {
544
7.18k
        auto& id_file_map = _state->get_id_file_map();
545
13.1k
        for (auto rs_reader : _tablet_reader_params.rs_splits) {
546
13.1k
            id_file_map->add_temp_rowset(rs_reader.rs_reader->rowset());
547
13.1k
        }
548
7.18k
    }
549
550
874k
    return Status::OK();
551
874k
}
552
553
873k
Status OlapScanner::_init_variant_columns() {
554
873k
    auto& tablet_schema = _tablet_reader_params.tablet_schema;
555
873k
    if (tablet_schema->num_variant_columns() == 0) {
556
868k
        return Status::OK();
557
868k
    }
558
    // Parent column has path info to distinction from each other
559
14.3k
    for (auto* slot : _output_tuple_desc->slots()) {
560
14.3k
        if (slot->type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
561
            // Such columns are not exist in frontend schema info, so we need to
562
            // add them into tablet_schema for later column indexing.
563
7.29k
            const auto& dt_variant =
564
7.29k
                    assert_cast<const DataTypeVariant&>(*remove_nullable(slot->type()));
565
7.29k
            TabletColumn subcol = TabletColumn::create_materialized_variant_column(
566
7.29k
                    tablet_schema->column_by_uid(slot->col_unique_id()).name_lower_case(),
567
7.29k
                    slot->column_paths(), slot->col_unique_id(),
568
7.29k
                    dt_variant.variant_max_subcolumns_count(), dt_variant.enable_doc_mode());
569
7.29k
            if (tablet_schema->field_index(*subcol.path_info_ptr()) < 0) {
570
5.35k
                tablet_schema->append_column(subcol, TabletSchema::ColumnType::VARIANT);
571
5.35k
            }
572
7.29k
        }
573
14.3k
    }
574
4.94k
    variant_util::inherit_column_attributes(tablet_schema);
575
4.94k
    return Status::OK();
576
873k
}
577
578
873k
Status OlapScanner::_init_return_columns() {
579
7.91M
    for (auto* slot : _output_tuple_desc->slots()) {
580
        // variant column using path to index a column
581
7.91M
        int32_t index = 0;
582
7.91M
        auto& tablet_schema = _tablet_reader_params.tablet_schema;
583
7.91M
        if (slot->type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
584
7.29k
            index = tablet_schema->field_index(PathInData(
585
7.29k
                    tablet_schema->column_by_uid(slot->col_unique_id()).name_lower_case(),
586
7.29k
                    slot->column_paths()));
587
7.90M
        } else {
588
7.91M
            index = slot->col_unique_id() >= 0 ? tablet_schema->field_index(slot->col_unique_id())
589
18.4E
                                               : tablet_schema->field_index(slot->col_name());
590
7.90M
        }
591
592
7.91M
        if (index < 0) {
593
0
            return Status::InternalError(
594
0
                    "field name is invalid. field={}, field_name_to_index={}, col_unique_id={}",
595
0
                    slot->col_name(), tablet_schema->get_all_field_names(), slot->col_unique_id());
596
0
        }
597
598
7.91M
        if (slot->get_virtual_column_expr()) {
599
479
            ColumnId virtual_column_cid = index;
600
479
            _virtual_column_exprs[virtual_column_cid] = _slot_id_to_virtual_column_expr[slot->id()];
601
479
            size_t idx_in_block = _slot_id_to_index_in_block[slot->id()];
602
479
            _vir_cid_to_idx_in_block[virtual_column_cid] = idx_in_block;
603
479
            _vir_col_idx_to_type[idx_in_block] = _slot_id_to_col_type[slot->id()];
604
605
479
            VLOG_DEBUG << fmt::format(
606
3
                    "Virtual column, slot id: {}, cid {}, column index: {}, type: {}", slot->id(),
607
3
                    virtual_column_cid, _vir_cid_to_idx_in_block[virtual_column_cid],
608
3
                    _vir_col_idx_to_type[idx_in_block]->get_name());
609
479
        }
610
611
7.91M
        const auto& column = tablet_schema->column(index);
612
7.91M
        int32_t unique_id =
613
7.91M
                column.unique_id() >= 0 ? column.unique_id() : column.parent_unique_id();
614
7.91M
        if (!slot->all_access_paths().empty()) {
615
79.5k
            _tablet_reader_params.all_access_paths.insert({unique_id, slot->all_access_paths()});
616
79.5k
        }
617
618
7.91M
        if (!slot->predicate_access_paths().empty()) {
619
9.45k
            _tablet_reader_params.predicate_access_paths.insert(
620
9.45k
                    {unique_id, slot->predicate_access_paths()});
621
9.45k
        }
622
623
7.91M
        if ((slot->type()->get_primitive_type() == PrimitiveType::TYPE_STRUCT ||
624
7.91M
             slot->type()->get_primitive_type() == PrimitiveType::TYPE_MAP ||
625
7.91M
             slot->type()->get_primitive_type() == PrimitiveType::TYPE_ARRAY) &&
626
7.91M
            !slot->all_access_paths().empty()) {
627
67.3k
            tablet_schema->add_pruned_columns_data_type(column.unique_id(), slot->type());
628
67.3k
        }
629
630
7.91M
        _return_columns.push_back(index);
631
7.91M
        if (slot->is_nullable() && !tablet_schema->column(index).is_nullable()) {
632
0
            _tablet_columns_convert_to_null_set.emplace(index);
633
7.91M
        } else if (!slot->is_nullable() && tablet_schema->column(index).is_nullable()) {
634
0
            return Status::Error<ErrorCode::INVALID_SCHEMA>(
635
0
                    "slot(id: {}, name: {})'s nullable does not match "
636
0
                    "column(tablet id: {}, index: {}, name: {}) ",
637
0
                    slot->id(), slot->col_name(), tablet_schema->table_id(), index,
638
0
                    tablet_schema->column(index).name());
639
0
        }
640
7.91M
    }
641
642
873k
    if (_return_columns.empty()) {
643
0
        return Status::InternalError("failed to build storage scanner, no materialized slot!");
644
0
    }
645
646
873k
    return Status::OK();
647
873k
}
648
649
1.83M
bool OlapScanner::check_partition_pruned() const {
650
1.83M
    if (!_local_state) {
651
0
        return false;
652
0
    }
653
1.83M
    return _local_state->is_partition_pruned(_tablet_reader_params.tablet->partition_id());
654
1.83M
}
655
656
920k
doris::TabletStorageType OlapScanner::get_storage_type() {
657
920k
    if (config::is_cloud_mode()) {
658
        // we don't have cold storage in cloud mode, all storage is treated as local
659
918k
        return doris::TabletStorageType::STORAGE_TYPE_LOCAL;
660
918k
    }
661
1.82k
    int local_reader = 0;
662
6.06k
    for (const auto& reader : _tablet_reader_params.rs_splits) {
663
6.06k
        local_reader += reader.rs_reader->rowset()->is_local();
664
6.06k
    }
665
1.82k
    int total_reader = _tablet_reader_params.rs_splits.size();
666
667
1.82k
    if (local_reader == total_reader) {
668
1.82k
        return doris::TabletStorageType::STORAGE_TYPE_LOCAL;
669
1.82k
    } else if (local_reader == 0) {
670
0
        return doris::TabletStorageType::STORAGE_TYPE_REMOTE;
671
0
    }
672
0
    return doris::TabletStorageType::STORAGE_TYPE_REMOTE_AND_LOCAL;
673
1.82k
}
674
675
1.12M
Status OlapScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {
676
    // Read one block from block reader
677
    // ATTN: Here we need to let the _get_block_impl method guarantee the semantics of the interface,
678
    // that is, eof can be set to true only when the returned block is empty.
679
1.12M
    RETURN_IF_ERROR(_tablet_reader->next_block_with_aggregation(block, eof));
680
1.12M
    if (block->rows() > 0) {
681
247k
        _tablet_reader_params.tablet->read_block_count.fetch_add(1, std::memory_order_relaxed);
682
247k
        *eof = false;
683
247k
    }
684
1.12M
#ifndef NDEBUG
685
1.12M
    RETURN_IF_ERROR(_check_ann_cache_hit_debug_points(_tablet_reader->stats()));
686
1.12M
#endif
687
1.12M
    return Status::OK();
688
1.12M
}
689
690
881k
Status OlapScanner::close(RuntimeState* state) {
691
881k
    if (!_try_close()) {
692
150
        return Status::OK();
693
150
    }
694
881k
    RETURN_IF_ERROR(Scanner::close(state));
695
881k
    return Status::OK();
696
881k
}
697
698
917k
void OlapScanner::update_realtime_counters() {
699
917k
    if (!_has_prepared) {
700
        // Counter update need prepare successfully, or it maybe core. For example, olap scanner
701
        // will open tablet reader during prepare, if not prepare successfully, tablet reader == nullptr.
702
0
        return;
703
0
    }
704
917k
    OlapScanLocalState* local_state = static_cast<OlapScanLocalState*>(_local_state);
705
917k
    const OlapReaderStatistics& stats = _tablet_reader->stats();
706
917k
    COUNTER_UPDATE(local_state->_read_compressed_counter, stats.compressed_bytes_read);
707
917k
    COUNTER_UPDATE(local_state->_read_uncompressed_counter, stats.uncompressed_bytes_read);
708
917k
    COUNTER_UPDATE(local_state->_scan_bytes, stats.uncompressed_bytes_read);
709
917k
    COUNTER_UPDATE(local_state->_scan_rows, stats.raw_rows_read);
710
711
    // Make sure the scan bytes and scan rows counter in audit log is the same as the counter in
712
    // doris metrics.
713
    // ScanBytes is the uncompressed bytes read from local + remote
714
    // bytes_read_from_local is the compressed bytes read from local
715
    // bytes_read_from_remote is the compressed bytes read from remote
716
    // scan bytes > bytes_read_from_local + bytes_read_from_remote
717
917k
    _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_rows(stats.raw_rows_read);
718
917k
    _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes(
719
917k
            stats.uncompressed_bytes_read);
720
721
    // In case of no cache, we still need to update the IO stats. uncompressed bytes read == local + remote
722
917k
    if (stats.file_cache_stats.bytes_read_from_local == 0 &&
723
917k
        stats.file_cache_stats.bytes_read_from_remote == 0) {
724
826k
        _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage(
725
826k
                stats.compressed_bytes_read);
726
826k
        DorisMetrics::instance()->query_scan_bytes_from_local->increment(
727
826k
                stats.compressed_bytes_read);
728
826k
    } else {
729
90.5k
        _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage(
730
90.5k
                stats.file_cache_stats.bytes_read_from_local);
731
90.5k
        _state->get_query_ctx()
732
90.5k
                ->resource_ctx()
733
90.5k
                ->io_context()
734
90.5k
                ->update_scan_bytes_from_remote_storage(
735
90.5k
                        stats.file_cache_stats.bytes_read_from_remote);
736
737
90.5k
        DorisMetrics::instance()->query_scan_bytes_from_local->increment(
738
90.5k
                stats.file_cache_stats.bytes_read_from_local);
739
90.5k
        DorisMetrics::instance()->query_scan_bytes_from_remote->increment(
740
90.5k
                stats.file_cache_stats.bytes_read_from_remote);
741
90.5k
    }
742
743
917k
    if (has_file_cache_statistics(stats.file_cache_stats)) {
744
93.1k
        io::FileCacheProfileReporter cache_profile(local_state->_segment_profile.get());
745
93.1k
        cache_profile.update(&stats.file_cache_stats);
746
93.1k
        _state->get_query_ctx()->resource_ctx()->io_context()->update_bytes_write_into_cache(
747
93.1k
                stats.file_cache_stats.bytes_write_into_cache);
748
93.1k
    }
749
750
917k
    _tablet_reader->mutable_stats()->compressed_bytes_read = 0;
751
917k
    _tablet_reader->mutable_stats()->uncompressed_bytes_read = 0;
752
917k
    _tablet_reader->mutable_stats()->raw_rows_read = 0;
753
917k
    _tablet_reader->mutable_stats()->file_cache_stats = {};
754
917k
}
755
756
871k
void OlapScanner::_collect_profile_before_close() {
757
    //  Please don't directly enable the profile here, we need to set QueryStatistics using the counter inside.
758
871k
    if (_has_updated_counter) {
759
0
        return;
760
0
    }
761
871k
    _has_updated_counter = true;
762
871k
    _tablet_reader->update_profile(_profile);
763
764
871k
    Scanner::_collect_profile_before_close();
765
766
    // Update counters for OlapScanner
767
    // Update counters from tablet reader's stats
768
871k
    auto& stats = _tablet_reader->stats();
769
871k
    auto* local_state = (OlapScanLocalState*)_local_state;
770
871k
    COUNTER_UPDATE(local_state->_io_timer, stats.io_ns);
771
871k
    COUNTER_UPDATE(local_state->_read_compressed_counter, stats.compressed_bytes_read);
772
871k
    COUNTER_UPDATE(local_state->_scan_bytes, stats.uncompressed_bytes_read);
773
871k
    COUNTER_UPDATE(local_state->_decompressor_timer, stats.decompress_ns);
774
871k
    COUNTER_UPDATE(local_state->_read_uncompressed_counter, stats.uncompressed_bytes_read);
775
871k
    COUNTER_UPDATE(local_state->_block_load_timer, stats.block_load_ns);
776
871k
    COUNTER_UPDATE(local_state->_block_load_counter, stats.blocks_load);
777
871k
    COUNTER_UPDATE(local_state->_block_fetch_timer, stats.block_fetch_ns);
778
871k
    COUNTER_UPDATE(local_state->_delete_bitmap_get_agg_timer, stats.delete_bitmap_get_agg_ns);
779
871k
    COUNTER_UPDATE(local_state->_scan_rows, stats.raw_rows_read);
780
871k
    COUNTER_UPDATE(local_state->_vec_cond_timer, stats.vec_cond_ns);
781
871k
    COUNTER_UPDATE(local_state->_short_cond_timer, stats.short_cond_ns);
782
871k
    COUNTER_UPDATE(local_state->_expr_filter_timer, stats.expr_filter_ns);
783
871k
    COUNTER_UPDATE(local_state->_block_init_timer, stats.block_init_ns);
784
871k
    COUNTER_UPDATE(local_state->_block_init_seek_timer, stats.block_init_seek_ns);
785
871k
    COUNTER_UPDATE(local_state->_block_init_seek_counter, stats.block_init_seek_num);
786
871k
    COUNTER_UPDATE(local_state->_segment_generate_row_range_by_keys_timer,
787
871k
                   stats.generate_row_ranges_by_keys_ns);
788
871k
    COUNTER_UPDATE(local_state->_segment_generate_row_range_by_column_conditions_timer,
789
871k
                   stats.generate_row_ranges_by_column_conditions_ns);
790
871k
    COUNTER_UPDATE(local_state->_segment_generate_row_range_by_bf_timer,
791
871k
                   stats.generate_row_ranges_by_bf_ns);
792
871k
    COUNTER_UPDATE(local_state->_collect_iterator_merge_next_timer,
793
871k
                   stats.collect_iterator_merge_next_timer);
794
871k
    COUNTER_UPDATE(local_state->_segment_generate_row_range_by_zonemap_timer,
795
871k
                   stats.generate_row_ranges_by_zonemap_ns);
796
871k
    COUNTER_UPDATE(local_state->_segment_generate_row_range_by_dict_timer,
797
871k
                   stats.generate_row_ranges_by_dict_ns);
798
871k
    COUNTER_UPDATE(local_state->_predicate_column_read_timer, stats.predicate_column_read_ns);
799
871k
    COUNTER_UPDATE(local_state->_non_predicate_column_read_timer, stats.non_predicate_read_ns);
800
871k
    COUNTER_UPDATE(local_state->_predicate_column_read_seek_timer,
801
871k
                   stats.predicate_column_read_seek_ns);
802
871k
    COUNTER_UPDATE(local_state->_predicate_column_read_seek_counter,
803
871k
                   stats.predicate_column_read_seek_num);
804
871k
    COUNTER_UPDATE(local_state->_lazy_read_timer, stats.lazy_read_ns);
805
871k
    COUNTER_UPDATE(local_state->_lazy_read_seek_timer, stats.block_lazy_read_seek_ns);
806
871k
    COUNTER_UPDATE(local_state->_lazy_read_seek_counter, stats.block_lazy_read_seek_num);
807
871k
    COUNTER_UPDATE(local_state->_output_col_timer, stats.output_col_ns);
808
871k
    COUNTER_UPDATE(local_state->_rows_vec_cond_filtered_counter, stats.rows_vec_cond_filtered);
809
871k
    COUNTER_UPDATE(local_state->_rows_short_circuit_cond_filtered_counter,
810
871k
                   stats.rows_short_circuit_cond_filtered);
811
871k
    COUNTER_UPDATE(local_state->_rows_expr_cond_filtered_counter, stats.rows_expr_cond_filtered);
812
871k
    COUNTER_UPDATE(local_state->_rows_vec_cond_input_counter, stats.vec_cond_input_rows);
813
871k
    COUNTER_UPDATE(local_state->_rows_short_circuit_cond_input_counter,
814
871k
                   stats.short_circuit_cond_input_rows);
815
871k
    COUNTER_UPDATE(local_state->_rows_expr_cond_input_counter, stats.expr_cond_input_rows);
816
871k
    COUNTER_UPDATE(local_state->_stats_filtered_counter, stats.rows_stats_filtered);
817
871k
    COUNTER_UPDATE(local_state->_stats_rp_filtered_counter, stats.rows_stats_rp_filtered);
818
871k
    COUNTER_UPDATE(local_state->_dict_filtered_counter, stats.segment_dict_filtered);
819
871k
    COUNTER_UPDATE(local_state->_bf_filtered_counter, stats.rows_bf_filtered);
820
871k
    COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_del_filtered);
821
871k
    COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_del_by_bitmap);
822
871k
    COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_vec_del_cond_filtered);
823
871k
    COUNTER_UPDATE(local_state->_conditions_filtered_counter, stats.rows_conditions_filtered);
824
871k
    COUNTER_UPDATE(local_state->_key_range_filtered_counter, stats.rows_key_range_filtered);
825
871k
    COUNTER_UPDATE(local_state->_total_pages_num_counter, stats.total_pages_num);
826
871k
    COUNTER_UPDATE(local_state->_cached_pages_num_counter, stats.cached_pages_num);
827
871k
    COUNTER_UPDATE(local_state->_inverted_index_filter_counter, stats.rows_inverted_index_filtered);
828
871k
    COUNTER_UPDATE(local_state->_inverted_index_filter_timer, stats.inverted_index_filter_timer);
829
871k
    COUNTER_UPDATE(local_state->_inverted_index_query_cache_hit_counter,
830
871k
                   stats.inverted_index_query_cache_hit);
831
871k
    COUNTER_UPDATE(local_state->_inverted_index_query_cache_miss_counter,
832
871k
                   stats.inverted_index_query_cache_miss);
833
871k
    COUNTER_UPDATE(local_state->_inverted_index_query_timer, stats.inverted_index_query_timer);
834
871k
    COUNTER_UPDATE(local_state->_inverted_index_query_null_bitmap_timer,
835
871k
                   stats.inverted_index_query_null_bitmap_timer);
836
871k
    COUNTER_UPDATE(local_state->_inverted_index_query_bitmap_copy_timer,
837
871k
                   stats.inverted_index_query_bitmap_copy_timer);
838
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_open_timer,
839
871k
                   stats.inverted_index_searcher_open_timer);
840
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_search_timer,
841
871k
                   stats.inverted_index_searcher_search_timer);
842
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_search_init_timer,
843
871k
                   stats.inverted_index_searcher_search_init_timer);
844
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_search_exec_timer,
845
871k
                   stats.inverted_index_searcher_search_exec_timer);
846
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_cache_hit_counter,
847
871k
                   stats.inverted_index_searcher_cache_hit);
848
871k
    COUNTER_UPDATE(local_state->_inverted_index_searcher_cache_miss_counter,
849
871k
                   stats.inverted_index_searcher_cache_miss);
850
871k
    COUNTER_UPDATE(local_state->_inverted_index_downgrade_count_counter,
851
871k
                   stats.inverted_index_downgrade_count);
852
871k
    COUNTER_UPDATE(local_state->_inverted_index_analyzer_timer,
853
871k
                   stats.inverted_index_analyzer_timer);
854
871k
    COUNTER_UPDATE(local_state->_inverted_index_lookup_timer, stats.inverted_index_lookup_timer);
855
871k
    COUNTER_UPDATE(local_state->_variant_scan_sparse_column_timer,
856
871k
                   stats.variant_scan_sparse_column_timer_ns);
857
871k
    COUNTER_UPDATE(local_state->_variant_scan_sparse_column_bytes,
858
871k
                   stats.variant_scan_sparse_column_bytes);
859
871k
    COUNTER_UPDATE(local_state->_variant_fill_path_from_sparse_column_timer,
860
871k
                   stats.variant_fill_path_from_sparse_column_timer_ns);
861
871k
    COUNTER_UPDATE(local_state->_variant_subtree_default_iter_count,
862
871k
                   stats.variant_subtree_default_iter_count);
863
871k
    COUNTER_UPDATE(local_state->_variant_subtree_leaf_iter_count,
864
871k
                   stats.variant_subtree_leaf_iter_count);
865
871k
    COUNTER_UPDATE(local_state->_variant_subtree_hierarchical_iter_count,
866
871k
                   stats.variant_subtree_hierarchical_iter_count);
867
871k
    COUNTER_UPDATE(local_state->_variant_subtree_sparse_iter_count,
868
871k
                   stats.variant_subtree_sparse_iter_count);
869
871k
    COUNTER_UPDATE(local_state->_variant_doc_value_column_iter_count,
870
871k
                   stats.variant_doc_value_column_iter_count);
871
872
871k
    if (stats.adaptive_batch_size_predict_max_rows > 0) {
873
605k
        local_state->_adaptive_batch_predict_min_rows_counter->set(
874
605k
                stats.adaptive_batch_size_predict_min_rows);
875
605k
        local_state->_adaptive_batch_predict_max_rows_counter->set(
876
605k
                stats.adaptive_batch_size_predict_max_rows);
877
605k
    }
878
879
871k
    InvertedIndexProfileReporter inverted_index_profile;
880
871k
    inverted_index_profile.update(local_state->_index_filter_profile.get(),
881
871k
                                  &stats.inverted_index_stats);
882
883
871k
    if (has_file_cache_statistics(stats.file_cache_stats)) {
884
0
        io::FileCacheProfileReporter cache_profile(local_state->_segment_profile.get());
885
0
        cache_profile.update(&stats.file_cache_stats);
886
0
        _state->get_query_ctx()->resource_ctx()->io_context()->update_bytes_write_into_cache(
887
0
                stats.file_cache_stats.bytes_write_into_cache);
888
0
    }
889
871k
    COUNTER_UPDATE(local_state->_output_index_result_column_timer,
890
871k
                   stats.output_index_result_column_timer);
891
871k
    COUNTER_UPDATE(local_state->_filtered_segment_counter, stats.filtered_segment_number);
892
871k
    COUNTER_UPDATE(local_state->_total_segment_counter, stats.total_segment_number);
893
871k
    COUNTER_UPDATE(local_state->_condition_cache_hit_counter, stats.condition_cache_hit_seg_nums);
894
871k
    COUNTER_UPDATE(local_state->_condition_cache_filtered_rows_counter,
895
871k
                   stats.condition_cache_filtered_rows);
896
897
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_timer, stats.tablet_reader_init_timer_ns);
898
871k
    COUNTER_UPDATE(local_state->_tablet_reader_capture_rs_readers_timer,
899
871k
                   stats.tablet_reader_capture_rs_readers_timer_ns);
900
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_return_columns_timer,
901
871k
                   stats.tablet_reader_init_return_columns_timer_ns);
902
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_keys_param_timer,
903
871k
                   stats.tablet_reader_init_keys_param_timer_ns);
904
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_orderby_keys_param_timer,
905
871k
                   stats.tablet_reader_init_orderby_keys_param_timer_ns);
906
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_conditions_param_timer,
907
871k
                   stats.tablet_reader_init_conditions_param_timer_ns);
908
871k
    COUNTER_UPDATE(local_state->_tablet_reader_init_delete_condition_param_timer,
909
871k
                   stats.tablet_reader_init_delete_condition_param_timer_ns);
910
871k
    COUNTER_UPDATE(local_state->_block_reader_vcollect_iter_init_timer,
911
871k
                   stats.block_reader_vcollect_iter_init_timer_ns);
912
871k
    COUNTER_UPDATE(local_state->_block_reader_rs_readers_init_timer,
913
871k
                   stats.block_reader_rs_readers_init_timer_ns);
914
871k
    COUNTER_UPDATE(local_state->_block_reader_build_heap_init_timer,
915
871k
                   stats.block_reader_build_heap_init_timer_ns);
916
917
871k
    COUNTER_UPDATE(local_state->_rowset_reader_get_segment_iterators_timer,
918
871k
                   stats.rowset_reader_get_segment_iterators_timer_ns);
919
871k
    COUNTER_UPDATE(local_state->_rowset_reader_create_iterators_timer,
920
871k
                   stats.rowset_reader_create_iterators_timer_ns);
921
871k
    COUNTER_UPDATE(local_state->_rowset_reader_init_iterators_timer,
922
871k
                   stats.rowset_reader_init_iterators_timer_ns);
923
871k
    COUNTER_UPDATE(local_state->_rowset_reader_load_segments_timer,
924
871k
                   stats.rowset_reader_load_segments_timer_ns);
925
926
871k
    COUNTER_UPDATE(local_state->_segment_iterator_init_timer, stats.segment_iterator_init_timer_ns);
927
871k
    COUNTER_UPDATE(local_state->_segment_iterator_init_return_column_iterators_timer,
928
871k
                   stats.segment_iterator_init_return_column_iterators_timer_ns);
929
871k
    COUNTER_UPDATE(local_state->_segment_iterator_init_index_iterators_timer,
930
871k
                   stats.segment_iterator_init_index_iterators_timer_ns);
931
871k
    COUNTER_UPDATE(local_state->_segment_iterator_init_segment_prefetchers_timer,
932
871k
                   stats.segment_iterator_init_segment_prefetchers_timer_ns);
933
934
871k
    COUNTER_UPDATE(local_state->_segment_create_column_readers_timer,
935
871k
                   stats.segment_create_column_readers_timer_ns);
936
871k
    COUNTER_UPDATE(local_state->_segment_load_index_timer, stats.segment_load_index_timer_ns);
937
938
    // Update metrics
939
871k
    DorisMetrics::instance()->query_scan_bytes->increment(
940
871k
            local_state->_read_uncompressed_counter->value());
941
871k
    DorisMetrics::instance()->query_scan_rows->increment(local_state->_scan_rows->value());
942
871k
    auto& tablet = _tablet_reader_params.tablet;
943
871k
    tablet->query_scan_bytes->increment(local_state->_read_uncompressed_counter->value());
944
871k
    tablet->query_scan_rows->increment(local_state->_scan_rows->value());
945
871k
    tablet->query_scan_count->increment(1);
946
947
871k
    COUNTER_UPDATE(local_state->_ann_range_search_filter_counter,
948
871k
                   stats.rows_ann_index_range_filtered);
949
871k
    COUNTER_UPDATE(local_state->_ann_topn_filter_counter, stats.rows_ann_index_topn_filtered);
950
871k
    COUNTER_UPDATE(local_state->_ann_index_load_costs, stats.ann_index_load_ns);
951
871k
    COUNTER_UPDATE(local_state->_ann_ivf_on_disk_load_costs, stats.ann_ivf_on_disk_load_ns);
952
871k
    COUNTER_UPDATE(local_state->_ann_ivf_on_disk_cache_hit_cnt,
953
871k
                   stats.ann_ivf_on_disk_cache_hit_cnt);
954
871k
    COUNTER_UPDATE(local_state->_ann_ivf_on_disk_cache_miss_cnt,
955
871k
                   stats.ann_ivf_on_disk_cache_miss_cnt);
956
871k
    COUNTER_UPDATE(local_state->_ann_range_search_costs, stats.ann_index_range_search_ns);
957
871k
    COUNTER_UPDATE(local_state->_ann_range_search_cnt, stats.ann_index_range_search_cnt);
958
871k
    COUNTER_UPDATE(local_state->_ann_range_engine_search_costs, stats.ann_range_engine_search_ns);
959
    // Engine prepare before search
960
871k
    COUNTER_UPDATE(local_state->_ann_range_pre_process_costs, stats.ann_range_pre_process_ns);
961
    // Post process parent: Doris result process + engine convert
962
871k
    COUNTER_UPDATE(local_state->_ann_range_post_process_costs,
963
871k
                   stats.ann_range_result_convert_ns + stats.ann_range_engine_convert_ns);
964
    // Engine convert (child under post-process)
965
871k
    COUNTER_UPDATE(local_state->_ann_range_engine_convert_costs, stats.ann_range_engine_convert_ns);
966
    // Doris-side result convert (child under post-process)
967
871k
    COUNTER_UPDATE(local_state->_ann_range_result_convert_costs, stats.ann_range_result_convert_ns);
968
969
871k
    COUNTER_UPDATE(local_state->_ann_topn_search_costs, stats.ann_topn_search_ns);
970
871k
    COUNTER_UPDATE(local_state->_ann_topn_search_cnt, stats.ann_index_topn_search_cnt);
971
871k
    COUNTER_UPDATE(local_state->_ann_cache_hit_cnt, stats.ann_index_cache_hits);
972
871k
    COUNTER_UPDATE(local_state->_ann_range_cache_hit_cnt, stats.ann_index_range_cache_hits);
973
974
    // Detailed ANN timers
975
    // ANN TopN timers with hierarchy
976
    // Engine search time (FAISS)
977
871k
    COUNTER_UPDATE(local_state->_ann_topn_engine_search_costs,
978
871k
                   stats.ann_index_topn_engine_search_ns);
979
    // Engine prepare time (allocations/buffer setup before search)
980
871k
    COUNTER_UPDATE(local_state->_ann_topn_pre_process_costs,
981
871k
                   stats.ann_index_topn_engine_prepare_ns);
982
    // Post process parent includes Doris result processing + engine convert
983
871k
    COUNTER_UPDATE(local_state->_ann_topn_post_process_costs,
984
871k
                   stats.ann_index_topn_result_process_ns + stats.ann_index_topn_engine_convert_ns);
985
    // Engine-side conversion time inside FAISS wrappers (child under post-process)
986
871k
    COUNTER_UPDATE(local_state->_ann_topn_engine_convert_costs,
987
871k
                   stats.ann_index_topn_engine_convert_ns);
988
989
    // Doris-side result convert costs (show separately as another child counter); use pure process time
990
871k
    COUNTER_UPDATE(local_state->_ann_topn_result_convert_costs,
991
871k
                   stats.ann_index_topn_result_process_ns);
992
993
871k
    COUNTER_UPDATE(local_state->_ann_fallback_brute_force_cnt, stats.ann_fall_back_brute_force_cnt);
994
995
    // Overhead counter removed; precise instrumentation is reported via engine_prepare above.
996
871k
}
997
998
#ifndef NDEBUG
999
1.11M
Status OlapScanner::_check_ann_cache_hit_debug_points(const OlapReaderStatistics& stats) {
1000
1.11M
    DBUG_EXECUTE_IF("olap_scanner.ann_topn_cache_hits", {
1001
1.11M
        auto expected_hits = dp->param<int32_t>("expected_hits", -1);
1002
1.11M
        auto min_hits = dp->param<int32_t>("min_hits", -1);
1003
1.11M
        if (expected_hits >= 0 && stats.ann_index_cache_hits != expected_hits) {
1004
1.11M
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
1005
1.11M
                    "ann_index_cache_hits: {} not equal to expected: {}",
1006
1.11M
                    stats.ann_index_cache_hits, expected_hits);
1007
1.11M
        }
1008
1.11M
        if (min_hits >= 0 && stats.ann_index_cache_hits < min_hits) {
1009
1.11M
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
1010
1.11M
                    "ann_index_cache_hits: {} less than expected min: {}",
1011
1.11M
                    stats.ann_index_cache_hits, min_hits);
1012
1.11M
        }
1013
1.11M
    })
1014
1.11M
    DBUG_EXECUTE_IF("olap_scanner.ann_range_cache_hits", {
1015
1.11M
        auto expected_hits = dp->param<int32_t>("expected_hits", -1);
1016
1.11M
        auto min_hits = dp->param<int32_t>("min_hits", -1);
1017
1.11M
        if (expected_hits >= 0 && stats.ann_index_range_cache_hits != expected_hits) {
1018
1.11M
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
1019
1.11M
                    "ann_index_range_cache_hits: {} not equal to expected: {}",
1020
1.11M
                    stats.ann_index_range_cache_hits, expected_hits);
1021
1.11M
        }
1022
1.11M
        if (min_hits >= 0 && stats.ann_index_range_cache_hits < min_hits) {
1023
1.11M
            return Status::Error<ErrorCode::INTERNAL_ERROR>(
1024
1.11M
                    "ann_index_range_cache_hits: {} less than expected min: {}",
1025
1.11M
                    stats.ann_index_range_cache_hits, min_hits);
1026
1.11M
        }
1027
1.11M
    })
1028
1.11M
    return Status::OK();
1029
1.11M
}
1030
#endif
1031
1032
#include "common/compile_check_avoid_end.h"
1033
} // namespace doris