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