/root/doris/be/src/exec/tablet_info.h
Line | Count | Source (jump to first uncovered line) |
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 <butil/fast_rand.h> |
21 | | #include <gen_cpp/Descriptors_types.h> |
22 | | #include <gen_cpp/descriptors.pb.h> |
23 | | #include <gen_cpp/olap_file.pb.h> |
24 | | |
25 | | #include <cstdint> |
26 | | #include <functional> |
27 | | #include <iterator> |
28 | | #include <map> |
29 | | #include <memory> |
30 | | #include <string> |
31 | | #include <tuple> |
32 | | #include <unordered_map> |
33 | | #include <utility> |
34 | | #include <vector> |
35 | | |
36 | | #include "common/logging.h" |
37 | | #include "common/object_pool.h" |
38 | | #include "common/status.h" |
39 | | #include "runtime/descriptors.h" |
40 | | #include "runtime/raw_value.h" |
41 | | #include "vec/columns/column.h" |
42 | | #include "vec/core/block.h" |
43 | | #include "vec/core/column_with_type_and_name.h" |
44 | | #include "vec/exprs/vexpr.h" |
45 | | #include "vec/exprs/vexpr_fwd.h" |
46 | | |
47 | | namespace doris { |
48 | | class MemTracker; |
49 | | class SlotDescriptor; |
50 | | class TExprNode; |
51 | | class TabletColumn; |
52 | | class TabletIndex; |
53 | | class TupleDescriptor; |
54 | | |
55 | | struct OlapTableIndexSchema { |
56 | | int64_t index_id; |
57 | | std::vector<SlotDescriptor*> slots; |
58 | | int32_t schema_hash; |
59 | | std::vector<TabletColumn*> columns; |
60 | | std::vector<TabletIndex*> indexes; |
61 | | vectorized::VExprContextSPtr where_clause; |
62 | | |
63 | | void to_protobuf(POlapTableIndexSchema* pindex) const; |
64 | | }; |
65 | | |
66 | | class OlapTableSchemaParam { |
67 | | public: |
68 | 41 | OlapTableSchemaParam() = default; |
69 | 41 | ~OlapTableSchemaParam() noexcept = default; |
70 | | |
71 | | Status init(const TOlapTableSchemaParam& tschema); |
72 | | Status init(const POlapTableSchemaParam& pschema); |
73 | | |
74 | 25 | int64_t db_id() const { return _db_id; } |
75 | 25 | int64_t table_id() const { return _table_id; } |
76 | 0 | int64_t version() const { return _version; } |
77 | | |
78 | 0 | TupleDescriptor* tuple_desc() const { return _tuple_desc; } |
79 | 25 | const std::vector<OlapTableIndexSchema*>& indexes() const { return _indexes; } |
80 | | |
81 | | void to_protobuf(POlapTableSchemaParam* pschema) const; |
82 | | |
83 | | // NOTE: this function is not thread-safe. |
84 | 16 | POlapTableSchemaParam* to_protobuf() const { |
85 | 16 | if (_proto_schema == nullptr) { |
86 | 16 | _proto_schema = _obj_pool.add(new POlapTableSchemaParam()); |
87 | 16 | to_protobuf(_proto_schema); |
88 | 16 | } |
89 | 16 | return _proto_schema; |
90 | 16 | } |
91 | | |
92 | 25 | UniqueKeyUpdateModePB unique_key_update_mode() const { return _unique_key_update_mode; } |
93 | | |
94 | 25 | bool is_partial_update() const { |
95 | 25 | return _unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT; |
96 | 25 | } |
97 | 0 | bool is_fixed_partial_update() const { |
98 | 0 | return _unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS; |
99 | 0 | } |
100 | 0 | bool is_flexible_partial_update() const { |
101 | 0 | return _unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS; |
102 | 0 | } |
103 | | |
104 | 25 | std::set<std::string> partial_update_input_columns() const { |
105 | 25 | return _partial_update_input_columns; |
106 | 25 | } |
107 | 25 | std::string auto_increment_coulumn() const { return _auto_increment_column; } |
108 | 0 | int32_t auto_increment_column_unique_id() const { return _auto_increment_column_unique_id; } |
109 | 0 | void set_timestamp_ms(int64_t timestamp_ms) { _timestamp_ms = timestamp_ms; } |
110 | 25 | int64_t timestamp_ms() const { return _timestamp_ms; } |
111 | 0 | void set_nano_seconds(int32_t nano_seconds) { _nano_seconds = nano_seconds; } |
112 | 25 | int32_t nano_seconds() const { return _nano_seconds; } |
113 | 0 | void set_timezone(std::string timezone) { _timezone = timezone; } |
114 | 25 | std::string timezone() const { return _timezone; } |
115 | 25 | bool is_strict_mode() const { return _is_strict_mode; } |
116 | 25 | int32_t sequence_map_col_uid() const { return _sequence_map_col_uid; } |
117 | | std::string debug_string() const; |
118 | | |
119 | | Status init_unique_key_update_mode(const TOlapTableSchemaParam& tschema); |
120 | | |
121 | | private: |
122 | | int64_t _db_id; |
123 | | int64_t _table_id; |
124 | | int64_t _version; |
125 | | |
126 | | TupleDescriptor* _tuple_desc = nullptr; |
127 | | mutable POlapTableSchemaParam* _proto_schema = nullptr; |
128 | | std::vector<OlapTableIndexSchema*> _indexes; |
129 | | mutable ObjectPool _obj_pool; |
130 | | UniqueKeyUpdateModePB _unique_key_update_mode {UniqueKeyUpdateModePB::UPSERT}; |
131 | | std::set<std::string> _partial_update_input_columns; |
132 | | bool _is_strict_mode = false; |
133 | | std::string _auto_increment_column; |
134 | | int32_t _auto_increment_column_unique_id; |
135 | | int64_t _timestamp_ms = 0; |
136 | | int32_t _nano_seconds {0}; |
137 | | std::string _timezone; |
138 | | int32_t _sequence_map_col_uid {-1}; |
139 | | }; |
140 | | |
141 | | using OlapTableIndexTablets = TOlapTableIndexTablets; |
142 | | // struct TOlapTableIndexTablets { |
143 | | // 1: required i64 index_id |
144 | | // 2: required list<i64> tablets |
145 | | // } |
146 | | |
147 | | using BlockRow = std::pair<vectorized::Block*, int32_t>; |
148 | | using BlockRowWithIndicator = |
149 | | std::tuple<vectorized::Block*, int32_t, bool>; // [block, row, is_transformed] |
150 | | |
151 | | struct VOlapTablePartition { |
152 | | int64_t id = 0; |
153 | | BlockRow start_key; |
154 | | BlockRow end_key; |
155 | | std::vector<BlockRow> in_keys; |
156 | | int64_t num_buckets = 0; |
157 | | std::vector<OlapTableIndexTablets> indexes; |
158 | | bool is_mutable; |
159 | | // -1 indicates partition with hash distribution |
160 | | int64_t load_tablet_idx = -1; |
161 | | |
162 | | VOlapTablePartition(vectorized::Block* partition_block) |
163 | | // the default value of partition bound is -1. |
164 | 0 | : start_key {partition_block, -1}, end_key {partition_block, -1} {} |
165 | | }; |
166 | | |
167 | | // this is only used by tablet_sink. so we can assume it's inited by its' descriptor. |
168 | | class VOlapTablePartKeyComparator { |
169 | | public: |
170 | | VOlapTablePartKeyComparator(const std::vector<uint16_t>& slot_locs, |
171 | | const std::vector<uint16_t>& params_locs) |
172 | 0 | : _slot_locs(slot_locs), _param_locs(params_locs) {} |
173 | | |
174 | | // return true if lhs < rhs |
175 | | // 'row' is -1 mean maximal boundary |
176 | | bool operator()(const BlockRowWithIndicator& lhs, const BlockRowWithIndicator& rhs) const; |
177 | | |
178 | | private: |
179 | | const std::vector<uint16_t>& _slot_locs; |
180 | | const std::vector<uint16_t>& _param_locs; |
181 | | }; |
182 | | |
183 | | // store an olap table's tablet information |
184 | | class VOlapTablePartitionParam { |
185 | | public: |
186 | | VOlapTablePartitionParam(std::shared_ptr<OlapTableSchemaParam>& schema, |
187 | | const TOlapTablePartitionParam& param); |
188 | | |
189 | | ~VOlapTablePartitionParam(); |
190 | | |
191 | | Status init(); |
192 | | |
193 | 0 | int64_t db_id() const { return _t_param.db_id; } |
194 | 0 | int64_t table_id() const { return _t_param.table_id; } |
195 | 0 | int64_t version() const { return _t_param.version; } |
196 | | |
197 | | // return true if we found this block_row in partition |
198 | | ALWAYS_INLINE bool find_partition(vectorized::Block* block, int row, |
199 | 0 | VOlapTablePartition*& partition) const { |
200 | 0 | auto it = _is_in_partition ? _partitions_map->find(std::tuple {block, row, true}) |
201 | 0 | : _partitions_map->upper_bound(std::tuple {block, row, true}); |
202 | 0 | VLOG_TRACE << "find row " << row << " of\n" |
203 | 0 | << block->dump_data() << "in:\n" |
204 | 0 | << _partition_block.dump_data() << "result line row: " << std::get<1>(it->first); |
205 | | |
206 | | // for list partition it might result in default partition |
207 | 0 | if (_is_in_partition) { |
208 | 0 | partition = (it != _partitions_map->end()) ? it->second : _default_partition; |
209 | 0 | it = _partitions_map->end(); |
210 | 0 | } |
211 | 0 | if (it != _partitions_map->end() && |
212 | 0 | _part_contains(it->second, std::tuple {block, row, true})) { |
213 | 0 | partition = it->second; |
214 | 0 | } |
215 | 0 | return (partition != nullptr); |
216 | 0 | } |
217 | | |
218 | | ALWAYS_INLINE void find_tablets( |
219 | | vectorized::Block* block, const std::vector<uint32_t>& indexes, |
220 | | const std::vector<VOlapTablePartition*>& partitions, |
221 | | std::vector<uint32_t>& tablet_indexes /*result*/, |
222 | | /*TODO: check if flat hash map will be better*/ |
223 | 0 | std::map<VOlapTablePartition*, int64_t>* partition_tablets_buffer = nullptr) const { |
224 | 0 | std::function<uint32_t(vectorized::Block*, uint32_t, const VOlapTablePartition&)> |
225 | 0 | compute_function; |
226 | 0 | if (!_distributed_slot_locs.empty()) { |
227 | | //TODO: refactor by saving the hash values. then we can calculate in columnwise. |
228 | 0 | compute_function = [this](vectorized::Block* block, uint32_t row, |
229 | 0 | const VOlapTablePartition& partition) -> uint32_t { |
230 | 0 | uint32_t hash_val = 0; |
231 | 0 | for (unsigned short _distributed_slot_loc : _distributed_slot_locs) { |
232 | 0 | auto* slot_desc = _slots[_distributed_slot_loc]; |
233 | 0 | auto& column = block->get_by_position(_distributed_slot_loc).column; |
234 | 0 | auto val = column->get_data_at(row); |
235 | 0 | if (val.data != nullptr) { |
236 | 0 | hash_val = RawValue::zlib_crc32(val.data, val.size, slot_desc->type().type, |
237 | 0 | hash_val); |
238 | 0 | } else { |
239 | 0 | hash_val = HashUtil::zlib_crc_hash_null(hash_val); |
240 | 0 | } |
241 | 0 | } |
242 | 0 | return hash_val % partition.num_buckets; |
243 | 0 | }; |
244 | 0 | } else { // random distribution |
245 | 0 | compute_function = [](vectorized::Block* block, uint32_t row, |
246 | 0 | const VOlapTablePartition& partition) -> uint32_t { |
247 | 0 | if (partition.load_tablet_idx == -1) { |
248 | | // for compatible with old version, just do random |
249 | 0 | return butil::fast_rand() % partition.num_buckets; |
250 | 0 | } |
251 | 0 | return partition.load_tablet_idx % partition.num_buckets; |
252 | 0 | }; |
253 | 0 | } |
254 | |
|
255 | 0 | if (partition_tablets_buffer == nullptr) { |
256 | 0 | for (auto index : indexes) { |
257 | 0 | tablet_indexes[index] = compute_function(block, index, *partitions[index]); |
258 | 0 | } |
259 | 0 | } else { // use buffer |
260 | 0 | for (auto index : indexes) { |
261 | 0 | auto* partition = partitions[index]; |
262 | 0 | if (auto it = partition_tablets_buffer->find(partition); |
263 | 0 | it != partition_tablets_buffer->end()) { |
264 | 0 | tablet_indexes[index] = it->second; // tablet |
265 | 0 | } else { |
266 | | // compute and save in buffer |
267 | 0 | (*partition_tablets_buffer)[partition] = tablet_indexes[index] = |
268 | 0 | compute_function(block, index, *partitions[index]); |
269 | 0 | } |
270 | 0 | } |
271 | 0 | } |
272 | 0 | } |
273 | | |
274 | 0 | const std::vector<VOlapTablePartition*>& get_partitions() const { return _partitions; } |
275 | | |
276 | | // it's same with auto now because we only support transformed partition in auto partition. may expand in future |
277 | 0 | bool is_projection_partition() const { return _is_auto_partition; } |
278 | 0 | bool is_auto_partition() const { return _is_auto_partition; } |
279 | | |
280 | 0 | bool is_auto_detect_overwrite() const { return _is_auto_detect_overwrite; } |
281 | 0 | int64_t get_overwrite_group_id() const { return _overwrite_group_id; } |
282 | | |
283 | 0 | std::vector<uint16_t> get_partition_keys() const { return _partition_slot_locs; } |
284 | | |
285 | | Status add_partitions(const std::vector<TOlapTablePartition>& partitions); |
286 | | // no need to del/reinsert partition keys, but change the link. reset the _partitions items |
287 | | Status replace_partitions(std::vector<int64_t>& old_partition_ids, |
288 | | const std::vector<TOlapTablePartition>& new_partitions); |
289 | | |
290 | 0 | vectorized::VExprContextSPtrs get_part_func_ctx() { return _part_func_ctx; } |
291 | 0 | vectorized::VExprSPtrs get_partition_function() { return _partition_function; } |
292 | | |
293 | | // which will affect _partition_block |
294 | | Status generate_partition_from(const TOlapTablePartition& t_part, |
295 | | VOlapTablePartition*& part_result); |
296 | | |
297 | 0 | void set_transformed_slots(const std::vector<uint16_t>& new_slots) { |
298 | 0 | _transformed_slot_locs = new_slots; |
299 | 0 | } |
300 | | |
301 | | private: |
302 | | Status _create_partition_keys(const std::vector<TExprNode>& t_exprs, BlockRow* part_key); |
303 | | |
304 | | // check if this partition contain this key |
305 | | bool _part_contains(VOlapTablePartition* part, BlockRowWithIndicator key) const; |
306 | | |
307 | | // this partition only valid in this schema |
308 | | std::shared_ptr<OlapTableSchemaParam> _schema; |
309 | | TOlapTablePartitionParam _t_param; |
310 | | |
311 | | const std::vector<SlotDescriptor*>& _slots; |
312 | | std::vector<uint16_t> _partition_slot_locs; |
313 | | std::vector<uint16_t> _transformed_slot_locs; |
314 | | std::vector<uint16_t> _distributed_slot_locs; |
315 | | |
316 | | ObjectPool _obj_pool; |
317 | | vectorized::Block _partition_block; |
318 | | std::unique_ptr<MemTracker> _mem_tracker; |
319 | | std::vector<VOlapTablePartition*> _partitions; |
320 | | // For all partition value rows saved in this map, indicator is false. whenever we use a value to find in it, the param is true. |
321 | | // so that we can distinguish which column index to use (origin slots or transformed slots). |
322 | | // For range partition we ONLY SAVE RIGHT ENDS. when we find a part's RIGHT by a value, check if part's left cover it then. |
323 | | std::unique_ptr< |
324 | | std::map<BlockRowWithIndicator, VOlapTablePartition*, VOlapTablePartKeyComparator>> |
325 | | _partitions_map; |
326 | | |
327 | | bool _is_in_partition = false; |
328 | | uint32_t _mem_usage = 0; |
329 | | // only works when using list partition, the resource is owned by _partitions |
330 | | VOlapTablePartition* _default_partition = nullptr; |
331 | | |
332 | | bool _is_auto_partition = false; |
333 | | vectorized::VExprContextSPtrs _part_func_ctx = {nullptr}; |
334 | | vectorized::VExprSPtrs _partition_function = {nullptr}; |
335 | | TPartitionType::type _part_type; // support list or range |
336 | | // "insert overwrite partition(*)", detect which partitions by BE |
337 | | bool _is_auto_detect_overwrite = false; |
338 | | int64_t _overwrite_group_id = 0; |
339 | | }; |
340 | | |
341 | | // indicate where's the tablet and all its replications (node-wise) |
342 | | using TabletLocation = TTabletLocation; |
343 | | // struct TTabletLocation { |
344 | | // 1: required i64 tablet_id |
345 | | // 2: required list<i64> node_ids |
346 | | // } |
347 | | |
348 | | class OlapTableLocationParam { |
349 | | public: |
350 | 0 | OlapTableLocationParam(const TOlapTableLocationParam& t_param) : _t_param(t_param) { |
351 | 0 | for (auto& location : _t_param.tablets) { |
352 | 0 | _tablets.emplace(location.tablet_id, &location); |
353 | 0 | } |
354 | 0 | } |
355 | | |
356 | 0 | int64_t db_id() const { return _t_param.db_id; } |
357 | 0 | int64_t table_id() const { return _t_param.table_id; } |
358 | 0 | int64_t version() const { return _t_param.version; } |
359 | | |
360 | 0 | TabletLocation* find_tablet(int64_t tablet_id) const { |
361 | 0 | auto it = _tablets.find(tablet_id); |
362 | 0 | if (it != std::end(_tablets)) { |
363 | 0 | return it->second; |
364 | 0 | } |
365 | 0 | return nullptr; |
366 | 0 | } |
367 | | |
368 | 0 | void add_locations(std::vector<TTabletLocation>& locations) { |
369 | 0 | for (auto& location : locations) { |
370 | 0 | if (_tablets.find(location.tablet_id) == _tablets.end()) { |
371 | 0 | _tablets[location.tablet_id] = &location; |
372 | 0 | } |
373 | 0 | } |
374 | 0 | } |
375 | | |
376 | | private: |
377 | | TOlapTableLocationParam _t_param; |
378 | | // [tablet_id, tablet]. tablet has id, also. |
379 | | std::unordered_map<int64_t, TabletLocation*> _tablets; |
380 | | }; |
381 | | |
382 | | struct NodeInfo { |
383 | | int64_t id; |
384 | | int64_t option; |
385 | | std::string host; |
386 | | int32_t brpc_port; |
387 | | |
388 | 0 | NodeInfo() = default; |
389 | | |
390 | | NodeInfo(const TNodeInfo& tnode) |
391 | | : id(tnode.id), |
392 | | option(tnode.option), |
393 | | host(tnode.host), |
394 | 0 | brpc_port(tnode.async_internal_port) {} |
395 | | }; |
396 | | |
397 | | class DorisNodesInfo { |
398 | | public: |
399 | 0 | DorisNodesInfo() = default; |
400 | 0 | DorisNodesInfo(const TPaloNodesInfo& t_nodes) { |
401 | 0 | for (const auto& node : t_nodes.nodes) { |
402 | 0 | _nodes.emplace(node.id, node); |
403 | 0 | } |
404 | 0 | } |
405 | 0 | void setNodes(const TPaloNodesInfo& t_nodes) { |
406 | 0 | _nodes.clear(); |
407 | 0 | for (const auto& node : t_nodes.nodes) { |
408 | 0 | _nodes.emplace(node.id, node); |
409 | 0 | } |
410 | 0 | } |
411 | 0 | const NodeInfo* find_node(int64_t id) const { |
412 | 0 | auto it = _nodes.find(id); |
413 | 0 | if (it != std::end(_nodes)) { |
414 | 0 | return &it->second; |
415 | 0 | } |
416 | 0 | return nullptr; |
417 | 0 | } |
418 | | |
419 | 0 | void add_nodes(const std::vector<TNodeInfo>& t_nodes) { |
420 | 0 | for (const auto& node : t_nodes) { |
421 | 0 | const auto* node_info = find_node(node.id); |
422 | 0 | if (node_info == nullptr) { |
423 | 0 | _nodes.emplace(node.id, node); |
424 | 0 | } |
425 | 0 | } |
426 | 0 | } |
427 | | |
428 | 0 | const std::unordered_map<int64_t, NodeInfo>& nodes_info() { return _nodes; } |
429 | | |
430 | | private: |
431 | | std::unordered_map<int64_t, NodeInfo> _nodes; |
432 | | }; |
433 | | |
434 | | } // namespace doris |