Coverage Report

Created: 2026-07-28 01:20

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/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/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
64
    OlapTableSchemaParam() = default;
72
64
    ~OlapTableSchemaParam() noexcept = default;
73
74
    Status init(const TOlapTableSchemaParam& tschema);
75
    Status init(const POlapTableSchemaParam& pschema);
76
77
82
    int64_t db_id() const { return _db_id; }
78
82
    int64_t table_id() const { return _table_id; }
79
39
    int64_t version() const { return _version; }
80
81
29
    TupleDescriptor* tuple_desc() const { return _tuple_desc; }
82
143
    const std::vector<OlapTableIndexSchema*>& indexes() const { return _indexes; }
83
1
    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
16
    POlapTableSchemaParam* to_protobuf() const {
89
16
        if (_proto_schema == nullptr) {
90
16
            _proto_schema = _obj_pool.add(new POlapTableSchemaParam());
91
16
            to_protobuf(_proto_schema);
92
16
        }
93
16
        return _proto_schema;
94
16
    }
95
96
30
    UniqueKeyUpdateModePB unique_key_update_mode() const { return _unique_key_update_mode; }
97
98
48
    bool is_partial_update() const {
99
48
        return _unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT;
100
48
    }
101
2
    bool is_fixed_partial_update() const {
102
2
        return _unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;
103
2
    }
104
0
    bool is_flexible_partial_update() const {
105
0
        return _unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS;
106
0
    }
107
108
30
    const std::set<std::string>& partial_update_input_columns() const {
109
30
        return _partial_update_input_columns;
110
30
    }
111
30
    PartialUpdateNewRowPolicyPB partial_update_new_key_policy() const {
112
30
        return _partial_update_new_row_policy;
113
30
    }
114
30
    std::string auto_increment_coulumn() const { return _auto_increment_column; }
115
2
    int32_t auto_increment_column_unique_id() const { return _auto_increment_column_unique_id; }
116
2
    void set_timestamp_ms(int64_t timestamp_ms) { _timestamp_ms = timestamp_ms; }
117
30
    int64_t timestamp_ms() const { return _timestamp_ms; }
118
2
    void set_nano_seconds(int32_t nano_seconds) { _nano_seconds = nano_seconds; }
119
30
    int32_t nano_seconds() const { return _nano_seconds; }
120
2
    void set_timezone(std::string timezone) { _timezone = timezone; }
121
30
    std::string timezone() const { return _timezone; }
122
30
    bool is_strict_mode() const { return _is_strict_mode; }
123
29
    int32_t sequence_map_col_uid() const { return _sequence_map_col_uid; }
124
32
    int32_t row_ttl_source_column_uid() const { return _row_ttl_source_column_uid; }
125
36
    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
25
            : 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
41
            : _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
4
    int64_t db_id() const { return _t_param.db_id; }
219
4
    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
30
                                      VOlapTablePartition*& partition) const {
225
30
        auto it = _is_in_partition ? _partitions_map->find(std::tuple {block, row, true})
226
30
                                   : _partitions_map->upper_bound(std::tuple {block, row, true});
227
30
        VLOG_TRACE << "find row " << row << " of\n"
228
0
                   << block->dump_data() << "in:\n"
229
0
                   << _partition_block.dump_data() << "result line row: " << std::get<1>(it->first);
230
231
        // for list partition it might result in default partition
232
30
        if (_is_in_partition) {
233
0
            partition = (it != _partitions_map->end()) ? it->second : _default_partition;
234
0
            it = _partitions_map->end();
235
0
        }
236
30
        if (it != _partitions_map->end() &&
237
30
            _part_contains(it->second, std::tuple {block, row, true})) {
238
26
            partition = it->second;
239
26
        }
240
30
        return (partition != nullptr);
241
30
    }
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
14
            std::map<VOlapTablePartition*, int64_t>* partition_tablets_buffer = nullptr) const {
249
14
        std::function<uint32_t(Block*, uint32_t, const VOlapTablePartition&)> compute_function;
250
14
        if (!_distributed_slot_locs.empty()) {
251
            //TODO: refactor by saving the hash values. then we can calculate in columnwise.
252
11
            compute_function = [this](Block* block, uint32_t row,
253
14
                                      const VOlapTablePartition& partition) -> uint32_t {
254
14
                uint32_t hash_val = 0;
255
14
                for (unsigned short _distributed_slot_loc : _distributed_slot_locs) {
256
14
                    auto* slot_desc = _slots[_distributed_slot_loc];
257
14
                    auto& column = block->get_by_position(_distributed_slot_loc).column;
258
14
                    auto val = column->get_data_at(row);
259
14
                    if (val.data != nullptr) {
260
14
                        hash_val = RawValue::zlib_crc32(val.data, val.size,
261
14
                                                        slot_desc->type()->get_primitive_type(),
262
14
                                                        hash_val);
263
14
                    } else {
264
0
                        hash_val = HashUtil::zlib_crc_hash_null(hash_val);
265
0
                    }
266
14
                }
267
14
                return cast_set<uint32_t>(hash_val % partition.num_buckets);
268
14
            };
269
11
        } else { // random distribution
270
3
            compute_function = [](Block* block, uint32_t row,
271
3
                                  const VOlapTablePartition& partition) -> uint32_t {
272
3
                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
3
                return cast_set<uint32_t>(partition.load_tablet_idx % partition.num_buckets);
277
3
            };
278
3
        }
279
280
14
        if (partition_tablets_buffer == nullptr) {
281
14
            for (auto index : indexes) {
282
14
                tablet_indexes[index] = compute_function(block, index, *partitions[index]);
283
14
            }
284
11
        } else { // use buffer
285
6
            for (auto index : indexes) {
286
6
                auto* partition = partitions[index];
287
6
                if (auto it = partition_tablets_buffer->find(partition);
288
6
                    it != partition_tablets_buffer->end()) {
289
3
                    tablet_indexes[index] = cast_set<uint32_t>(it->second); // tablet
290
3
                } else {
291
                    // compute and save in buffer
292
3
                    (*partition_tablets_buffer)[partition] = tablet_indexes[index] =
293
3
                            compute_function(block, index, *partitions[index]);
294
3
                }
295
6
            }
296
3
        }
297
14
    }
298
299
0
    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
9
    bool is_projection_partition() const { return _is_auto_partition; }
303
19
    bool is_auto_partition() const { return _is_auto_partition; }
304
305
9
    bool is_auto_detect_overwrite() const { return _is_auto_detect_overwrite; }
306
2
    int64_t get_overwrite_group_id() const { return _overwrite_group_id; }
307
308
0
    std::shared_ptr<TNetworkAddress> get_master_address() const { return _master_address; }
309
310
2
    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
13
    VExprContextSPtrs get_part_func_ctx() { return _part_func_ctx; }
318
13
    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
2
    void set_transformed_slots(const std::vector<uint16_t>& new_slots) {
325
2
        _transformed_slot_locs = new_slots;
326
2
    }
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
10
    OlapTableLocationParam(const TOlapTableLocationParam& t_param) : _t_param(t_param) {
379
20
        for (auto& location : _t_param.tablets) {
380
20
            _tablets.emplace(location.tablet_id, &location);
381
20
        }
382
10
    }
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
1
    TabletLocation* find_tablet(int64_t tablet_id) const {
389
1
        auto it = _tablets.find(tablet_id);
390
1
        if (it != std::end(_tablets)) {
391
1
            return it->second;
392
1
        }
393
0
        return nullptr;
394
1
    }
395
396
0
    void add_locations(std::vector<TTabletLocation>& locations) {
397
0
        for (auto& location : locations) {
398
0
            if (_tablets.find(location.tablet_id) == _tablets.end()) {
399
0
                _tablets[location.tablet_id] = &location;
400
0
            }
401
0
        }
402
0
    }
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
0
    NodeInfo() = default;
417
418
    NodeInfo(const TNodeInfo& tnode)
419
2
            : id(tnode.id),
420
2
              option(tnode.option),
421
2
              host(tnode.host),
422
2
              brpc_port(tnode.async_internal_port) {}
423
};
424
425
class DorisNodesInfo {
426
public:
427
0
    DorisNodesInfo() = default;
428
2
    DorisNodesInfo(const TPaloNodesInfo& t_nodes) {
429
2
        for (const auto& node : t_nodes.nodes) {
430
2
            _nodes.emplace(node.id, node);
431
2
        }
432
2
    }
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
0
    const NodeInfo* find_node(int64_t id) const {
440
0
        auto it = _nodes.find(id);
441
0
        if (it != std::end(_nodes)) {
442
0
            return &it->second;
443
0
        }
444
0
        return nullptr;
445
0
    }
446
447
0
    void add_nodes(const std::vector<TNodeInfo>& t_nodes) {
448
0
        for (const auto& node : t_nodes) {
449
0
            const auto* node_info = find_node(node.id);
450
0
            if (node_info == nullptr) {
451
0
                _nodes.emplace(node.id, node);
452
0
            }
453
0
        }
454
0
    }
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