Coverage Report

Created: 2026-06-02 20:24

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/tablet/tablet.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/macros.h>
21
#include <glog/logging.h>
22
23
#include <array>
24
#include <atomic>
25
#include <cstddef>
26
#include <cstdint>
27
#include <functional>
28
#include <memory>
29
#include <mutex>
30
#include <ostream>
31
#include <set>
32
#include <shared_mutex>
33
#include <string>
34
#include <string_view>
35
#include <utility>
36
#include <vector>
37
38
#include "common/atomic_shared_ptr.h"
39
#include "common/config.h"
40
#include "common/metrics/metrics.h"
41
#include "common/status.h"
42
#include "gen_cpp/AgentService_types.h"
43
#include "storage/binlog.h"
44
#include "storage/binlog_config.h"
45
#include "storage/compaction/binlog_compaction_policy.h"
46
#include "storage/data_dir.h"
47
#include "storage/olap_common.h"
48
#include "storage/partial_update_info.h"
49
#include "storage/rowset/rowset.h"
50
#include "storage/rowset/rowset_meta.h"
51
#include "storage/rowset/rowset_reader.h"
52
#include "storage/segment/segment.h"
53
#include "storage/segment/segment_loader.h"
54
#include "storage/tablet/base_tablet.h"
55
#include "storage/version_graph.h"
56
#include "util/once.h"
57
#include "util/slice.h"
58
59
namespace bvar {
60
template <typename T>
61
class Adder;
62
}
63
64
namespace doris {
65
66
class Tablet;
67
class CumulativeCompactionPolicy;
68
class CompactionMixin;
69
class SingleReplicaCompaction;
70
class RowsetWriter;
71
struct RowsetWriterContext;
72
class TTabletInfo;
73
class TabletMetaPB;
74
class TupleDescriptor;
75
class CalcDeleteBitmapToken;
76
enum CompressKind : int;
77
class RowsetBinlogMetasPB;
78
79
namespace io {
80
class RemoteFileSystem;
81
} // namespace io
82
class Block;
83
struct RowLocation;
84
enum KeysType : int;
85
enum SortType : int;
86
87
enum TabletStorageType { STORAGE_TYPE_LOCAL, STORAGE_TYPE_REMOTE, STORAGE_TYPE_REMOTE_AND_LOCAL };
88
89
extern bvar::Adder<uint64_t> unused_remote_rowset_num;
90
91
static inline constexpr auto TRACE_TABLET_LOCK_THRESHOLD = std::chrono::seconds(1);
92
93
struct WriteCooldownMetaExecutors {
94
    WriteCooldownMetaExecutors(size_t executor_nums = 5);
95
96
    void stop();
97
98
    void submit(TabletSharedPtr tablet);
99
124k
    size_t _get_executor_pos(int64_t tablet_id) const {
100
124k
        return std::hash<int64_t>()(tablet_id) % _executor_nums;
101
124k
    };
102
    // Each executor is a mpsc to ensure uploads of the same tablet meta are not concurrent
103
    // FIXME(AlexYue): Use mpsc instead of `ThreadPool` with 1 thread
104
    // We use PriorityThreadPool since it would call status inside it's `shutdown` function.
105
    // Consider one situation where the StackTraceCache's singleton is detructed before
106
    // this WriteCooldownMetaExecutors's singleton, then invoking the status would also call
107
    // StackTraceCache which would then result in heap use after free like #23834
108
    std::vector<std::unique_ptr<PriorityThreadPool>> _executors;
109
    std::unordered_set<int64_t> _pending_tablets;
110
    std::mutex _latch;
111
    size_t _executor_nums;
112
};
113
114
class Tablet final : public BaseTablet {
115
public:
116
    Tablet(StorageEngine& engine, TabletMetaSharedPtr tablet_meta, DataDir* data_dir,
117
           const std::string_view& cumulative_compaction_type = "");
118
119
32.0M
    DataDir* data_dir() const { return _data_dir; }
120
1.27M
    int64_t replica_id() const { return _tablet_meta->replica_id(); }
121
122
515k
    std::string tablet_path() const override { return _tablet_path; }
123
83
    std::string row_binlog_path() const {
124
83
        return fmt::format("{}/{}", _tablet_path, FDRowBinlogSuffix);
125
83
    }
126
485k
    std::string get_rowset_path(const RowsetMetaSharedPtr& rowset_meta) const {
127
485k
        if (!rowset_meta->is_local()) {
128
18
            return "";
129
18
        }
130
485k
        return rowset_meta->is_row_binlog() ? row_binlog_path() : tablet_path();
131
485k
    }
132
133
    bool set_tablet_schema_into_rowset_meta();
134
    Status init();
135
    bool init_succeeded();
136
137
    bool is_used();
138
139
    void register_tablet_into_dir();
140
    void deregister_tablet_from_dir();
141
142
    void save_meta();
143
    // Used in clone task, to update local meta when finishing a clone job
144
    Status revise_tablet_meta(const std::vector<RowsetSharedPtr>& to_add,
145
                              const std::vector<RowsetSharedPtr>& to_delete,
146
                              bool is_incremental_clone, bool copy_row_binlog = false);
147
148
    int64_t cumulative_layer_point() const;
149
    void set_cumulative_layer_point(int64_t new_point);
150
    inline int64_t cumulative_promotion_size() const;
151
    inline void set_cumulative_promotion_size(int64_t new_size);
152
153
    // Disk space occupied by tablet, contain local and remote.
154
    size_t tablet_footprint() override;
155
    // Local disk space occupied by tablet.
156
    size_t tablet_local_size();
157
    // Remote disk space occupied by tablet.
158
    size_t tablet_remote_size();
159
160
    size_t num_rows();
161
    size_t version_count() const;
162
    size_t stale_version_count() const;
163
    bool exceed_version_limit(int32_t limit) override;
164
    uint64_t segment_count() const;
165
    Version max_version() const;
166
    CumulativeCompactionPolicy* cumulative_compaction_policy();
167
    BinlogCompactionPolicy* binlog_compaction_policy();
168
169
    // properties encapsulated in TabletSchema
170
    SortType sort_type() const;
171
    size_t sort_col_num() const;
172
    size_t num_columns() const;
173
    size_t num_null_columns() const;
174
    size_t num_short_key_columns() const;
175
    size_t num_rows_per_row_block() const;
176
    double bloom_filter_fpp() const;
177
    size_t next_unique_id() const;
178
    int64_t avg_rs_meta_serialize_size() const;
179
180
    // operation in rowsets
181
    Status add_rowset(RowsetSharedPtr rowset, RowsetSharedPtr row_binlog_rowset = nullptr);
182
    Status create_initial_rowset(const int64_t version);
183
184
    // MUST hold EXCLUSIVE `_meta_lock`.
185
    Status modify_rowsets(std::vector<RowsetSharedPtr>& to_add,
186
                          std::vector<RowsetSharedPtr>& to_delete, bool check_delete = false);
187
    Status modify_row_binlog_rowsets(std::vector<RowsetSharedPtr>& to_add,
188
                                     std::vector<RowsetSharedPtr>& to_delete);
189
    bool rowset_exists_unlocked(const RowsetSharedPtr& rowset);
190
191
    // Add a committed data rowset and its row binlog rowset
192
    Status add_inc_rowset(const RowsetSharedPtr& rowset,
193
                          const RowsetSharedPtr& row_binlog_rowset = nullptr);
194
    /// Delete stale rowset by timing. This delete policy uses now() minutes
195
    /// config::tablet_rowset_expired_stale_sweep_time_sec to compute the deadline of expired rowset
196
    /// to delete.  When rowset is deleted, it will be added to StorageEngine unused map and record
197
    /// need to delete flag.
198
    void delete_expired_stale_rowset();
199
200
    // if quiet is true, no error log will be printed if there are missing versions
201
    Status check_version_integrity(const Version& version, bool quiet = false);
202
    bool check_version_exist(const Version& version) const;
203
    void acquire_version_and_rowsets(
204
            std::vector<std::pair<Version, RowsetSharedPtr>>* version_rowsets) const;
205
206
    // If skip_missing_version is true, skip versions if they are missing.
207
    Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
208
                              const CaptureRowsetOps& opts) override;
209
210
    // Find the missed versions until the spec_version.
211
    //
212
    // for example:
213
    //     [0-4][5-5][8-8][9-9][14-14]
214
    // if spec_version = 12, it will return [6, 6], [7, 7], [10, 10], [11, 11], [12, 12]
215
    Versions calc_missed_versions(int64_t spec_version, Versions existing_versions) const override;
216
217
    // meta lock
218
167k
    std::shared_mutex& get_header_lock() { return _meta_lock; }
219
39.1k
    std::mutex& get_rowset_update_lock() { return _rowset_update_lock; }
220
37.1k
    std::mutex& get_push_lock() { return _ingest_lock; }
221
6.31M
    std::mutex& get_base_compaction_lock() { return _base_compaction_lock; }
222
8.13M
    std::mutex& get_cumulative_compaction_lock() { return _cumulative_compaction_lock; }
223
996
    std::mutex& get_binlog_compaction_lock() { return _binlog_compaction_lock; }
224
173
    std::shared_mutex& get_meta_store_lock() { return _meta_store_lock; }
225
226
74.1k
    std::shared_timed_mutex& get_migration_lock() { return _migration_lock; }
227
228
194
    std::mutex& get_build_inverted_index_lock() { return _build_inverted_index_lock; }
229
230
    // operation for compaction
231
    bool can_do_compaction(size_t path_hash, CompactionType compaction_type);
232
    bool suitable_for_compaction(
233
            CompactionType compaction_type,
234
            std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy);
235
236
    uint32_t calc_compaction_score(CompactionType compaction_type,
237
                                   int8_t* prefer_compaction_level = nullptr);
238
239
    // This function to find max continuous version from the beginning.
240
    // For example: If there are 1, 2, 3, 5, 6, 7 versions belongs tablet, then 3 is target.
241
    // 3 will be saved in "version", and 7 will be saved in "max_version", if max_version != nullptr
242
    void max_continuous_version_from_beginning(Version* version, Version* max_version = nullptr);
243
244
0
    void set_bad(bool is_bad) { _is_bad = is_bad; }
245
246
15.0M
    int64_t last_cumu_compaction_failure_time() { return _last_cumu_compaction_failure_millis; }
247
178k
    void set_last_cumu_compaction_failure_time(int64_t millis) {
248
178k
        _last_cumu_compaction_failure_millis = millis;
249
178k
    }
250
251
6.26M
    int64_t last_base_compaction_failure_time() { return _last_base_compaction_failure_millis; }
252
100k
    void set_last_base_compaction_failure_time(int64_t millis) {
253
100k
        _last_base_compaction_failure_millis = millis;
254
100k
    }
255
256
0
    int64_t last_full_compaction_failure_time() { return _last_full_compaction_failure_millis; }
257
0
    void set_last_full_compaction_failure_time(int64_t millis) {
258
0
        _last_full_compaction_failure_millis = millis;
259
0
    }
260
261
172k
    int64_t last_cumu_compaction_success_time() { return _last_cumu_compaction_success_millis; }
262
8.68k
    void set_last_cumu_compaction_success_time(int64_t millis) {
263
8.68k
        _last_cumu_compaction_success_millis = millis;
264
8.68k
    }
265
266
172k
    int64_t last_base_compaction_success_time() { return _last_base_compaction_success_millis; }
267
5.19k
    void set_last_base_compaction_success_time(int64_t millis) {
268
5.19k
        _last_base_compaction_success_millis = millis;
269
5.19k
    }
270
271
0
    int64_t last_full_compaction_success_time() { return _last_full_compaction_success_millis; }
272
0
    void set_last_full_compaction_success_time(int64_t millis) {
273
0
        _last_full_compaction_success_millis = millis;
274
0
    }
275
276
492
    int64_t last_binlog_compaction_success_time(int8_t compaction_level) {
277
492
        DCHECK(compaction_level >= 0 &&
278
492
               compaction_level < BinlogCompactionPolicy::kBinlogCompactionMaxLevel);
279
492
        return _last_binlog_compaction_success_millis[compaction_level].load();
280
492
    }
281
504
    int64_t last_binlog_compaction_failure_time() {
282
504
        return _last_binlog_compaction_failure_millis.load();
283
504
    }
284
    void init_last_binlog_compaction_success_and_failure_time();
285
0
    void set_last_binlog_compaction_success_time(int8_t compaction_level, int64_t millis) {
286
0
        DCHECK(compaction_level >= 0 &&
287
0
               compaction_level < BinlogCompactionPolicy::kBinlogCompactionMaxLevel);
288
0
        _last_binlog_compaction_success_millis[compaction_level] = millis;
289
0
    }
290
492
    void set_last_binlog_compaction_failure_time(int64_t millis) {
291
492
        _last_binlog_compaction_failure_millis = millis;
292
492
    }
293
294
0
    int64_t last_cumu_compaction_schedule_time() { return _last_cumu_compaction_schedule_millis; }
295
178k
    void set_last_cumu_compaction_schedule_time(int64_t millis) {
296
178k
        _last_cumu_compaction_schedule_millis = millis;
297
178k
    }
298
299
0
    int64_t last_base_compaction_schedule_time() { return _last_base_compaction_schedule_millis; }
300
50.2k
    void set_last_base_compaction_schedule_time(int64_t millis) {
301
50.2k
        _last_base_compaction_schedule_millis = millis;
302
50.2k
    }
303
304
0
    int64_t last_full_compaction_schedule_time() { return _last_full_compaction_schedule_millis; }
305
0
    void set_last_full_compaction_schedule_time(int64_t millis) {
306
0
        _last_full_compaction_schedule_millis = millis;
307
0
    }
308
309
0
    void set_last_single_compaction_failure_status(std::string status) {
310
0
        _last_single_compaction_failure_status = std::move(status);
311
0
    }
312
313
0
    void set_last_fetched_version(Version version) { _last_fetched_version = std::move(version); }
314
315
    void delete_all_files();
316
317
    void check_tablet_path_exists();
318
319
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_cumulative_compaction();
320
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
321
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction();
322
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_binlog_compaction();
323
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_build_inverted_index(
324
            const std::set<int64_t>& alter_index_uids, bool is_drop_op);
325
326
    // used for single compaction to get the local versions
327
    // Single compaction does not require remote rowsets and cannot violate the cooldown semantics
328
    std::vector<Version> get_all_local_versions();
329
330
    void calculate_cumulative_point();
331
    // TODO(ygl):
332
0
    bool is_primary_replica() { return false; }
333
334
    // return true if the checkpoint is actually done
335
    bool do_tablet_meta_checkpoint();
336
337
    // Check whether the rowset is useful or not, unuseful rowset can be swept up then.
338
    // Rowset which is under tablet's management is useful, i.e. rowset is in
339
    // _rs_version_map, or _stale_rs_version_map.
340
    // Rowset whose version range is not covered by this tablet is also useful.
341
    bool rowset_meta_is_useful(RowsetMetaSharedPtr rowset_meta);
342
343
    void build_tablet_report_info(TTabletInfo* tablet_info,
344
                                  bool enable_consecutive_missing_check = false,
345
                                  bool enable_path_check = false);
346
347
    // return a json string to show the compaction status of this tablet
348
    void get_compaction_status(std::string* json_result);
349
350
    static Status prepare_compaction_and_calculate_permits(
351
            CompactionType compaction_type, const TabletSharedPtr& tablet,
352
            std::shared_ptr<CompactionMixin>& compaction, int64_t& permits,
353
            int8_t prefer_compaction_level = -1);
354
355
    void execute_compaction(CompactionMixin& compaction);
356
    void execute_single_replica_compaction(SingleReplicaCompaction& compaction);
357
358
    void set_cumulative_compaction_policy(
359
0
            std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy) {
360
0
        _cumulative_compaction_policy = cumulative_compaction_policy;
361
0
    }
362
363
0
    std::shared_ptr<CumulativeCompactionPolicy> get_cumulative_compaction_policy() {
364
0
        return _cumulative_compaction_policy;
365
0
    }
366
367
178k
    void set_last_cumu_compaction_status(std::string status) {
368
178k
        _last_cumu_compaction_status = std::move(status);
369
178k
    }
370
371
0
    std::string get_last_cumu_compaction_status() { return _last_cumu_compaction_status; }
372
373
100k
    void set_last_base_compaction_status(std::string status) {
374
100k
        _last_base_compaction_status = std::move(status);
375
100k
    }
376
377
0
    std::string get_last_base_compaction_status() { return _last_base_compaction_status; }
378
379
0
    void set_last_full_compaction_status(std::string status) {
380
0
        _last_full_compaction_status = std::move(status);
381
0
    }
382
383
0
    std::string get_last_full_compaction_status() { return _last_full_compaction_status; }
384
385
492
    void set_last_binlog_compaction_status(std::string status) {
386
492
        _last_binlog_compaction_status = std::move(status);
387
492
    }
388
389
0
    std::string get_last_binlog_compaction_status() { return _last_binlog_compaction_status; }
390
391
    std::tuple<int64_t, int64_t> get_visible_version_and_time() const;
392
393
295k
    void set_visible_version(const std::shared_ptr<const VersionWithTime>& visible_version) {
394
295k
        _visible_version.store(visible_version);
395
295k
    }
396
397
    bool should_fetch_from_peer();
398
399
0
    inline bool all_beta() const {
400
0
        std::shared_lock rdlock(_meta_lock);
401
0
        return _tablet_meta->all_beta();
402
0
    }
403
404
0
    const TabletSchemaSPtr& tablet_schema_unlocked() const { return _max_version_schema; }
405
406
    Result<std::unique_ptr<RowsetWriter>> create_rowset_writer(RowsetWriterContext& context,
407
                                                               bool vertical) override;
408
409
    Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer(
410
            const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
411
            int64_t txn_expiration = 0) override;
412
    Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer(
413
            RowsetWriterContext& context, const RowsetId& rowset_id);
414
415
    Status create_rowset(const RowsetMetaSharedPtr& rowset_meta, RowsetSharedPtr* rowset);
416
417
    // MUST hold EXCLUSIVE `_meta_lock`
418
    void add_rowsets(const std::vector<RowsetSharedPtr>& to_add, bool copy_row_binlog = false);
419
    // MUST hold EXCLUSIVE `_meta_lock`
420
    Status delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete, bool move_to_stale,
421
                          bool copy_row_binlog = false);
422
423
    // MUST hold SHARED `_meta_lock`
424
9.99k
    const auto& rowset_map() const { return _rs_version_map; }
425
    // MUST hold SHARED `_meta_lock`
426
9.98k
    const auto& stale_rowset_map() const { return _stale_rs_version_map; }
427
    // MUST hold SHARED `_meta_lock`
428
0
    const auto& row_binlog_rowset_map() const { return _row_binlog_rs_version_map; }
429
430
    ////////////////////////////////////////////////////////////////////////////
431
    // begin cooldown functions
432
    ////////////////////////////////////////////////////////////////////////////
433
945k
    int64_t storage_policy_id() const { return _tablet_meta->storage_policy_id(); }
434
    void set_storage_policy_id(int64_t id) { _tablet_meta->set_storage_policy_id(id); }
435
436
3.93M
    int64_t last_failed_follow_cooldown_time() const { return _last_failed_follow_cooldown_time; }
437
438
    // Cooldown to remote fs.
439
    Status cooldown(RowsetSharedPtr rowset = nullptr);
440
441
    RowsetSharedPtr pick_cooldown_rowset();
442
443
    RowsetSharedPtr need_cooldown(int64_t* cooldown_timestamp, size_t* file_size);
444
445
    struct CooldownConf {
446
        int64_t term = -1;
447
        int64_t cooldown_replica_id = -1;
448
    };
449
450
7
    CooldownConf cooldown_conf() const {
451
7
        std::shared_lock rlock(_cooldown_conf_lock);
452
7
        return _cooldown_conf;
453
7
    }
454
455
21
    CooldownConf cooldown_conf_unlocked() const { return _cooldown_conf; }
456
457
    // Return `true` if update success
458
    bool update_cooldown_conf(int64_t cooldown_term, int64_t cooldown_replica_id);
459
460
    Status remove_all_remote_rowsets();
461
462
    void record_unused_remote_rowset(const RowsetId& rowset_id, const std::string& resource,
463
                                     int64_t num_segments);
464
465
    uint32_t calc_cold_data_compaction_score() const;
466
467
194
    std::mutex& get_cold_compaction_lock() { return _cold_compaction_lock; }
468
469
0
    std::shared_mutex& get_cooldown_conf_lock() { return _cooldown_conf_lock; }
470
471
    static void async_write_cooldown_meta(TabletSharedPtr tablet);
472
    // Return `ABORTED` if should not to retry again
473
    Status write_cooldown_meta();
474
    ////////////////////////////////////////////////////////////////////////////
475
    // end cooldown functions
476
    ////////////////////////////////////////////////////////////////////////////
477
478
    CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() override;
479
    Status save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
480
                              DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
481
                              const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id = -1,
482
                              int64_t next_visible_version = -1) override;
483
484
    void merge_delete_bitmap(const DeleteBitmap& delete_bitmap);
485
    bool check_all_rowset_segment();
486
487
    void update_max_version_schema(const TabletSchemaSPtr& tablet_schema);
488
489
    void set_skip_compaction(bool skip,
490
                             CompactionType compaction_type = CompactionType::CUMULATIVE_COMPACTION,
491
                             int64_t start = -1);
492
    bool should_skip_compaction(CompactionType compaction_type, int64_t now);
493
494
    std::vector<std::string> get_binlog_filepath(std::string_view binlog_version) const;
495
    std::pair<std::string, int64_t> get_binlog_info(std::string_view binlog_version) const;
496
    std::string get_rowset_binlog_meta(std::string_view binlog_version,
497
                                       std::string_view rowset_id) const;
498
    Status get_rowset_binlog_metas(const std::vector<int64_t>& binlog_versions,
499
                                   RowsetBinlogMetasPB* metas_pb);
500
    Status get_rowset_binlog_metas(Version binlog_versions, RowsetBinlogMetasPB* metas_pb);
501
    std::string get_segment_filepath(std::string_view rowset_id,
502
                                     std::string_view segment_index) const;
503
    std::string get_segment_filepath(std::string_view rowset_id, int64_t segment_index) const;
504
    bool can_add_binlog(uint64_t total_binlog_size) const;
505
    void gc_binlogs(int64_t version);
506
    Status ingest_binlog_metas(RowsetBinlogMetasPB* metas_pb);
507
508
    void report_error(const Status& st);
509
510
0
    inline int64_t get_io_error_times() const { return _io_error_times; }
511
512
1.26M
    inline bool is_io_error_too_times() const {
513
1.26M
        return config::max_tablet_io_errors > 0 && _io_error_times >= config::max_tablet_io_errors;
514
1.26M
    }
515
516
40.3k
    int64_t get_table_id() { return _tablet_meta->table_id(); }
517
518
    // binlog related functions
519
73.3k
    bool enable_binlog() const {
520
73.3k
        return config::enable_feature_binlog && _tablet_meta->binlog_config().is_enable();
521
73.3k
    }
522
73.3k
    bool enable_ccr_binlog() const {
523
73.3k
        return enable_binlog() && _tablet_meta->binlog_config().is_ccr_binlog_format();
524
73.3k
    }
525
407M
    bool enable_row_binlog() const {
526
407M
        return _tablet_meta->binlog_config().is_enable() &&
527
407M
               _tablet_meta->binlog_config().is_row_binlog_format();
528
407M
    }
529
530
0
    int64_t binlog_ttl_ms() const { return _tablet_meta->binlog_config().ttl_seconds(); }
531
0
    int64_t binlog_max_bytes() const { return _tablet_meta->binlog_config().max_bytes(); }
532
533
    void set_binlog_config(BinlogConfig binlog_config);
534
535
    // row_binlog
536
9
    int32_t row_binlog_schema_hash() const { return _tablet_meta->row_binlog_schema_hash(); }
537
538
0
    void set_is_full_compaction_running(bool is_full_compaction_running) {
539
0
        _is_full_compaction_running = is_full_compaction_running;
540
0
    }
541
0
    inline bool is_full_compaction_running() const { return _is_full_compaction_running; }
542
    void clear_cache() override;
543
544
    int32_t get_compaction_score() const { return _compaction_score; }
545
546
0
    void set_compaction_score(int32_t compaction_score) { _compaction_score = compaction_score; }
547
548
60.7k
    void add_compaction_score(int32_t score) {
549
60.7k
        if (_compaction_score < 0) {
550
26.4k
            return;
551
26.4k
        }
552
34.2k
        _compaction_score += score;
553
34.2k
    }
554
555
0
    void minus_compaction_score(int32_t score) {
556
0
        if (_compaction_score < 0) {
557
0
            return;
558
0
        }
559
0
        _compaction_score -= score;
560
0
    }
561
562
    Status prepare_txn(TPartitionId partition_id, TTransactionId transaction_id,
563
                       const PUniqueId& load_id, bool ingest);
564
    // TODO: commit_txn
565
566
private:
567
    Status _init_once_action();
568
    bool _contains_rowset(const RowsetId rowset_id);
569
    Status _contains_version(const Version& version);
570
    Status _add_row_binlog_rowset_unlocked(const RowsetSharedPtr& rowset,
571
                                           const RowsetSharedPtr& row_binlog_rowset);
572
573
    // Returns:
574
    // version: the max continuous version from beginning
575
    // max_version: the max version of this tablet
576
    void _max_continuous_version_from_beginning_unlocked(Version* version, Version* max_version,
577
                                                         bool* has_version_cross) const;
578
    RowsetSharedPtr _rowset_with_largest_size();
579
    /// Delete stale rowset by version. This method not only delete the version in expired rowset map,
580
    /// but also delete the version in rowset meta vector.
581
    void _delete_stale_rowset_by_version(const Version& version);
582
583
    uint32_t _calc_cumulative_compaction_score(
584
            std::shared_ptr<CumulativeCompactionPolicy> cumulative_compaction_policy);
585
    uint32_t _calc_base_compaction_score() const;
586
    uint32_t _calc_binlog_compaction_score(int8_t* prefer_compaction_level = nullptr);
587
588
    std::vector<RowsetSharedPtr> _pick_visible_rowsets_to_compaction(int64_t min_start_version,
589
                                                                     int64_t max_start_version);
590
591
    void _init_context_common_fields(RowsetWriterContext& context);
592
593
    ////////////////////////////////////////////////////////////////////////////
594
    // begin cooldown functions
595
    ////////////////////////////////////////////////////////////////////////////
596
    Status _cooldown_data(RowsetSharedPtr rowset);
597
    Status _follow_cooldowned_data();
598
    Status _read_cooldown_meta(const StorageResource& storage_resource,
599
                               TabletMetaPB* tablet_meta_pb);
600
    bool _has_data_to_cooldown();
601
    int64_t _get_newest_cooldown_time(const RowsetSharedPtr& rowset);
602
    ////////////////////////////////////////////////////////////////////////////
603
    // end cooldown functions
604
    ////////////////////////////////////////////////////////////////////////////
605
606
    void _clear_cache_by_rowset(const BetaRowsetSharedPtr& rowset);
607
    void check_table_size_correctness();
608
    std::string get_segment_path(const RowsetMetaSharedPtr& rs_meta, int64_t seg_id);
609
    int64_t get_segment_file_size(const RowsetMetaSharedPtr& rs_meta);
610
    int64_t get_inverted_index_file_size(const RowsetMetaSharedPtr& rs_meta);
611
612
public:
613
    static const int64_t K_INVALID_CUMULATIVE_POINT = -1;
614
615
private:
616
    StorageEngine& _engine;
617
    DataDir* _data_dir = nullptr;
618
619
    std::string _tablet_path;
620
621
    DorisCallOnce<Status> _init_once;
622
    // meta store lock is used for prevent 2 threads do checkpoint concurrently
623
    // it will be used in econ-mode in the future
624
    std::shared_mutex _meta_store_lock;
625
    std::mutex _ingest_lock;
626
    std::mutex _base_compaction_lock;
627
    std::mutex _cumulative_compaction_lock;
628
    std::mutex _binlog_compaction_lock;
629
    std::shared_timed_mutex _migration_lock;
630
    std::mutex _build_inverted_index_lock;
631
632
    // In unique key table with MoW, we should guarantee that only one
633
    // writer can update rowset and delete bitmap at the same time.
634
    // We use a separate lock rather than _meta_lock, to avoid blocking read queries
635
    // during publish_txn, which might take hundreds of milliseconds
636
    mutable std::mutex _rowset_update_lock;
637
638
    // if this tablet is broken, set to true. default is false
639
    std::atomic<bool> _is_bad;
640
    // timestamp of last cumu compaction failure
641
    std::atomic<int64_t> _last_cumu_compaction_failure_millis;
642
    // timestamp of last base compaction failure
643
    std::atomic<int64_t> _last_base_compaction_failure_millis;
644
    // timestamp of last full compaction failure
645
    std::atomic<int64_t> _last_full_compaction_failure_millis;
646
    // timestamp of last cumu compaction success
647
    std::atomic<int64_t> _last_cumu_compaction_success_millis;
648
    // timestamp of last base compaction success
649
    std::atomic<int64_t> _last_base_compaction_success_millis;
650
    // timestamp of last full compaction success
651
    std::atomic<int64_t> _last_full_compaction_success_millis;
652
    // timestamp of last binlog compaction success for each compaction level
653
    std::array<std::atomic<int64_t>, BinlogCompactionPolicy::kBinlogCompactionMaxLevel>
654
            _last_binlog_compaction_success_millis;
655
    // timestamp of last binlog compaction failure
656
    std::atomic<int64_t> _last_binlog_compaction_failure_millis;
657
    // timestamp of last cumu compaction schedule time
658
    std::atomic<int64_t> _last_cumu_compaction_schedule_millis;
659
    // timestamp of last base compaction schedule time
660
    std::atomic<int64_t> _last_base_compaction_schedule_millis;
661
    // timestamp of last full compaction schedule time
662
    std::atomic<int64_t> _last_full_compaction_schedule_millis;
663
    std::atomic<int64_t> _cumulative_point;
664
    std::atomic<int64_t> _cumulative_promotion_size;
665
    std::atomic<int32_t> _newly_created_rowset_num;
666
    std::atomic<int64_t> _last_checkpoint_time;
667
    std::string _last_cumu_compaction_status;
668
    std::string _last_base_compaction_status;
669
    std::string _last_full_compaction_status;
670
    std::string _last_binlog_compaction_status;
671
672
    // single replica compaction status
673
    std::string _last_single_compaction_failure_status;
674
    Version _last_fetched_version;
675
676
    // cumulative compaction policy
677
    std::shared_ptr<CumulativeCompactionPolicy> _cumulative_compaction_policy;
678
    std::string_view _cumulative_compaction_type;
679
    std::shared_ptr<BinlogCompactionPolicy> _binlog_compaction_policy;
680
681
    // use a separate thread to check all tablets paths existence
682
    std::atomic<bool> _is_tablet_path_exists;
683
684
    int64_t _last_missed_version;
685
    int64_t _last_missed_time_s;
686
687
    bool _skip_cumu_compaction = false;
688
    int64_t _skip_cumu_compaction_ts;
689
690
    bool _skip_base_compaction = false;
691
    int64_t _skip_base_compaction_ts;
692
693
    bool _skip_binlog_compaction = false;
694
    int64_t _skip_binlog_compaction_ts;
695
696
    // cooldown related
697
    CooldownConf _cooldown_conf;
698
    // `_cooldown_conf_lock` is used to serialize update cooldown conf and all operations that:
699
    // 1. read cooldown conf
700
    // 2. upload rowsets to remote storage
701
    // 3. update cooldown meta id
702
    mutable std::shared_mutex _cooldown_conf_lock;
703
    // `_cold_compaction_lock` is used to serialize cold data compaction and all operations that
704
    // may delete compaction input rowsets.
705
    std::mutex _cold_compaction_lock;
706
    int64_t _last_failed_follow_cooldown_time = 0;
707
708
    int64_t _io_error_times = 0;
709
710
    // partition's visible version. it sync from fe, but not real-time.
711
    atomic_shared_ptr<const VersionWithTime> _visible_version;
712
713
    std::atomic_bool _is_full_compaction_running = false;
714
715
    int32_t _compaction_score = -1;
716
    int32_t _score_check_cnt = 0;
717
};
718
719
185k
inline CumulativeCompactionPolicy* Tablet::cumulative_compaction_policy() {
720
185k
    return _cumulative_compaction_policy.get();
721
185k
}
722
723
492
inline BinlogCompactionPolicy* Tablet::binlog_compaction_policy() {
724
492
    return _binlog_compaction_policy.get();
725
492
}
726
727
15.5M
inline bool Tablet::init_succeeded() {
728
15.5M
    return _init_once.has_called() && _init_once.stored_result().ok();
729
15.5M
}
730
731
18.2M
inline bool Tablet::is_used() {
732
18.3M
    return !_is_bad && _data_dir->is_used();
733
18.2M
}
734
735
290k
inline void Tablet::register_tablet_into_dir() {
736
290k
    _data_dir->register_tablet(this);
737
290k
}
738
739
5.14k
inline void Tablet::deregister_tablet_from_dir() {
740
5.14k
    _data_dir->deregister_tablet(this);
741
5.14k
}
742
743
1.39M
inline int64_t Tablet::cumulative_layer_point() const {
744
1.39M
    return _cumulative_point;
745
1.39M
}
746
747
9.46k
inline void Tablet::set_cumulative_layer_point(int64_t new_point) {
748
    // cumulative point should only be reset to -1, or be increased
749
18.4E
    CHECK(new_point == Tablet::K_INVALID_CUMULATIVE_POINT || new_point >= _cumulative_point)
750
18.4E
            << "Unexpected cumulative point: " << new_point
751
18.4E
            << ", origin: " << _cumulative_point.load();
752
9.46k
    _cumulative_point = new_point;
753
9.46k
}
754
755
179k
inline int64_t Tablet::cumulative_promotion_size() const {
756
179k
    return _cumulative_promotion_size;
757
179k
}
758
759
299k
inline void Tablet::set_cumulative_promotion_size(int64_t new_size) {
760
299k
    _cumulative_promotion_size = new_size;
761
299k
}
762
763
// TODO(lingbin): Why other methods that need to get information from _tablet_meta
764
// are not locked, here needs a comment to explain.
765
31.2k
inline size_t Tablet::tablet_footprint() {
766
31.2k
    std::shared_lock rdlock(_meta_lock);
767
31.2k
    return _tablet_meta->tablet_footprint();
768
31.2k
}
769
770
3.27M
inline size_t Tablet::tablet_local_size() {
771
3.27M
    std::shared_lock rdlock(_meta_lock);
772
3.27M
    return _tablet_meta->tablet_local_size();
773
3.27M
}
774
775
3.27M
inline size_t Tablet::tablet_remote_size() {
776
3.27M
    std::shared_lock rdlock(_meta_lock);
777
3.27M
    return _tablet_meta->tablet_remote_size();
778
3.27M
}
779
780
// TODO(lingbin): Why other methods which need to get information from _tablet_meta
781
// are not locked, here needs a comment to explain.
782
inline size_t Tablet::num_rows() {
783
    std::shared_lock rdlock(_meta_lock);
784
    return _tablet_meta->num_rows();
785
}
786
787
4.85M
inline size_t Tablet::version_count() const {
788
4.85M
    std::shared_lock rdlock(_meta_lock);
789
4.85M
    return _tablet_meta->version_count();
790
4.85M
}
791
792
37.1k
inline size_t Tablet::stale_version_count() const {
793
37.1k
    std::shared_lock rdlock(_meta_lock);
794
37.1k
    return _tablet_meta->stale_version_count();
795
37.1k
}
796
797
463k
inline Version Tablet::max_version() const {
798
463k
    std::shared_lock rdlock(_meta_lock);
799
463k
    return _tablet_meta->max_version();
800
463k
}
801
802
4.77M
inline uint64_t Tablet::segment_count() const {
803
4.77M
    std::shared_lock rdlock(_meta_lock);
804
4.77M
    uint64_t segment_nums = 0;
805
8.42M
    for (const auto& [_, rs_meta] : _tablet_meta->all_rs_metas()) {
806
8.42M
        segment_nums += rs_meta->num_segments();
807
8.42M
    }
808
4.77M
    return segment_nums;
809
4.77M
}
810
811
0
inline SortType Tablet::sort_type() const {
812
0
    return _tablet_meta->tablet_schema()->sort_type();
813
0
}
814
815
0
inline size_t Tablet::sort_col_num() const {
816
0
    return _tablet_meta->tablet_schema()->sort_col_num();
817
0
}
818
819
0
inline size_t Tablet::num_columns() const {
820
0
    return _tablet_meta->tablet_schema()->num_columns();
821
0
}
822
823
0
inline size_t Tablet::num_null_columns() const {
824
0
    return _tablet_meta->tablet_schema()->num_null_columns();
825
0
}
826
827
0
inline size_t Tablet::num_short_key_columns() const {
828
0
    return _tablet_meta->tablet_schema()->num_short_key_columns();
829
0
}
830
831
0
inline size_t Tablet::num_rows_per_row_block() const {
832
0
    return _tablet_meta->tablet_schema()->num_rows_per_row_block();
833
0
}
834
835
0
inline double Tablet::bloom_filter_fpp() const {
836
0
    return _tablet_meta->tablet_schema()->bloom_filter_fpp();
837
0
}
838
839
0
inline size_t Tablet::next_unique_id() const {
840
0
    return _tablet_meta->tablet_schema()->next_column_unique_id();
841
0
}
842
843
37.1k
inline int64_t Tablet::avg_rs_meta_serialize_size() const {
844
37.1k
    return _tablet_meta->avg_rs_meta_serialize_size();
845
37.1k
}
846
847
class TabletCopyType {
848
public:
849
    static constexpr int32_t DEFAULT = TTabletCopyType::DATA | TTabletCopyType::CCR_BINLOG;
850
851
640
    static bool has(int32_t copy_type, TTabletCopyType::type flag) {
852
640
        return (copy_type & static_cast<int32_t>(flag)) != 0;
853
640
    }
854
855
320
    static Status validate(int32_t copy_type) {
856
322
        if (copy_type <= 0 || (copy_type & ~all_types()) != 0) {
857
0
            return Status::Error<ErrorCode::INVALID_ARGUMENT>(
858
0
                    "invalid copy_type bitmask: {}, valid bits: {}", copy_type, all_types());
859
0
        }
860
320
        return Status::OK();
861
320
    }
862
863
private:
864
322
    static constexpr int32_t all_types() {
865
322
        return TTabletCopyType::DATA | TTabletCopyType::ROW_BINLOG | TTabletCopyType::CCR_BINLOG;
866
322
    }
867
};
868
869
} // namespace doris