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