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