Coverage Report

Created: 2025-06-18 11:42

/root/doris/be/src/cloud/cloud_tablet.h
Line
Count
Source (jump to first uncovered line)
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <memory>
21
22
#include "olap/base_tablet.h"
23
#include "olap/partial_update_info.h"
24
25
namespace doris {
26
27
class CloudStorageEngine;
28
29
struct SyncRowsetStats {
30
    int64_t get_remote_rowsets_num {0};
31
    int64_t get_remote_rowsets_rpc_ns {0};
32
33
    int64_t get_local_delete_bitmap_rowsets_num {0};
34
    int64_t get_remote_delete_bitmap_rowsets_num {0};
35
    int64_t get_remote_delete_bitmap_key_count {0};
36
    int64_t get_remote_delete_bitmap_bytes {0};
37
    int64_t get_remote_delete_bitmap_rpc_ns {0};
38
39
    int64_t get_remote_tablet_meta_rpc_ns {0};
40
    int64_t tablet_meta_cache_hit {0};
41
    int64_t tablet_meta_cache_miss {0};
42
};
43
44
struct SyncOptions {
45
    bool warmup_delta_data = false;
46
    bool sync_delete_bitmap = true;
47
    bool full_sync = false;
48
    bool merge_schema = false;
49
    int64_t query_version = -1;
50
};
51
52
class CloudTablet final : public BaseTablet {
53
public:
54
    CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta);
55
56
    ~CloudTablet() override;
57
58
    bool exceed_version_limit(int32_t limit) override;
59
60
    Result<std::unique_ptr<RowsetWriter>> create_rowset_writer(RowsetWriterContext& context,
61
                                                               bool vertical) override;
62
63
    Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
64
                              bool skip_missing_version) override;
65
66
    Status capture_consistent_rowsets_unlocked(
67
            const Version& spec_version, std::vector<RowsetSharedPtr>* rowsets) const override;
68
69
0
    size_t tablet_footprint() override {
70
0
        return _approximate_data_size.load(std::memory_order_relaxed);
71
0
    }
72
73
    const std::string& tablet_path() const override;
74
75
    // clang-format off
76
0
    int64_t fetch_add_approximate_num_rowsets (int64_t x) { return _approximate_num_rowsets .fetch_add(x, std::memory_order_relaxed); }
77
0
    int64_t fetch_add_approximate_num_segments(int64_t x) { return _approximate_num_segments.fetch_add(x, std::memory_order_relaxed); }
78
0
    int64_t fetch_add_approximate_num_rows    (int64_t x) { return _approximate_num_rows    .fetch_add(x, std::memory_order_relaxed); }
79
0
    int64_t fetch_add_approximate_data_size   (int64_t x) { return _approximate_data_size   .fetch_add(x, std::memory_order_relaxed); }
80
0
    int64_t fetch_add_approximate_cumu_num_rowsets (int64_t x) { return _approximate_cumu_num_rowsets.fetch_add(x, std::memory_order_relaxed); }
81
0
    int64_t fetch_add_approximate_cumu_num_deltas   (int64_t x) { return _approximate_cumu_num_deltas.fetch_add(x, std::memory_order_relaxed); }
82
    // clang-format on
83
84
    // meta lock must be held when calling this function
85
    void reset_approximate_stats(int64_t num_rowsets, int64_t num_segments, int64_t num_rows,
86
                                 int64_t data_size);
87
88
    // return a json string to show the compaction status of this tablet
89
    void get_compaction_status(std::string* json_result);
90
91
    // Synchronize the rowsets from meta service.
92
    // If tablet state is not `TABLET_RUNNING`, sync tablet meta and all visible rowsets.
93
    // If `query_version` > 0 and local max_version of the tablet >= `query_version`, do nothing.
94
    // If 'need_download_data_async' is true, it means that we need to download the new version
95
    // rowsets datum async.
96
    Status sync_rowsets(const SyncOptions& options = {}, SyncRowsetStats* stats = nullptr);
97
98
    // Synchronize the tablet meta from meta service.
99
    Status sync_meta();
100
101
    // If `version_overlap` is true, function will delete rowsets with overlapped version in this tablet.
102
    // If 'warmup_delta_data' is true, download the new version rowset data in background.
103
    // MUST hold EXCLUSIVE `_meta_lock`.
104
    // If 'need_download_data_async' is true, it means that we need to download the new version
105
    // rowsets datum async.
106
    void add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
107
                     std::unique_lock<std::shared_mutex>& meta_lock,
108
                     bool warmup_delta_data = false);
109
110
    // MUST hold EXCLUSIVE `_meta_lock`.
111
    void delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
112
                        std::unique_lock<std::shared_mutex>& meta_lock);
113
114
    // When the tablet is dropped, we need to recycle cached data:
115
    // 1. The data in file cache
116
    // 2. The memory in tablet cache
117
    void clear_cache() override;
118
119
    // Return number of deleted stale rowsets
120
    uint64_t delete_expired_stale_rowsets();
121
122
0
    bool has_stale_rowsets() const { return !_stale_rs_version_map.empty(); }
123
124
    int64_t get_cloud_base_compaction_score() const;
125
    int64_t get_cloud_cumu_compaction_score() const;
126
127
0
    int64_t max_version_unlocked() const override { return _max_version; }
128
0
    int64_t base_compaction_cnt() const { return _base_compaction_cnt; }
129
0
    int64_t cumulative_compaction_cnt() const { return _cumulative_compaction_cnt; }
130
0
    int64_t cumulative_layer_point() const {
131
0
        return _cumulative_point.load(std::memory_order_relaxed);
132
0
    }
133
134
0
    void set_base_compaction_cnt(int64_t cnt) { _base_compaction_cnt = cnt; }
135
0
    void set_cumulative_compaction_cnt(int64_t cnt) { _cumulative_compaction_cnt = cnt; }
136
    void set_cumulative_layer_point(int64_t new_point);
137
138
1
    int64_t last_cumu_compaction_failure_time() { return _last_cumu_compaction_failure_millis; }
139
3
    void set_last_cumu_compaction_failure_time(int64_t millis) {
140
3
        _last_cumu_compaction_failure_millis = millis;
141
3
    }
142
143
2
    int64_t last_base_compaction_failure_time() { return _last_base_compaction_failure_millis; }
144
3
    void set_last_base_compaction_failure_time(int64_t millis) {
145
3
        _last_base_compaction_failure_millis = millis;
146
3
    }
147
148
0
    int64_t last_full_compaction_failure_time() { return _last_full_compaction_failure_millis; }
149
0
    void set_last_full_compaction_failure_time(int64_t millis) {
150
0
        _last_full_compaction_failure_millis = millis;
151
0
    }
152
153
0
    int64_t last_cumu_compaction_success_time() { return _last_cumu_compaction_success_millis; }
154
0
    void set_last_cumu_compaction_success_time(int64_t millis) {
155
0
        _last_cumu_compaction_success_millis = millis;
156
0
    }
157
158
0
    int64_t last_base_compaction_success_time() { return _last_base_compaction_success_millis; }
159
0
    void set_last_base_compaction_success_time(int64_t millis) {
160
0
        _last_base_compaction_success_millis = millis;
161
0
    }
162
163
0
    int64_t last_full_compaction_success_time() { return _last_full_compaction_success_millis; }
164
0
    void set_last_full_compaction_success_time(int64_t millis) {
165
0
        _last_full_compaction_success_millis = millis;
166
0
    }
167
168
0
    int64_t last_cumu_compaction_schedule_time() { return _last_cumu_compaction_schedule_millis; }
169
0
    void set_last_cumu_compaction_schedule_time(int64_t millis) {
170
0
        _last_cumu_compaction_schedule_millis = millis;
171
0
    }
172
173
0
    int64_t last_base_compaction_schedule_time() { return _last_base_compaction_schedule_millis; }
174
0
    void set_last_base_compaction_schedule_time(int64_t millis) {
175
0
        _last_base_compaction_schedule_millis = millis;
176
0
    }
177
178
0
    int64_t last_full_compaction_schedule_time() { return _last_full_compaction_schedule_millis; }
179
0
    void set_last_full_compaction_schedule_time(int64_t millis) {
180
0
        _last_full_compaction_schedule_millis = millis;
181
0
    }
182
183
0
    void set_last_cumu_compaction_status(std::string status) {
184
0
        _last_cumu_compaction_status = std::move(status);
185
0
    }
186
187
0
    std::string get_last_cumu_compaction_status() { return _last_cumu_compaction_status; }
188
189
0
    void set_last_base_compaction_status(std::string status) {
190
0
        _last_base_compaction_status = std::move(status);
191
0
    }
192
193
0
    std::string get_last_base_compaction_status() { return _last_base_compaction_status; }
194
195
0
    void set_last_full_compaction_status(std::string status) {
196
0
        _last_full_compaction_status = std::move(status);
197
0
    }
198
199
0
    std::string get_last_full_compaction_status() { return _last_full_compaction_status; }
200
201
0
    int64_t alter_version() const { return _alter_version; }
202
0
    void set_alter_version(int64_t alter_version) { _alter_version = alter_version; }
203
204
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
205
206
0
    inline Version max_version() const {
207
0
        std::shared_lock rdlock(_meta_lock);
208
0
        return _tablet_meta->max_version();
209
0
    }
210
211
0
    int64_t base_size() const { return _base_size; }
212
213
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction();
214
215
0
    std::mutex& get_base_compaction_lock() { return _base_compaction_lock; }
216
0
    std::mutex& get_cumulative_compaction_lock() { return _cumulative_compaction_lock; }
217
218
    Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer(
219
            const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
220
            int64_t txn_expiration = 0) override;
221
222
    CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() override;
223
224
    Status save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
225
                              DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
226
                              const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id = -1,
227
                              int64_t next_visible_version = -1) override;
228
229
    Status save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id,
230
                                    DeleteBitmapPtr delete_bitmap, int64_t lock_id,
231
                                    int64_t next_visible_version);
232
233
    Status calc_delete_bitmap_for_compaction(const std::vector<RowsetSharedPtr>& input_rowsets,
234
                                             const RowsetSharedPtr& output_rowset,
235
                                             const RowIdConversion& rowid_conversion,
236
                                             ReaderType compaction_type, int64_t merged_rows,
237
                                             int64_t filtered_rows, int64_t initiator,
238
                                             DeleteBitmapPtr& output_rowset_delete_bitmap,
239
                                             bool allow_delete_in_cumu_compaction,
240
                                             int64_t& get_delete_bitmap_lock_start_time);
241
242
    // Find the missed versions until the spec_version.
243
    //
244
    // for example:
245
    //     [0-4][5-5][8-8][9-9][14-14]
246
    // if spec_version = 12, it will return [6-7],[10-12]
247
    Versions calc_missed_versions(int64_t spec_version, Versions existing_versions) const override;
248
249
0
    std::mutex& get_rowset_update_lock() { return _rowset_update_lock; }
250
251
0
    bthread::Mutex& get_sync_meta_lock() { return _sync_meta_lock; }
252
253
0
    const auto& rowset_map() const { return _rs_version_map; }
254
255
    // Merge all rowset schemas within a CloudTablet
256
    Status merge_rowsets_schema();
257
258
    int64_t last_sync_time_s = 0;
259
    int64_t last_load_time_ms = 0;
260
    int64_t last_base_compaction_success_time_ms = 0;
261
    int64_t last_cumu_compaction_success_time_ms = 0;
262
    int64_t last_cumu_no_suitable_version_ms = 0;
263
    int64_t last_access_time_ms = 0;
264
265
    std::atomic<int64_t> local_read_time_us = 0;
266
    std::atomic<int64_t> remote_read_time_us = 0;
267
    std::atomic<int64_t> exec_compaction_time_us = 0;
268
269
    // Return merged extended schema
270
    TabletSchemaSPtr merged_tablet_schema() const override;
271
272
    void build_tablet_report_info(TTabletInfo* tablet_info);
273
274
    static void recycle_cached_data(const std::vector<RowsetSharedPtr>& rowsets);
275
276
    // check that if the delete bitmap in delete bitmap cache has the same cardinality with the expected_delete_bitmap's
277
    Status check_delete_bitmap_cache(int64_t txn_id, DeleteBitmap* expected_delete_bitmap) override;
278
279
    void agg_delete_bitmap_for_compaction(int64_t start_version, int64_t end_version,
280
                                          const std::vector<RowsetSharedPtr>& pre_rowsets,
281
                                          DeleteBitmapPtr& new_delete_bitmap,
282
                                          std::map<std::string, int64_t>& pre_rowset_to_versions);
283
284
    bool need_remove_unused_rowsets();
285
286
    void add_unused_rowsets(const std::vector<RowsetSharedPtr>& rowsets);
287
    void remove_unused_rowsets();
288
289
private:
290
    // FIXME(plat1ko): No need to record base size if rowsets are ordered by version
291
    void update_base_size(const Rowset& rs);
292
293
    Status sync_if_not_running(SyncRowsetStats* stats = nullptr);
294
295
    CloudStorageEngine& _engine;
296
297
    // this mutex MUST ONLY be used when sync meta
298
    bthread::Mutex _sync_meta_lock;
299
    // ATTENTION: lock order should be: _sync_meta_lock -> _meta_lock
300
301
    std::atomic<int64_t> _cumulative_point {-1};
302
    std::atomic<int64_t> _approximate_num_rowsets {-1};
303
    std::atomic<int64_t> _approximate_num_segments {-1};
304
    std::atomic<int64_t> _approximate_num_rows {-1};
305
    std::atomic<int64_t> _approximate_data_size {-1};
306
    std::atomic<int64_t> _approximate_cumu_num_rowsets {-1};
307
    // Number of sorted arrays (e.g. for rowset with N segments, if rowset is overlapping, delta is N, otherwise 1) after cumu point
308
    std::atomic<int64_t> _approximate_cumu_num_deltas {-1};
309
310
    // timestamp of last cumu compaction failure
311
    std::atomic<int64_t> _last_cumu_compaction_failure_millis;
312
    // timestamp of last base compaction failure
313
    std::atomic<int64_t> _last_base_compaction_failure_millis;
314
    // timestamp of last full compaction failure
315
    std::atomic<int64_t> _last_full_compaction_failure_millis;
316
    // timestamp of last cumu compaction success
317
    std::atomic<int64_t> _last_cumu_compaction_success_millis;
318
    // timestamp of last base compaction success
319
    std::atomic<int64_t> _last_base_compaction_success_millis;
320
    // timestamp of last full compaction success
321
    std::atomic<int64_t> _last_full_compaction_success_millis;
322
    // timestamp of last cumu compaction schedule time
323
    std::atomic<int64_t> _last_cumu_compaction_schedule_millis;
324
    // timestamp of last base compaction schedule time
325
    std::atomic<int64_t> _last_base_compaction_schedule_millis;
326
    // timestamp of last full compaction schedule time
327
    std::atomic<int64_t> _last_full_compaction_schedule_millis;
328
329
    std::string _last_cumu_compaction_status;
330
    std::string _last_base_compaction_status;
331
    std::string _last_full_compaction_status;
332
333
    int64_t _base_compaction_cnt = 0;
334
    int64_t _cumulative_compaction_cnt = 0;
335
    int64_t _max_version = -1;
336
    int64_t _base_size = 0;
337
    int64_t _alter_version = -1;
338
339
    std::mutex _base_compaction_lock;
340
    std::mutex _cumulative_compaction_lock;
341
342
    // To avoid multiple calc delete bitmap tasks on same (txn_id, tablet_id) with different
343
    // signatures being executed concurrently, we use _rowset_update_lock to serialize them
344
    mutable std::mutex _rowset_update_lock;
345
346
    // Schema will be merged from all rowsets when sync_rowsets
347
    TabletSchemaSPtr _merged_tablet_schema;
348
349
    // unused_rowsets, [start_version, end_version]
350
    std::mutex _gc_mutex;
351
    std::unordered_map<RowsetId, RowsetSharedPtr> _unused_rowsets;
352
    std::vector<std::pair<std::vector<RowsetId>, DeleteBitmapKeyRanges>> _unused_delete_bitmap;
353
};
354
355
using CloudTabletSPtr = std::shared_ptr<CloudTablet>;
356
357
} // namespace doris