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 | 1.15M | : Scanner(params.state, parent, params.limit, params.profile), |
72 | 1.15M | _key_ranges(std::move(params.key_ranges)), |
73 | 1.15M | _tablet_reader_params({.tablet = std::move(params.tablet), |
74 | 1.15M | .tablet_schema {}, |
75 | 1.15M | .aggregation = params.aggregation, |
76 | 1.15M | .version = {0, params.version}, |
77 | 1.15M | .start_key {}, |
78 | 1.15M | .end_key {}, |
79 | 1.15M | .predicates {}, |
80 | 1.15M | .function_filters {}, |
81 | 1.15M | .delete_predicates {}, |
82 | 1.15M | .target_cast_type_for_variants {}, |
83 | 1.15M | .all_access_paths {}, |
84 | 1.15M | .predicate_access_paths {}, |
85 | 1.15M | .rs_splits {}, |
86 | 1.15M | .return_columns {}, |
87 | 1.15M | .output_columns {}, |
88 | 1.15M | .remaining_conjunct_roots {}, |
89 | 1.15M | .common_expr_ctxs_push_down {}, |
90 | 1.15M | .topn_filter_source_node_ids {}, |
91 | 1.15M | .filter_block_conjuncts {}, |
92 | 1.15M | .key_group_cluster_key_idxes {}, |
93 | 1.15M | .virtual_column_exprs {}, |
94 | 1.15M | .vir_cid_to_idx_in_block {}, |
95 | 1.15M | .vir_col_idx_to_type {}, |
96 | 1.15M | .score_runtime {}, |
97 | 1.15M | .collection_statistics {}, |
98 | 1.15M | .ann_topn_runtime {}, |
99 | 1.15M | .condition_cache_digest = parent->get_condition_cache_digest()}) { |
100 | 1.15M | _tablet_reader_params.set_read_source(std::move(params.read_source), |
101 | 1.15M | _state->skip_delete_bitmap()); |
102 | 1.15M | _has_prepared = false; |
103 | 1.15M | _vector_search_params = params.state->get_vector_search_params(); |
104 | 1.15M | } |
105 | | |
106 | | static std::string read_columns_to_string(TabletSchemaSPtr tablet_schema, |
107 | 3.12k | 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 | 3.12k | const int col_per_line = 10; |
111 | 3.12k | int i = 0; |
112 | 3.12k | std::string read_columns_string; |
113 | 3.12k | read_columns_string += "["; |
114 | 16.1k | for (auto it = read_columns.cbegin(); it != read_columns.cend(); it++) { |
115 | 13.0k | if (it != read_columns.cbegin()) { |
116 | 9.99k | read_columns_string += ", "; |
117 | 9.99k | } |
118 | 13.0k | read_columns_string += tablet_schema->columns().at(*it)->name(); |
119 | 13.0k | if (i >= col_per_line) { |
120 | 13 | read_columns_string += "\n"; |
121 | 13 | i = 0; |
122 | 13.0k | } else { |
123 | 13.0k | ++i; |
124 | 13.0k | } |
125 | 13.0k | } |
126 | 3.12k | read_columns_string += "]"; |
127 | 3.12k | return read_columns_string; |
128 | 3.12k | } |
129 | | |
130 | 1.15M | Status OlapScanner::prepare() { |
131 | 1.15M | auto* local_state = static_cast<OlapScanLocalState*>(_local_state); |
132 | 1.15M | auto& tablet = _tablet_reader_params.tablet; |
133 | 1.15M | auto& tablet_schema = _tablet_reader_params.tablet_schema; |
134 | 1.15M | DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", { |
135 | 1.15M | LOG_WARNING("CloudTablet.capture_rs_readers.return e-230 init") |
136 | 1.15M | .tag("tablet_id", tablet->tablet_id()); |
137 | 1.15M | return Status::Error<false>(-230, "injected error"); |
138 | 1.15M | }); |
139 | | |
140 | 1.15M | for (auto& ctx : local_state->_common_expr_ctxs_push_down) { |
141 | 13.3k | VExprContextSPtr context; |
142 | 13.3k | RETURN_IF_ERROR(ctx->clone(_state, context)); |
143 | 13.3k | _common_expr_ctxs_push_down.emplace_back(context); |
144 | 13.3k | context->prepare_ann_range_search(_vector_search_params); |
145 | 13.3k | } |
146 | | |
147 | 1.15M | 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 | 461 | VExprContextSPtr context; |
150 | 461 | RETURN_IF_ERROR(pair.second->clone(_state, context)); |
151 | 461 | _slot_id_to_virtual_column_expr[pair.first] = context; |
152 | 461 | } |
153 | | |
154 | 1.15M | _slot_id_to_index_in_block = local_state->_slot_id_to_index_in_block; |
155 | 1.15M | _slot_id_to_col_type = local_state->_slot_id_to_col_type; |
156 | 1.15M | _score_runtime = local_state->_score_runtime; |
157 | | // All scanners share the same ann_topn_runtime. |
158 | 1.15M | _ann_topn_runtime = local_state->_ann_topn_runtime; |
159 | | |
160 | | // set limit to reduce end of rowset and segment mem use |
161 | 1.15M | _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 | 1.15M | _tablet_reader->set_batch_size(_state->batch_size()); |
167 | 1.15M | TabletSchemaSPtr cached_schema; |
168 | 1.15M | std::string schema_key; |
169 | 1.15M | { |
170 | 1.15M | TOlapScanNode& olap_scan_node = local_state->olap_scan_node(); |
171 | | |
172 | 1.15M | const auto check_can_use_cache = [&]() { |
173 | 1.15M | if (!(olap_scan_node.__isset.schema_version && olap_scan_node.__isset.columns_desc && |
174 | 1.15M | !olap_scan_node.columns_desc.empty() && |
175 | 1.15M | olap_scan_node.columns_desc[0].col_unique_id >= 0 && // Why check first column? |
176 | 1.15M | tablet->tablet_schema()->num_variant_columns() == 0 && |
177 | 1.15M | tablet->tablet_schema()->num_virtual_columns() == 0)) { |
178 | 5.59k | return false; |
179 | 5.59k | } |
180 | | |
181 | | // If `delete_predicates` is not empty, will merge the columns in delete predicate into current tablet schema |
182 | 1.15M | if (!_tablet_reader_params.delete_predicates.empty()) { |
183 | 3.47k | return false; |
184 | 3.47k | } |
185 | | |
186 | 1.14M | const bool has_pruned_column = |
187 | 12.2M | std::ranges::any_of(_output_tuple_desc->slots(), [](const auto& slot) { |
188 | 12.2M | if ((slot->type()->get_primitive_type() == PrimitiveType::TYPE_STRUCT || |
189 | 12.2M | slot->type()->get_primitive_type() == PrimitiveType::TYPE_MAP || |
190 | 12.2M | slot->type()->get_primitive_type() == PrimitiveType::TYPE_ARRAY) && |
191 | 12.2M | !slot->all_access_paths().empty()) { |
192 | 23.8k | return true; |
193 | 23.8k | } |
194 | 12.2M | return false; |
195 | 12.2M | }); |
196 | 1.14M | return !has_pruned_column; |
197 | 1.15M | }(); |
198 | | |
199 | 1.15M | if (check_can_use_cache) { |
200 | 1.12M | schema_key = |
201 | 1.12M | SchemaCache::get_schema_key(tablet->tablet_id(), olap_scan_node.columns_desc, |
202 | 1.12M | olap_scan_node.schema_version); |
203 | 1.12M | cached_schema = SchemaCache::instance()->get_schema(schema_key); |
204 | 1.12M | } |
205 | 1.15M | if (cached_schema && cached_schema->num_virtual_columns() == 0) { |
206 | 991k | tablet_schema = cached_schema; |
207 | 991k | } else { |
208 | | // If schema is not cached or cached schema has virtual columns, |
209 | | // we need to create a new TabletSchema. |
210 | 166k | tablet_schema = std::make_shared<TabletSchema>(); |
211 | 166k | tablet_schema->copy_from(*tablet->tablet_schema()); |
212 | 166k | if (olap_scan_node.__isset.columns_desc && !olap_scan_node.columns_desc.empty() && |
213 | 166k | 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 | 165k | tablet_schema->clear_columns(); |
219 | 2.31M | for (const auto& column_desc : olap_scan_node.columns_desc) { |
220 | 2.31M | tablet_schema->append_column(TabletColumn(column_desc)); |
221 | 2.31M | } |
222 | 165k | if (olap_scan_node.__isset.schema_version) { |
223 | 165k | tablet_schema->set_schema_version(olap_scan_node.schema_version); |
224 | 165k | } |
225 | 165k | } |
226 | 166k | if (olap_scan_node.__isset.indexes_desc) { |
227 | 165k | tablet_schema->update_indexes_from_thrift(olap_scan_node.indexes_desc); |
228 | 165k | } |
229 | 166k | } |
230 | | |
231 | 1.15M | 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 | 1.15M | RETURN_IF_ERROR(_init_tablet_reader_params( |
276 | 1.15M | local_state->_parent->cast<OlapScanOperatorX>()._slot_id_to_slot_desc, _key_ranges, |
277 | 1.15M | local_state->_slot_id_to_predicates, local_state->_push_down_functions)); |
278 | 1.15M | } |
279 | | |
280 | | // add read columns in profile |
281 | 1.15M | if (_state->enable_profile()) { |
282 | 3.12k | _profile->add_info_string("ReadColumns", |
283 | 3.12k | read_columns_to_string(tablet_schema, _return_columns)); |
284 | 3.12k | } |
285 | | |
286 | | // Add newly created tablet schema to schema cache if it does not have virtual columns. |
287 | 1.15M | if (cached_schema == nullptr && !schema_key.empty() && |
288 | 1.15M | tablet_schema->num_virtual_columns() == 0 && !tablet_schema->has_pruned_columns()) { |
289 | 131k | SchemaCache::instance()->insert_schema(schema_key, tablet_schema); |
290 | 131k | } |
291 | | |
292 | 1.15M | 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 | 1.15M | _has_prepared = true; |
310 | 1.15M | return Status::OK(); |
311 | 1.15M | } |
312 | | |
313 | 1.15M | Status OlapScanner::_open_impl(RuntimeState* state) { |
314 | 1.15M | RETURN_IF_ERROR(Scanner::_open_impl(state)); |
315 | 1.15M | SCOPED_TIMER(_local_state->cast<OlapScanLocalState>()._reader_init_timer); |
316 | | |
317 | 1.15M | auto res = _tablet_reader->init(_tablet_reader_params); |
318 | 1.15M | if (!res.ok()) { |
319 | 55 | res.append("failed to initialize storage reader. tablet=" + |
320 | 55 | std::to_string(_tablet_reader_params.tablet->tablet_id()) + |
321 | 55 | ", backend=" + BackendOptions::get_localhost()); |
322 | 55 | return res; |
323 | 55 | } |
324 | | |
325 | | // Do not hold rs_splits any more to release memory. |
326 | 1.15M | _tablet_reader_params.rs_splits.clear(); |
327 | | |
328 | 1.15M | return Status::OK(); |
329 | 1.15M | } |
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 | 1.15M | 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 | 1.15M | const bool single_version = _tablet_reader_params.has_single_version(); |
340 | | |
341 | 1.15M | auto* olap_local_state = static_cast<OlapScanLocalState*>(_local_state); |
342 | 1.15M | bool read_mor_as_dup = olap_local_state->olap_scan_node().__isset.read_mor_as_dup && |
343 | 1.15M | olap_local_state->olap_scan_node().read_mor_as_dup; |
344 | 1.15M | 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 | 1.15M | } else { |
348 | 1.15M | auto push_down_agg_type = _local_state->get_push_down_agg_type(); |
349 | 1.15M | _tablet_reader_params.direct_mode = _tablet_reader_params.aggregation || single_version || |
350 | 1.15M | (push_down_agg_type != TPushAggOp::NONE && |
351 | 10.4k | push_down_agg_type != TPushAggOp::COUNT_ON_INDEX); |
352 | 1.15M | } |
353 | | |
354 | 1.15M | RETURN_IF_ERROR(_init_variant_columns()); |
355 | 1.15M | RETURN_IF_ERROR(_init_return_columns()); |
356 | | |
357 | 1.15M | _tablet_reader_params.reader_type = ReaderType::READER_QUERY; |
358 | 1.15M | _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 | 1.15M | if (_common_expr_ctxs_push_down.empty()) { |
362 | 1.14M | for (auto& conjunct : _conjuncts) { |
363 | 12.4k | _tablet_reader_params.remaining_conjunct_roots.emplace_back(conjunct->root()); |
364 | 12.4k | } |
365 | 1.14M | } else { |
366 | 13.2k | for (auto& ctx : _common_expr_ctxs_push_down) { |
367 | 13.2k | _tablet_reader_params.remaining_conjunct_roots.emplace_back(ctx->root()); |
368 | 13.2k | } |
369 | 10.8k | } |
370 | | |
371 | 1.15M | _tablet_reader_params.common_expr_ctxs_push_down = _common_expr_ctxs_push_down; |
372 | 1.15M | _tablet_reader_params.virtual_column_exprs = _virtual_column_exprs; |
373 | 1.15M | _tablet_reader_params.vir_cid_to_idx_in_block = _vir_cid_to_idx_in_block; |
374 | 1.15M | _tablet_reader_params.vir_col_idx_to_type = _vir_col_idx_to_type; |
375 | 1.15M | _tablet_reader_params.score_runtime = _score_runtime; |
376 | 1.15M | _tablet_reader_params.output_columns = ((OlapScanLocalState*)_local_state)->_output_column_ids; |
377 | 1.15M | _tablet_reader_params.ann_topn_runtime = _ann_topn_runtime; |
378 | 1.15M | for (const auto& ele : ((OlapScanLocalState*)_local_state)->_cast_types_for_variants) { |
379 | 1.48k | _tablet_reader_params.target_cast_type_for_variants[ele.first] = ele.second; |
380 | 1.48k | }; |
381 | 1.15M | auto& tablet_schema = _tablet_reader_params.tablet_schema; |
382 | 9.33M | for (auto& predicates : slot_to_predicates) { |
383 | 9.33M | const int sid = predicates.first; |
384 | 9.33M | DCHECK(slot_id_to_slot_desc.contains(sid)); |
385 | 9.33M | int32_t index = |
386 | 9.33M | tablet_schema->field_index(slot_id_to_slot_desc.find(sid)->second->col_name()); |
387 | 9.33M | 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 | 9.33M | for (auto& predicate : predicates.second) { |
393 | 938k | _tablet_reader_params.predicates.push_back(predicate->clone(index)); |
394 | 938k | } |
395 | 9.33M | } |
396 | | |
397 | 1.15M | std::copy(function_filters.cbegin(), function_filters.cend(), |
398 | 1.15M | std::inserter(_tablet_reader_params.function_filters, |
399 | 1.15M | _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 | 1.15M | for (auto& del_pred : _tablet_reader_params.delete_predicates) { |
403 | 7.14k | tablet_schema->merge_dropped_columns(*del_pred->tablet_schema()); |
404 | 7.14k | } |
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.79M | for (auto* key_range : key_ranges) { |
410 | 1.79M | if (!key_range->has_lower_bound) { |
411 | 151k | continue; |
412 | 151k | } |
413 | | |
414 | 1.64M | _tablet_reader_params.start_key_include = key_range->begin_include; |
415 | 1.64M | _tablet_reader_params.end_key_include = key_range->end_include; |
416 | | |
417 | 1.64M | _tablet_reader_params.start_key.push_back(key_range->begin_scan_range); |
418 | 1.64M | _tablet_reader_params.end_key.push_back(key_range->end_scan_range); |
419 | 1.64M | } |
420 | | |
421 | 1.15M | _tablet_reader_params.profile = _local_state->custom_profile(); |
422 | 1.15M | _tablet_reader_params.runtime_state = _state; |
423 | | |
424 | 1.15M | _tablet_reader_params.origin_return_columns = &_return_columns; |
425 | 1.15M | _tablet_reader_params.tablet_columns_convert_to_null_set = &_tablet_columns_convert_to_null_set; |
426 | | |
427 | 1.15M | if (_tablet_reader_params.direct_mode) { |
428 | 1.14M | _tablet_reader_params.return_columns = _return_columns; |
429 | 1.14M | } else { |
430 | | // we need to fetch all key columns to do the right aggregation on storage engine side. |
431 | 38.2k | for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) { |
432 | 26.8k | _tablet_reader_params.return_columns.push_back(i); |
433 | 26.8k | } |
434 | 49.7k | for (auto index : _return_columns) { |
435 | 49.7k | if (tablet_schema->column(index).is_key()) { |
436 | 18.2k | continue; |
437 | 18.2k | } |
438 | 31.4k | _tablet_reader_params.return_columns.push_back(index); |
439 | 31.4k | } |
440 | | // expand the sequence column |
441 | 11.3k | if (tablet_schema->has_sequence_col() || tablet_schema->has_seq_map()) { |
442 | 40 | bool has_replace_col = false; |
443 | 91 | for (auto col : _return_columns) { |
444 | 91 | if (tablet_schema->column(col).aggregation() == |
445 | 91 | FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE) { |
446 | 40 | has_replace_col = true; |
447 | 40 | break; |
448 | 40 | } |
449 | 91 | } |
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 | 28 | _return_columns.end()) { |
454 | 16 | _tablet_reader_params.return_columns.push_back(sequence_col_idx); |
455 | 16 | } |
456 | 40 | if (has_replace_col) { |
457 | 40 | const auto& val_to_seq = tablet_schema->value_col_idx_to_seq_col_idx(); |
458 | 40 | std::set<uint32_t> return_seq_columns; |
459 | | |
460 | 239 | 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 | 239 | const auto val_iter = val_to_seq.find(col); |
464 | 239 | 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 | 239 | } |
473 | 40 | _tablet_reader_params.return_columns.insert( |
474 | 40 | std::end(_tablet_reader_params.return_columns), |
475 | 40 | std::begin(return_seq_columns), std::end(return_seq_columns)); |
476 | 40 | } |
477 | 40 | } |
478 | 11.3k | } |
479 | | |
480 | 1.15M | _tablet_reader_params.use_page_cache = _state->enable_page_cache(); |
481 | | |
482 | 1.15M | DBUG_EXECUTE_IF("NewOlapScanner::_init_tablet_reader_params.block", DBUG_BLOCK); |
483 | | |
484 | 1.15M | if (!_state->skip_storage_engine_merge()) { |
485 | 1.15M | auto* olap_scan_local_state = (OlapScanLocalState*)_local_state; |
486 | 1.15M | TOlapScanNode& olap_scan_node = olap_scan_local_state->olap_scan_node(); |
487 | | |
488 | | // Set MOR value predicate pushdown flag |
489 | 1.15M | if (olap_scan_node.__isset.enable_mor_value_predicate_pushdown && |
490 | 1.15M | 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 | 1.15M | 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 | 1.11M | if (olap_scan_node.__isset.sort_info && |
503 | 1.11M | !olap_scan_node.sort_info.is_asc_order.empty()) { |
504 | 1.59k | _limit = _local_state->limit_per_scanner(); |
505 | 1.59k | _tablet_reader_params.read_orderby_key = true; |
506 | 1.59k | if (!olap_scan_node.sort_info.is_asc_order[0]) { |
507 | 370 | _tablet_reader_params.read_orderby_key_reverse = true; |
508 | 370 | } |
509 | 1.59k | _tablet_reader_params.read_orderby_key_num_prefix_columns = |
510 | 1.59k | olap_scan_node.sort_info.is_asc_order.size(); |
511 | 1.59k | _tablet_reader_params.read_orderby_key_limit = _limit; |
512 | | |
513 | 1.59k | if (_tablet_reader_params.read_orderby_key_limit > 0 && |
514 | 1.60k | olap_scan_local_state->_storage_no_merge()) { |
515 | 1.49k | _tablet_reader_params.filter_block_conjuncts = _conjuncts; |
516 | 1.49k | _conjuncts.clear(); |
517 | 1.49k | } |
518 | 1.11M | } 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.86k | _tablet_reader_params.general_read_limit = _limit; |
529 | 1.86k | _tablet_reader_params.filter_block_conjuncts = _conjuncts; |
530 | 1.86k | _conjuncts.clear(); |
531 | 1.86k | } |
532 | 1.11M | } |
533 | | |
534 | | // set push down topn filter |
535 | 1.15M | _tablet_reader_params.topn_filter_source_node_ids = |
536 | 1.15M | olap_scan_local_state->get_topn_filter_source_node_ids(_state, true); |
537 | 1.15M | if (!_tablet_reader_params.topn_filter_source_node_ids.empty()) { |
538 | 4.74k | _tablet_reader_params.topn_filter_target_node_id = |
539 | 4.74k | olap_scan_local_state->parent()->node_id(); |
540 | 4.74k | } |
541 | 1.15M | } |
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 | 1.15M | if (tablet_schema->field_index(BeConsts::ROWID_COL) >= 0) { |
546 | 9 | constexpr static int delayed_s = 60; |
547 | 12 | for (auto rs_reader : _tablet_reader_params.rs_splits) { |
548 | 12 | uint64_t delayed_expired_timestamp = |
549 | 12 | UnixSeconds() + _tablet_reader_params.runtime_state->execution_timeout() + |
550 | 12 | delayed_s; |
551 | 12 | rs_reader.rs_reader->rowset()->update_delayed_expired_timestamp( |
552 | 12 | delayed_expired_timestamp); |
553 | 12 | ExecEnv::GetInstance()->storage_engine().add_quering_rowset( |
554 | 12 | rs_reader.rs_reader->rowset()); |
555 | 12 | } |
556 | 9 | } |
557 | | |
558 | 1.15M | if (tablet_schema->has_global_row_id()) { |
559 | 6.62k | auto& id_file_map = _state->get_id_file_map(); |
560 | 13.7k | for (auto rs_reader : _tablet_reader_params.rs_splits) { |
561 | 13.7k | id_file_map->add_temp_rowset(rs_reader.rs_reader->rowset()); |
562 | 13.7k | } |
563 | 6.62k | } |
564 | | |
565 | 1.15M | return Status::OK(); |
566 | 1.15M | } |
567 | | |
568 | 1.15M | Status OlapScanner::_init_variant_columns() { |
569 | 1.15M | auto& tablet_schema = _tablet_reader_params.tablet_schema; |
570 | 1.15M | if (tablet_schema->num_variant_columns() == 0) { |
571 | 1.15M | return Status::OK(); |
572 | 1.15M | } |
573 | | // Parent column has path info to distinction from each other |
574 | 13.6k | for (auto* slot : _output_tuple_desc->slots()) { |
575 | 13.6k | 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 | 6.93k | const auto& dt_variant = |
579 | 6.93k | assert_cast<const DataTypeVariant&>(*remove_nullable(slot->type())); |
580 | 6.93k | TabletColumn subcol = TabletColumn::create_materialized_variant_column( |
581 | 6.93k | tablet_schema->column_by_uid(slot->col_unique_id()).name_lower_case(), |
582 | 6.93k | slot->column_paths(), slot->col_unique_id(), |
583 | 6.93k | dt_variant.variant_max_subcolumns_count(), dt_variant.enable_doc_mode()); |
584 | 6.93k | if (tablet_schema->field_index(*subcol.path_info_ptr()) < 0) { |
585 | 5.38k | tablet_schema->append_column(subcol, TabletSchema::ColumnType::VARIANT); |
586 | 5.38k | } |
587 | 6.93k | } |
588 | 13.6k | } |
589 | 5.05k | variant_util::inherit_column_attributes(tablet_schema); |
590 | 5.05k | return Status::OK(); |
591 | 1.15M | } |
592 | | |
593 | 1.15M | Status OlapScanner::_init_return_columns() { |
594 | 12.2M | for (auto* slot : _output_tuple_desc->slots()) { |
595 | | // variant column using path to index a column |
596 | 12.2M | int32_t index = 0; |
597 | 12.2M | auto& tablet_schema = _tablet_reader_params.tablet_schema; |
598 | 12.2M | if (slot->type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) { |
599 | 6.94k | index = tablet_schema->field_index(PathInData( |
600 | 6.94k | tablet_schema->column_by_uid(slot->col_unique_id()).name_lower_case(), |
601 | 6.94k | slot->column_paths())); |
602 | 12.2M | } else { |
603 | 12.3M | 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 | 12.2M | } |
606 | | |
607 | 12.2M | 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 | 12.2M | if (slot->get_virtual_column_expr()) { |
614 | 460 | ColumnId virtual_column_cid = index; |
615 | 460 | _virtual_column_exprs[virtual_column_cid] = _slot_id_to_virtual_column_expr[slot->id()]; |
616 | 460 | size_t idx_in_block = _slot_id_to_index_in_block[slot->id()]; |
617 | 460 | _vir_cid_to_idx_in_block[virtual_column_cid] = idx_in_block; |
618 | 460 | _vir_col_idx_to_type[idx_in_block] = _slot_id_to_col_type[slot->id()]; |
619 | | |
620 | 460 | VLOG_DEBUG << fmt::format( |
621 | 6 | "Virtual column, slot id: {}, cid {}, column index: {}, type: {}", slot->id(), |
622 | 6 | virtual_column_cid, _vir_cid_to_idx_in_block[virtual_column_cid], |
623 | 6 | _vir_col_idx_to_type[idx_in_block]->get_name()); |
624 | 460 | } |
625 | | |
626 | 12.2M | const auto& column = tablet_schema->column(index); |
627 | 12.2M | int32_t unique_id = column.unique_id() > 0 ? column.unique_id() : column.parent_unique_id(); |
628 | 12.2M | if (!slot->all_access_paths().empty()) { |
629 | 59.2k | _tablet_reader_params.all_access_paths.insert({unique_id, slot->all_access_paths()}); |
630 | 59.2k | } |
631 | | |
632 | 12.2M | if (!slot->predicate_access_paths().empty()) { |
633 | 5.56k | _tablet_reader_params.predicate_access_paths.insert( |
634 | 5.56k | {unique_id, slot->predicate_access_paths()}); |
635 | 5.56k | } |
636 | | |
637 | 12.2M | if ((slot->type()->get_primitive_type() == PrimitiveType::TYPE_STRUCT || |
638 | 12.3M | slot->type()->get_primitive_type() == PrimitiveType::TYPE_MAP || |
639 | 12.3M | slot->type()->get_primitive_type() == PrimitiveType::TYPE_ARRAY) && |
640 | 12.2M | !slot->all_access_paths().empty()) { |
641 | 54.0k | tablet_schema->add_pruned_columns_data_type(column.unique_id(), slot->type()); |
642 | 54.0k | } |
643 | | |
644 | 12.2M | _return_columns.push_back(index); |
645 | 12.2M | if (slot->is_nullable() && !tablet_schema->column(index).is_nullable()) { |
646 | 0 | _tablet_columns_convert_to_null_set.emplace(index); |
647 | 12.2M | } 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 | 12.2M | } |
655 | | |
656 | 1.15M | if (_return_columns.empty()) { |
657 | 0 | return Status::InternalError("failed to build storage scanner, no materialized slot!"); |
658 | 0 | } |
659 | | |
660 | 1.15M | return Status::OK(); |
661 | 1.15M | } |
662 | | |
663 | 1.16M | doris::TabletStorageType OlapScanner::get_storage_type() { |
664 | 1.16M | if (config::is_cloud_mode()) { |
665 | | // we don't have cold storage in cloud mode, all storage is treated as local |
666 | 876k | return doris::TabletStorageType::STORAGE_TYPE_LOCAL; |
667 | 876k | } |
668 | 285k | int local_reader = 0; |
669 | 501k | for (const auto& reader : _tablet_reader_params.rs_splits) { |
670 | 501k | local_reader += reader.rs_reader->rowset()->is_local(); |
671 | 501k | } |
672 | 285k | int total_reader = _tablet_reader_params.rs_splits.size(); |
673 | | |
674 | 285k | if (local_reader == total_reader) { |
675 | 285k | return doris::TabletStorageType::STORAGE_TYPE_LOCAL; |
676 | 285k | } 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 | 285k | } |
681 | | |
682 | 1.42M | 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.42M | RETURN_IF_ERROR(_tablet_reader->next_block_with_aggregation(block, eof)); |
687 | 1.42M | if (block->rows() > 0) { |
688 | 270k | _tablet_reader_params.tablet->read_block_count.fetch_add(1, std::memory_order_relaxed); |
689 | 270k | *eof = false; |
690 | 270k | } |
691 | 1.42M | return Status::OK(); |
692 | 1.42M | } |
693 | | |
694 | 1.16M | Status OlapScanner::close(RuntimeState* state) { |
695 | 1.16M | if (!_try_close()) { |
696 | 134 | return Status::OK(); |
697 | 134 | } |
698 | 1.16M | RETURN_IF_ERROR(Scanner::close(state)); |
699 | 1.16M | return Status::OK(); |
700 | 1.16M | } |
701 | | |
702 | 1.15M | void OlapScanner::update_realtime_counters() { |
703 | 1.15M | 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 | 1.15M | OlapScanLocalState* local_state = static_cast<OlapScanLocalState*>(_local_state); |
709 | 1.15M | const OlapReaderStatistics& stats = _tablet_reader->stats(); |
710 | 1.15M | COUNTER_UPDATE(local_state->_read_compressed_counter, stats.compressed_bytes_read); |
711 | 1.15M | COUNTER_UPDATE(local_state->_read_uncompressed_counter, stats.uncompressed_bytes_read); |
712 | 1.15M | COUNTER_UPDATE(local_state->_scan_bytes, stats.uncompressed_bytes_read); |
713 | 1.15M | 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 | 1.15M | _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_rows(stats.raw_rows_read); |
722 | 1.15M | _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes( |
723 | 1.15M | 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 | 1.15M | if (stats.file_cache_stats.bytes_read_from_local == 0 && |
727 | 1.15M | stats.file_cache_stats.bytes_read_from_remote == 0) { |
728 | 1.06M | _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage( |
729 | 1.06M | stats.compressed_bytes_read); |
730 | 1.06M | DorisMetrics::instance()->query_scan_bytes_from_local->increment( |
731 | 1.06M | stats.compressed_bytes_read); |
732 | 1.06M | } else { |
733 | 85.4k | _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage( |
734 | 85.4k | stats.file_cache_stats.bytes_read_from_local - _bytes_read_from_local); |
735 | 85.4k | _state->get_query_ctx() |
736 | 85.4k | ->resource_ctx() |
737 | 85.4k | ->io_context() |
738 | 85.4k | ->update_scan_bytes_from_remote_storage( |
739 | 85.4k | stats.file_cache_stats.bytes_read_from_remote - _bytes_read_from_remote); |
740 | | |
741 | 85.4k | DorisMetrics::instance()->query_scan_bytes_from_local->increment( |
742 | 85.4k | stats.file_cache_stats.bytes_read_from_local - _bytes_read_from_local); |
743 | 85.4k | DorisMetrics::instance()->query_scan_bytes_from_remote->increment( |
744 | 85.4k | stats.file_cache_stats.bytes_read_from_remote - _bytes_read_from_remote); |
745 | 85.4k | } |
746 | | |
747 | 1.15M | _tablet_reader->mutable_stats()->compressed_bytes_read = 0; |
748 | 1.15M | _tablet_reader->mutable_stats()->uncompressed_bytes_read = 0; |
749 | 1.15M | _tablet_reader->mutable_stats()->raw_rows_read = 0; |
750 | | |
751 | 1.15M | _bytes_read_from_local = _tablet_reader->stats().file_cache_stats.bytes_read_from_local; |
752 | 1.15M | _bytes_read_from_remote = _tablet_reader->stats().file_cache_stats.bytes_read_from_remote; |
753 | 1.15M | } |
754 | | |
755 | 1.15M | 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 | 1.15M | if (_has_updated_counter) { |
758 | 0 | return; |
759 | 0 | } |
760 | 1.15M | _has_updated_counter = true; |
761 | 1.15M | _tablet_reader->update_profile(_profile); |
762 | | |
763 | 1.15M | Scanner::_collect_profile_before_close(); |
764 | | |
765 | | // Update counters for OlapScanner |
766 | | // Update counters from tablet reader's stats |
767 | 1.15M | auto& stats = _tablet_reader->stats(); |
768 | 1.15M | auto* local_state = (OlapScanLocalState*)_local_state; |
769 | 1.15M | COUNTER_UPDATE(local_state->_io_timer, stats.io_ns); |
770 | 1.15M | COUNTER_UPDATE(local_state->_read_compressed_counter, stats.compressed_bytes_read); |
771 | 1.15M | COUNTER_UPDATE(local_state->_scan_bytes, stats.uncompressed_bytes_read); |
772 | 1.15M | COUNTER_UPDATE(local_state->_decompressor_timer, stats.decompress_ns); |
773 | 1.15M | COUNTER_UPDATE(local_state->_read_uncompressed_counter, stats.uncompressed_bytes_read); |
774 | 1.15M | COUNTER_UPDATE(local_state->_block_load_timer, stats.block_load_ns); |
775 | 1.15M | COUNTER_UPDATE(local_state->_block_load_counter, stats.blocks_load); |
776 | 1.15M | COUNTER_UPDATE(local_state->_block_fetch_timer, stats.block_fetch_ns); |
777 | 1.15M | COUNTER_UPDATE(local_state->_delete_bitmap_get_agg_timer, stats.delete_bitmap_get_agg_ns); |
778 | 1.15M | COUNTER_UPDATE(local_state->_scan_rows, stats.raw_rows_read); |
779 | 1.15M | COUNTER_UPDATE(local_state->_vec_cond_timer, stats.vec_cond_ns); |
780 | 1.15M | COUNTER_UPDATE(local_state->_short_cond_timer, stats.short_cond_ns); |
781 | 1.15M | COUNTER_UPDATE(local_state->_expr_filter_timer, stats.expr_filter_ns); |
782 | 1.15M | COUNTER_UPDATE(local_state->_block_init_timer, stats.block_init_ns); |
783 | 1.15M | COUNTER_UPDATE(local_state->_block_init_seek_timer, stats.block_init_seek_ns); |
784 | 1.15M | COUNTER_UPDATE(local_state->_block_init_seek_counter, stats.block_init_seek_num); |
785 | 1.15M | COUNTER_UPDATE(local_state->_segment_generate_row_range_by_keys_timer, |
786 | 1.15M | stats.generate_row_ranges_by_keys_ns); |
787 | 1.15M | COUNTER_UPDATE(local_state->_segment_generate_row_range_by_column_conditions_timer, |
788 | 1.15M | stats.generate_row_ranges_by_column_conditions_ns); |
789 | 1.15M | COUNTER_UPDATE(local_state->_segment_generate_row_range_by_bf_timer, |
790 | 1.15M | stats.generate_row_ranges_by_bf_ns); |
791 | 1.15M | COUNTER_UPDATE(local_state->_collect_iterator_merge_next_timer, |
792 | 1.15M | stats.collect_iterator_merge_next_timer); |
793 | 1.15M | COUNTER_UPDATE(local_state->_segment_generate_row_range_by_zonemap_timer, |
794 | 1.15M | stats.generate_row_ranges_by_zonemap_ns); |
795 | 1.15M | COUNTER_UPDATE(local_state->_segment_generate_row_range_by_dict_timer, |
796 | 1.15M | stats.generate_row_ranges_by_dict_ns); |
797 | 1.15M | COUNTER_UPDATE(local_state->_predicate_column_read_timer, stats.predicate_column_read_ns); |
798 | 1.15M | COUNTER_UPDATE(local_state->_non_predicate_column_read_timer, stats.non_predicate_read_ns); |
799 | 1.15M | COUNTER_UPDATE(local_state->_predicate_column_read_seek_timer, |
800 | 1.15M | stats.predicate_column_read_seek_ns); |
801 | 1.15M | COUNTER_UPDATE(local_state->_predicate_column_read_seek_counter, |
802 | 1.15M | stats.predicate_column_read_seek_num); |
803 | 1.15M | COUNTER_UPDATE(local_state->_lazy_read_timer, stats.lazy_read_ns); |
804 | 1.15M | COUNTER_UPDATE(local_state->_lazy_read_seek_timer, stats.block_lazy_read_seek_ns); |
805 | 1.15M | COUNTER_UPDATE(local_state->_lazy_read_seek_counter, stats.block_lazy_read_seek_num); |
806 | 1.15M | COUNTER_UPDATE(local_state->_output_col_timer, stats.output_col_ns); |
807 | 1.15M | COUNTER_UPDATE(local_state->_rows_vec_cond_filtered_counter, stats.rows_vec_cond_filtered); |
808 | 1.15M | COUNTER_UPDATE(local_state->_rows_short_circuit_cond_filtered_counter, |
809 | 1.15M | stats.rows_short_circuit_cond_filtered); |
810 | 1.15M | COUNTER_UPDATE(local_state->_rows_expr_cond_filtered_counter, stats.rows_expr_cond_filtered); |
811 | 1.15M | COUNTER_UPDATE(local_state->_rows_vec_cond_input_counter, stats.vec_cond_input_rows); |
812 | 1.15M | COUNTER_UPDATE(local_state->_rows_short_circuit_cond_input_counter, |
813 | 1.15M | stats.short_circuit_cond_input_rows); |
814 | 1.15M | COUNTER_UPDATE(local_state->_rows_expr_cond_input_counter, stats.expr_cond_input_rows); |
815 | 1.15M | COUNTER_UPDATE(local_state->_stats_filtered_counter, stats.rows_stats_filtered); |
816 | 1.15M | COUNTER_UPDATE(local_state->_stats_rp_filtered_counter, stats.rows_stats_rp_filtered); |
817 | 1.15M | COUNTER_UPDATE(local_state->_dict_filtered_counter, stats.segment_dict_filtered); |
818 | 1.15M | COUNTER_UPDATE(local_state->_bf_filtered_counter, stats.rows_bf_filtered); |
819 | 1.15M | COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_del_filtered); |
820 | 1.15M | COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_del_by_bitmap); |
821 | 1.15M | COUNTER_UPDATE(local_state->_del_filtered_counter, stats.rows_vec_del_cond_filtered); |
822 | 1.15M | COUNTER_UPDATE(local_state->_conditions_filtered_counter, stats.rows_conditions_filtered); |
823 | 1.15M | COUNTER_UPDATE(local_state->_key_range_filtered_counter, stats.rows_key_range_filtered); |
824 | 1.15M | COUNTER_UPDATE(local_state->_total_pages_num_counter, stats.total_pages_num); |
825 | 1.15M | COUNTER_UPDATE(local_state->_cached_pages_num_counter, stats.cached_pages_num); |
826 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_filter_counter, stats.rows_inverted_index_filtered); |
827 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_filter_timer, stats.inverted_index_filter_timer); |
828 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_query_cache_hit_counter, |
829 | 1.15M | stats.inverted_index_query_cache_hit); |
830 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_query_cache_miss_counter, |
831 | 1.15M | stats.inverted_index_query_cache_miss); |
832 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_query_timer, stats.inverted_index_query_timer); |
833 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_query_null_bitmap_timer, |
834 | 1.15M | stats.inverted_index_query_null_bitmap_timer); |
835 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_query_bitmap_copy_timer, |
836 | 1.15M | stats.inverted_index_query_bitmap_copy_timer); |
837 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_open_timer, |
838 | 1.15M | stats.inverted_index_searcher_open_timer); |
839 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_search_timer, |
840 | 1.15M | stats.inverted_index_searcher_search_timer); |
841 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_search_init_timer, |
842 | 1.15M | stats.inverted_index_searcher_search_init_timer); |
843 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_search_exec_timer, |
844 | 1.15M | stats.inverted_index_searcher_search_exec_timer); |
845 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_cache_hit_counter, |
846 | 1.15M | stats.inverted_index_searcher_cache_hit); |
847 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_searcher_cache_miss_counter, |
848 | 1.15M | stats.inverted_index_searcher_cache_miss); |
849 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_downgrade_count_counter, |
850 | 1.15M | stats.inverted_index_downgrade_count); |
851 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_analyzer_timer, |
852 | 1.15M | stats.inverted_index_analyzer_timer); |
853 | 1.15M | COUNTER_UPDATE(local_state->_inverted_index_lookup_timer, stats.inverted_index_lookup_timer); |
854 | 1.15M | COUNTER_UPDATE(local_state->_variant_scan_sparse_column_timer, |
855 | 1.15M | stats.variant_scan_sparse_column_timer_ns); |
856 | 1.15M | COUNTER_UPDATE(local_state->_variant_scan_sparse_column_bytes, |
857 | 1.15M | stats.variant_scan_sparse_column_bytes); |
858 | 1.15M | COUNTER_UPDATE(local_state->_variant_fill_path_from_sparse_column_timer, |
859 | 1.15M | stats.variant_fill_path_from_sparse_column_timer_ns); |
860 | 1.15M | COUNTER_UPDATE(local_state->_variant_subtree_default_iter_count, |
861 | 1.15M | stats.variant_subtree_default_iter_count); |
862 | 1.15M | COUNTER_UPDATE(local_state->_variant_subtree_leaf_iter_count, |
863 | 1.15M | stats.variant_subtree_leaf_iter_count); |
864 | 1.15M | COUNTER_UPDATE(local_state->_variant_subtree_hierarchical_iter_count, |
865 | 1.15M | stats.variant_subtree_hierarchical_iter_count); |
866 | 1.15M | COUNTER_UPDATE(local_state->_variant_subtree_sparse_iter_count, |
867 | 1.15M | stats.variant_subtree_sparse_iter_count); |
868 | 1.15M | COUNTER_UPDATE(local_state->_variant_doc_value_column_iter_count, |
869 | 1.15M | stats.variant_doc_value_column_iter_count); |
870 | | |
871 | 1.15M | InvertedIndexProfileReporter inverted_index_profile; |
872 | 1.15M | inverted_index_profile.update(local_state->_index_filter_profile.get(), |
873 | 1.15M | &stats.inverted_index_stats); |
874 | | |
875 | | // only cloud deploy mode will use file cache. |
876 | 1.15M | if (config::is_cloud_mode() && config::enable_file_cache) { |
877 | 870k | io::FileCacheProfileReporter cache_profile(local_state->_segment_profile.get()); |
878 | 870k | cache_profile.update(&stats.file_cache_stats); |
879 | 870k | _state->get_query_ctx()->resource_ctx()->io_context()->update_bytes_write_into_cache( |
880 | 870k | stats.file_cache_stats.bytes_write_into_cache); |
881 | 870k | } |
882 | 1.15M | COUNTER_UPDATE(local_state->_output_index_result_column_timer, |
883 | 1.15M | stats.output_index_result_column_timer); |
884 | 1.15M | COUNTER_UPDATE(local_state->_filtered_segment_counter, stats.filtered_segment_number); |
885 | 1.15M | COUNTER_UPDATE(local_state->_total_segment_counter, stats.total_segment_number); |
886 | 1.15M | COUNTER_UPDATE(local_state->_condition_cache_hit_counter, stats.condition_cache_hit_seg_nums); |
887 | 1.15M | COUNTER_UPDATE(local_state->_condition_cache_filtered_rows_counter, |
888 | 1.15M | stats.condition_cache_filtered_rows); |
889 | | |
890 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_timer, stats.tablet_reader_init_timer_ns); |
891 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_capture_rs_readers_timer, |
892 | 1.15M | stats.tablet_reader_capture_rs_readers_timer_ns); |
893 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_return_columns_timer, |
894 | 1.15M | stats.tablet_reader_init_return_columns_timer_ns); |
895 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_keys_param_timer, |
896 | 1.15M | stats.tablet_reader_init_keys_param_timer_ns); |
897 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_orderby_keys_param_timer, |
898 | 1.15M | stats.tablet_reader_init_orderby_keys_param_timer_ns); |
899 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_conditions_param_timer, |
900 | 1.15M | stats.tablet_reader_init_conditions_param_timer_ns); |
901 | 1.15M | COUNTER_UPDATE(local_state->_tablet_reader_init_delete_condition_param_timer, |
902 | 1.15M | stats.tablet_reader_init_delete_condition_param_timer_ns); |
903 | 1.15M | COUNTER_UPDATE(local_state->_block_reader_vcollect_iter_init_timer, |
904 | 1.15M | stats.block_reader_vcollect_iter_init_timer_ns); |
905 | 1.15M | COUNTER_UPDATE(local_state->_block_reader_rs_readers_init_timer, |
906 | 1.15M | stats.block_reader_rs_readers_init_timer_ns); |
907 | 1.15M | COUNTER_UPDATE(local_state->_block_reader_build_heap_init_timer, |
908 | 1.15M | stats.block_reader_build_heap_init_timer_ns); |
909 | | |
910 | 1.15M | COUNTER_UPDATE(local_state->_rowset_reader_get_segment_iterators_timer, |
911 | 1.15M | stats.rowset_reader_get_segment_iterators_timer_ns); |
912 | 1.15M | COUNTER_UPDATE(local_state->_rowset_reader_create_iterators_timer, |
913 | 1.15M | stats.rowset_reader_create_iterators_timer_ns); |
914 | 1.15M | COUNTER_UPDATE(local_state->_rowset_reader_init_iterators_timer, |
915 | 1.15M | stats.rowset_reader_init_iterators_timer_ns); |
916 | 1.15M | COUNTER_UPDATE(local_state->_rowset_reader_load_segments_timer, |
917 | 1.15M | stats.rowset_reader_load_segments_timer_ns); |
918 | | |
919 | 1.15M | COUNTER_UPDATE(local_state->_segment_iterator_init_timer, stats.segment_iterator_init_timer_ns); |
920 | 1.15M | COUNTER_UPDATE(local_state->_segment_iterator_init_return_column_iterators_timer, |
921 | 1.15M | stats.segment_iterator_init_return_column_iterators_timer_ns); |
922 | 1.15M | COUNTER_UPDATE(local_state->_segment_iterator_init_index_iterators_timer, |
923 | 1.15M | stats.segment_iterator_init_index_iterators_timer_ns); |
924 | | |
925 | 1.15M | COUNTER_UPDATE(local_state->_segment_create_column_readers_timer, |
926 | 1.15M | stats.segment_create_column_readers_timer_ns); |
927 | 1.15M | COUNTER_UPDATE(local_state->_segment_load_index_timer, stats.segment_load_index_timer_ns); |
928 | | |
929 | | // Update metrics |
930 | 1.15M | DorisMetrics::instance()->query_scan_bytes->increment( |
931 | 1.15M | local_state->_read_uncompressed_counter->value()); |
932 | 1.15M | DorisMetrics::instance()->query_scan_rows->increment(local_state->_scan_rows->value()); |
933 | 1.15M | auto& tablet = _tablet_reader_params.tablet; |
934 | 1.15M | tablet->query_scan_bytes->increment(local_state->_read_uncompressed_counter->value()); |
935 | 1.15M | tablet->query_scan_rows->increment(local_state->_scan_rows->value()); |
936 | 1.15M | tablet->query_scan_count->increment(1); |
937 | | |
938 | 1.15M | COUNTER_UPDATE(local_state->_ann_range_search_filter_counter, |
939 | 1.15M | stats.rows_ann_index_range_filtered); |
940 | 1.15M | COUNTER_UPDATE(local_state->_ann_topn_filter_counter, stats.rows_ann_index_topn_filtered); |
941 | 1.15M | COUNTER_UPDATE(local_state->_ann_index_load_costs, stats.ann_index_load_ns); |
942 | 1.15M | COUNTER_UPDATE(local_state->_ann_ivf_on_disk_load_costs, stats.ann_ivf_on_disk_load_ns); |
943 | 1.15M | COUNTER_UPDATE(local_state->_ann_ivf_on_disk_cache_hit_cnt, |
944 | 1.15M | stats.ann_ivf_on_disk_cache_hit_cnt); |
945 | | COUNTER_UPDATE(local_state->_ann_ivf_on_disk_cache_miss_cnt, |
946 | 1.15M | stats.ann_ivf_on_disk_cache_miss_cnt); |
947 | | COUNTER_UPDATE(local_state->_ann_range_search_costs, stats.ann_index_range_search_ns); |
948 | 1.15M | COUNTER_UPDATE(local_state->_ann_range_search_cnt, stats.ann_index_range_search_cnt); |
949 | 1.15M | COUNTER_UPDATE(local_state->_ann_range_engine_search_costs, stats.ann_range_engine_search_ns); |
950 | | // Engine prepare before search |
951 | 1.15M | 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 | 1.15M | COUNTER_UPDATE(local_state->_ann_range_post_process_costs, |
954 | | stats.ann_range_result_convert_ns + stats.ann_range_engine_convert_ns); |
955 | 1.15M | // Engine convert (child under post-process) |
956 | 1.15M | 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 | | COUNTER_UPDATE(local_state->_ann_range_result_convert_costs, stats.ann_range_result_convert_ns); |
959 | | |
960 | | COUNTER_UPDATE(local_state->_ann_topn_search_costs, stats.ann_topn_search_ns); |
961 | 1.15M | COUNTER_UPDATE(local_state->_ann_topn_search_cnt, stats.ann_index_topn_search_cnt); |
962 | 1.15M | |
963 | | // Detailed ANN timers |
964 | 1.15M | // ANN TopN timers with hierarchy |
965 | 1.15M | // Engine search time (FAISS) |
966 | | COUNTER_UPDATE(local_state->_ann_topn_engine_search_costs, |
967 | 1.15M | stats.ann_index_topn_engine_search_ns); |
968 | 1.15M | // Engine prepare time (allocations/buffer setup before search) |
969 | | COUNTER_UPDATE(local_state->_ann_topn_pre_process_costs, |
970 | 1.15M | stats.ann_index_topn_engine_prepare_ns); |
971 | 1.15M | // Post process parent includes Doris result processing + engine convert |
972 | | COUNTER_UPDATE(local_state->_ann_topn_post_process_costs, |
973 | | stats.ann_index_topn_result_process_ns + stats.ann_index_topn_engine_convert_ns); |
974 | 1.15M | // Engine-side conversion time inside FAISS wrappers (child under post-process) |
975 | 1.15M | COUNTER_UPDATE(local_state->_ann_topn_engine_convert_costs, |
976 | | stats.ann_index_topn_engine_convert_ns); |
977 | 1.15M | |
978 | | // Doris-side result convert costs (show separately as another child counter); use pure process time |
979 | | COUNTER_UPDATE(local_state->_ann_topn_result_convert_costs, |
980 | 1.15M | stats.ann_index_topn_result_process_ns); |
981 | | |
982 | | 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 | | } |
986 | | |
987 | | #include "common/compile_check_avoid_end.h" |
988 | | } // namespace doris |