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