Coverage Report

Created: 2026-04-03 07:19

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