be/src/storage/tablet/tablet_reader.h
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 | | #pragma once |
19 | | |
20 | | #include <gen_cpp/Descriptors_types.h> |
21 | | #include <gen_cpp/PaloInternalService_types.h> |
22 | | #include <gen_cpp/PlanNodes_types.h> |
23 | | #include <stddef.h> |
24 | | #include <stdint.h> |
25 | | |
26 | | #include <memory> |
27 | | #include <optional> |
28 | | #include <set> |
29 | | #include <string> |
30 | | #include <unordered_set> |
31 | | #include <utility> |
32 | | #include <vector> |
33 | | |
34 | | #include "agent/be_exec_version_manager.h" |
35 | | #include "common/status.h" |
36 | | #include "io/io_common.h" |
37 | | #include "storage/delete/delete_handler.h" |
38 | | #include "storage/iterators.h" |
39 | | #include "storage/olap_common.h" |
40 | | #include "storage/olap_tuple.h" |
41 | | #include "storage/predicate/filter_olap_param.h" |
42 | | #include "storage/row_cursor.h" |
43 | | #include "storage/rowid_conversion.h" |
44 | | #include "storage/rowset/rowset.h" |
45 | | #include "storage/rowset/rowset_meta.h" |
46 | | #include "storage/rowset/rowset_reader.h" |
47 | | #include "storage/rowset/rowset_reader_context.h" |
48 | | #include "storage/segment/variant/variant_compaction_paths.h" |
49 | | #include "storage/tablet/base_tablet.h" |
50 | | #include "storage/tablet/tablet_fwd.h" |
51 | | |
52 | | namespace doris { |
53 | | |
54 | | class RuntimeState; |
55 | | class BloomFilterFuncBase; |
56 | | class ColumnPredicate; |
57 | | class DeleteBitmap; |
58 | | class HybridSetBase; |
59 | | class RuntimeProfile; |
60 | | |
61 | | class VCollectIterator; |
62 | | class Block; |
63 | | class VExpr; |
64 | | class VExprContext; |
65 | | |
66 | | // Used to compare row with input scan key. Scan key only contains key columns, |
67 | | // row contains all key columns, which is superset of key columns. |
68 | | // So we should compare the common prefix columns of lhs and rhs. |
69 | | // |
70 | | // NOTE: if you are not sure if you can use it, please don't use this function. |
71 | 1.33M | inline int compare_row_key(const RowCursor& lhs, const RowCursor& rhs) { |
72 | 1.33M | auto cmp_cids = std::min(lhs.field_count(), rhs.field_count()); |
73 | 2.61M | for (uint32_t cid = 0; cid < cmp_cids; ++cid) { |
74 | 2.58M | const auto& lf = lhs.field(cid); |
75 | 2.58M | const auto& rf = rhs.field(cid); |
76 | | // Handle nulls: null < non-null |
77 | 2.58M | if (lf.is_null() != rf.is_null()) { |
78 | 18.4E | return lf.is_null() ? -1 : 1; |
79 | 179k | } |
80 | 2.40M | if (lf.is_null()) { |
81 | 3.86k | continue; // both null |
82 | 3.86k | } |
83 | 2.39M | auto cmp = lf <=> rf; |
84 | 2.39M | if (cmp < 0) return -1; |
85 | 1.28M | if (cmp > 0) return 1; |
86 | 1.28M | } |
87 | 33.1k | return 0; |
88 | 1.33M | } |
89 | | |
90 | | class TabletReader { |
91 | | struct KeysParam { |
92 | | std::vector<std::optional<RowCursor>> start_keys; |
93 | | std::vector<std::optional<RowCursor>> end_keys; |
94 | | bool start_key_include = false; |
95 | | bool end_key_include = false; |
96 | | }; |
97 | | |
98 | | public: |
99 | | // Params for Reader, |
100 | | // mainly include tablet, data version and fetch range. |
101 | | struct ReaderParams { |
102 | 901k | bool has_single_version() const { |
103 | 901k | return (rs_splits.size() == 1 && |
104 | 901k | rs_splits[0].rs_reader->rowset()->start_version() == 0 && |
105 | 901k | !rs_splits[0].rs_reader->rowset()->rowset_meta()->is_segments_overlapping()) || |
106 | 901k | (rs_splits.size() == 2 && |
107 | 894k | rs_splits[0].rs_reader->rowset()->rowset_meta()->num_rows() == 0 && |
108 | 894k | rs_splits[1].rs_reader->rowset()->start_version() == 2 && |
109 | 894k | !rs_splits[1].rs_reader->rowset()->rowset_meta()->is_segments_overlapping()); |
110 | 901k | } |
111 | | |
112 | 14.3k | int get_be_exec_version() const { |
113 | 14.3k | if (runtime_state) { |
114 | 10.4k | return runtime_state->be_exec_version(); |
115 | 10.4k | } |
116 | 3.91k | return BeExecVersionManager::get_newest_version(); |
117 | 14.3k | } |
118 | | |
119 | 934k | void set_read_source(TabletReadSource read_source, bool skip_delete_bitmap = false) { |
120 | 934k | rs_splits = std::move(read_source.rs_splits); |
121 | 934k | delete_predicates = std::move(read_source.delete_predicates); |
122 | 934k | #ifndef BE_TEST |
123 | 935k | if (!skip_delete_bitmap && tablet->need_read_delete_bitmap()) { |
124 | 500k | delete_bitmap = std::move(read_source.delete_bitmap); |
125 | 500k | } |
126 | 934k | #endif |
127 | 934k | } |
128 | | |
129 | | BaseTabletSPtr tablet; |
130 | | TabletSchemaSPtr tablet_schema; |
131 | | // Set only by compaction, alongside its extended tablet_schema. |
132 | | VariantCompactionPathsSPtr variant_compaction_paths = nullptr; |
133 | | ReaderType reader_type = ReaderType::READER_QUERY; |
134 | | bool read_row_binlog = false; |
135 | | bool direct_mode = false; |
136 | | bool aggregation = false; |
137 | | // for compaction, schema_change, check_sum: we don't use page cache |
138 | | // for query, when the BE config disable_storage_page_cache is false, we use page cache |
139 | | bool use_page_cache = false; |
140 | | Version version = Version(-1, 0); |
141 | | |
142 | | // The vectors are range-aligned; nullopt represents an unbounded endpoint. |
143 | | std::vector<std::optional<OlapTuple>> start_key; |
144 | | std::vector<std::optional<OlapTuple>> end_key; |
145 | | bool start_key_include = false; |
146 | | bool end_key_include = false; |
147 | | |
148 | | std::vector<std::shared_ptr<ColumnPredicate>> predicates; |
149 | | std::vector<RowsetMetaSharedPtr> delete_predicates; |
150 | | // slots that cast may be eliminated in storage layer |
151 | | std::map<std::string, DataTypePtr> target_cast_type_for_variants; |
152 | | |
153 | | std::map<int32_t, TColumnAccessPaths> all_access_paths; |
154 | | std::map<int32_t, TColumnAccessPaths> predicate_access_paths; |
155 | | |
156 | | std::vector<RowSetSplits> rs_splits; |
157 | | // For unique key table with merge-on-write |
158 | | DeleteBitmapPtr delete_bitmap = nullptr; |
159 | | |
160 | | // The read schema of every Block at the TabletReader boundary. |
161 | | // Query scanners construct the FE-slot prefix in scan-tuple order; |
162 | | // TabletReader may append storage-only delete-predicate columns. |
163 | | ReadSchemaSPtr read_schema; |
164 | | // output_columns only contain columns in OrderByExprs and outputExprs |
165 | | // (column unique ids, not ordinals) |
166 | | std::set<int32_t> output_columns; |
167 | | // Ordinals (positions in read_schema) of extra storage key columns |
168 | | // that are present only for scan-schema alignment. |
169 | | // Example: for AGG keys (k1, k2), a query that returns k2 can scan |
170 | | // (k1, k2) and project away k1. Direct readers may avoid reading such |
171 | | // columns only if the lower iterator proves their real values are not |
172 | | // required by predicates, delete conditions, or expressions. |
173 | | std::set<ColumnId> extra_columns; |
174 | | RuntimeProfile* profile = nullptr; |
175 | | RuntimeState* runtime_state = nullptr; |
176 | | |
177 | | TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE; |
178 | | VExprContextSPtrs common_expr_ctxs_push_down; |
179 | | |
180 | | // used for compaction to record row ids |
181 | | bool record_rowids = false; |
182 | | RowIdConversion* rowid_conversion = nullptr; |
183 | | std::vector<int> topn_filter_source_node_ids; |
184 | | // used for special optimization for query : ORDER BY key LIMIT n |
185 | | bool read_orderby_key = false; |
186 | | // used for special optimization for query : ORDER BY key DESC LIMIT n |
187 | | bool read_orderby_key_reverse = false; |
188 | | // For rows with the same key, use ascending order (small-to-large) for tie-breakers. |
189 | | // For example, use lower rowset version / segment id first. |
190 | | bool use_insert_order_when_same = false; |
191 | | // Force globally key-ordered reading for row-binlog scans (e.g. so MIN_DELTA can |
192 | | // group consecutive same-key changes across segments). Overlapping segments use the |
193 | | // merge iterator; segments proven non-overlapping use an ordered union. |
194 | | // See BetaRowsetReader::is_merge_iterator() in beta_rowset_reader.h:62. |
195 | | bool force_key_ordered_read = false; |
196 | | // num of columns for orderby key |
197 | | size_t read_orderby_key_num_prefix_columns = 0; |
198 | | // limit of rows for read_orderby_key |
199 | | size_t read_orderby_key_limit = 0; |
200 | | // for vertical compaction |
201 | | bool is_key_column_group = false; |
202 | | std::vector<uint32_t> key_group_cluster_key_idxes; |
203 | | |
204 | | // For sparse column compaction optimization |
205 | | // When true, use optimized path for sparse wide tables |
206 | | bool enable_sparse_optimization = false; |
207 | | |
208 | | bool is_segcompaction = false; |
209 | | |
210 | | // Enable value predicate pushdown for MOR tables |
211 | | bool enable_mor_value_predicate_pushdown = false; |
212 | | |
213 | | std::vector<RowwiseIteratorUPtr>* segment_iters_ptr = nullptr; |
214 | | |
215 | | void check_validation() const; |
216 | | |
217 | | int64_t batch_size = -1; |
218 | | |
219 | | // virtual column ordinal (position in read_schema) -> expression |
220 | | std::map<ColumnId, VExprContextSPtr> virtual_column_exprs; |
221 | | |
222 | | std::shared_ptr<ScoreRuntime> score_runtime; |
223 | | CollectionStatisticsPtr collection_statistics; |
224 | | std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime; |
225 | | |
226 | | uint64_t condition_cache_digest = 0; |
227 | | |
228 | | // General LIMIT budget forwarded to SegmentIterator. -1 means no limit. |
229 | | int64_t general_read_limit = -1; |
230 | | TBinlogScanType::type binlog_scan_type = TBinlogScanType::NONE; |
231 | | }; |
232 | | |
233 | 933k | TabletReader() = default; |
234 | | |
235 | 932k | virtual ~TabletReader() = default; |
236 | | |
237 | | TabletReader(const TabletReader&) = delete; |
238 | | void operator=(const TabletReader&) = delete; |
239 | | |
240 | | // Initialize TabletReader with tablet, data version and fetch range. |
241 | | virtual Status init(const ReaderParams& read_params); |
242 | | |
243 | | // Read next block with aggregation. |
244 | | // Return OK and set `*eof` to false when next block is read |
245 | | // Return OK and set `*eof` to true when no more rows can be read. |
246 | | // Return others when unexpected error happens. |
247 | 0 | virtual Status next_block_with_aggregation(Block* block, bool* eof) { |
248 | 0 | return Status::Error<ErrorCode::READER_INITIALIZE_ERROR>( |
249 | 0 | "TabletReader not support next_block_with_aggregation"); |
250 | 0 | } |
251 | | |
252 | 1.36k | virtual uint64_t merged_rows() const { return _merged_rows; } |
253 | | |
254 | 11.9k | uint64_t filtered_rows() const { |
255 | 11.9k | return _stats.rows_del_filtered + _stats.rows_del_by_bitmap + |
256 | 11.9k | _stats.rows_conditions_filtered + _stats.rows_vec_del_cond_filtered + |
257 | 11.9k | _stats.rows_vec_cond_filtered + _stats.rows_short_circuit_cond_filtered; |
258 | 11.9k | } |
259 | | |
260 | 902k | void set_batch_size(int batch_size) { _reader_context.batch_size = batch_size; } |
261 | | |
262 | 10.3M | size_t batch_max_rows() const { return _reader_context.batch_size; } |
263 | | |
264 | 902k | void set_preferred_block_size_bytes(size_t bytes) { |
265 | 902k | _reader_context.preferred_block_size_bytes = bytes; |
266 | 902k | } |
267 | | |
268 | | // Returns the preferred output block byte budget. Subclasses that support adaptive batch size |
269 | | // should override this; the base returns 0 (disabled) so VCollectIterator degrades safely |
270 | | // when called through a TabletReader* that has not been configured. |
271 | 0 | virtual size_t preferred_block_size_bytes() const { return 0; } |
272 | | |
273 | 3.17M | const OlapReaderStatistics& stats() const { return _stats; } |
274 | 4.68M | OlapReaderStatistics* mutable_stats() { return &_stats; } |
275 | | |
276 | 2 | virtual void update_profile(RuntimeProfile* profile) {} |
277 | | |
278 | | // Remove the delete-condition columns from `all_access_paths` so they fall back to a full |
279 | | // read (a meta-only read would make the storage delete predicate match nothing and leak |
280 | | // deleted rows). |
281 | | static void remove_delete_columns_from_access_paths( |
282 | | const DeleteHandler& delete_handler, const ReadSchema& read_schema, |
283 | | std::map<int32_t, TColumnAccessPaths>& all_access_paths); |
284 | | |
285 | | protected: |
286 | | friend class VCollectIterator; |
287 | | |
288 | | Status _init_params(const ReaderParams& read_params); |
289 | | |
290 | | Status _capture_rs_readers(const ReaderParams& read_params); |
291 | | |
292 | | Status _init_keys_param(const ReaderParams& read_params); |
293 | | |
294 | | Status _init_orderby_keys_param(const ReaderParams& read_params); |
295 | | |
296 | | Status _init_column_predicates(const ReaderParams& read_params); |
297 | | |
298 | | Status _init_delete_condition(const ReaderParams& read_params); |
299 | | |
300 | 1.85M | const BaseTabletSPtr& tablet() { return _tablet; } |
301 | | |
302 | | // The read schema shared by the caller, TabletReader, VCollect and each rowset reader. |
303 | | ReadSchemaSPtr _read_schema; |
304 | | |
305 | | // used for special optimization for query : ORDER BY key [ASC|DESC] LIMIT n |
306 | | // columns for orderby keys |
307 | | std::vector<uint32_t> _orderby_key_columns; |
308 | | BaseTabletSPtr _tablet; |
309 | | RowsetReaderContext _reader_context; |
310 | | TabletSchemaSPtr _tablet_schema; |
311 | | KeysParam _keys_param; |
312 | | std::vector<bool> _is_lower_keys_included; |
313 | | std::vector<bool> _is_upper_keys_included; |
314 | | std::vector<std::shared_ptr<ColumnPredicate>> _col_predicates; |
315 | | std::vector<std::shared_ptr<ColumnPredicate>> _value_col_predicates; |
316 | | DeleteHandler _delete_handler; |
317 | | |
318 | | // Indicates whether the tablets has do a aggregation in storage engine. |
319 | | bool _aggregation = false; |
320 | | // for agg query, we don't need to finalize when scan agg object data |
321 | | ReaderType _reader_type = ReaderType::READER_QUERY; |
322 | | bool _delete_sign_available = false; |
323 | | bool _filter_delete = false; |
324 | | bool _direct_mode = false; |
325 | | |
326 | | uint64_t _merged_rows = 0; |
327 | | OlapReaderStatistics _stats; |
328 | | }; |
329 | | |
330 | | } // namespace doris |