Coverage Report

Created: 2026-10-26 12:21

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/cloud/cloud_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 <memory>
21
#include <optional>
22
#include <tuple>
23
24
#include "storage/partial_update_info.h"
25
#include "storage/rowset/rowset.h"
26
#include "storage/tablet/base_tablet.h"
27
28
namespace doris {
29
30
class CloudStorageEngine;
31
32
enum class WarmUpTriggerSource : int { NONE, SYNC_ROWSET, EVENT_DRIVEN, JOB };
33
34
enum class WarmUpProgress : int { NONE, DOING, DONE };
35
36
struct WarmUpState {
37
    WarmUpTriggerSource trigger_source {WarmUpTriggerSource::NONE};
38
    WarmUpProgress progress {WarmUpProgress::NONE};
39
40
    bool operator==(const WarmUpState& other) const {
41
        return trigger_source == other.trigger_source && progress == other.progress;
42
    }
43
};
44
45
struct SyncRowsetStats {
46
    int64_t get_remote_rowsets_num {0};
47
    int64_t get_remote_rowsets_rpc_ns {0};
48
49
    int64_t get_local_delete_bitmap_rowsets_num {0};
50
    int64_t get_remote_delete_bitmap_rowsets_num {0};
51
    int64_t get_remote_delete_bitmap_key_count {0};
52
    int64_t get_remote_delete_bitmap_bytes {0};
53
    int64_t get_remote_delete_bitmap_rpc_ns {0};
54
55
    int64_t get_remote_tablet_meta_rpc_ns {0};
56
    int64_t tablet_meta_cache_hit {0};
57
    int64_t tablet_meta_cache_miss {0};
58
59
    int64_t bthread_schedule_delay_ns {0};
60
    int64_t meta_lock_wait_ns {0}; // _meta_lock (BthreadSharedMutex) wait across all acquisitions
61
    int64_t sync_meta_lock_wait_ns {
62
            0}; // _sync_meta_lock (bthread::Mutex) wait across all acquisitions
63
};
64
65
struct SyncOptions {
66
    bool warmup_delta_data = false;
67
    bool sync_delete_bitmap = true;
68
    bool full_sync = false;
69
    bool merge_schema = false;
70
    int64_t query_version = -1;
71
};
72
73
struct RecycledRowsets {
74
    RowsetId rowset_id;
75
    int64_t num_segments;
76
    std::vector<int64_t> segment_ids;
77
    std::vector<std::string> index_file_names;
78
};
79
80
class CloudTablet final : public BaseTablet {
81
public:
82
    // rowset id -> [(segment id, version, serialized delete bitmap size)]
83
    using PreRowsetDeleteBitmapStats = std::map<
84
            std::string,
85
            std::vector<std::tuple<DeleteBitmap::SegmentId, DeleteBitmap::Version, size_t>>>;
86
87
    CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta);
88
89
    ~CloudTablet() override;
90
91
    bool exceed_version_limit(int32_t limit) override;
92
93
    Result<std::unique_ptr<RowsetWriter>> create_rowset_writer(RowsetWriterContext& context,
94
                                                               bool vertical) override;
95
96
    Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
97
                              const CaptureRowsetOps& opts) override;
98
99
    [[nodiscard]] Result<std::vector<Version>> capture_consistent_versions_unlocked(
100
            const Version& version_range, const CaptureRowsetOps& options) const override;
101
102
    // Capture versions with cache preference optimization.
103
    // This method prioritizes using cached/warmed-up rowsets when building version paths,
104
    // avoiding cold data reads when possible. It uses capture_consistent_versions_prefer_cache
105
    // to find a consistent version path that prefers already warmed-up rowsets.
106
    Result<std::vector<Version>> capture_versions_prefer_cache(const Version& spec_version) const;
107
108
    // Capture versions with query freshness tolerance.
109
    // This method finds a consistent version path where all rowsets are warmed up,
110
    // but allows fallback to normal capture if there are newer rowsets that should be
111
    // visible (based on freshness tolerance) but haven't been warmed up yet.
112
    // For merge-on-write tables, uses special validation to ensure data correctness.
113
    //
114
    // IMPORTANT: The returned version may be smaller than the requested version if newer
115
    // data hasn't been warmed up yet. This can cause different tablets in the same query
116
    // to read from different versions, potentially leading to inconsistent query results.
117
    //
118
    // @param options.query_freshness_tolerance_ms: Time tolerance in milliseconds. Rowsets that
119
    //        became visible within this time range (after current_time - query_freshness_tolerance_ms)
120
    //        can be skipped if not warmed up. However, if older rowsets (before this time point)
121
    //        are not warmed up, the method will fallback to normal capture.
122
    Result<std::vector<Version>> capture_versions_with_freshness_tolerance(
123
            const Version& spec_version, const CaptureRowsetOps& options) const;
124
125
554k
    size_t tablet_footprint() override {
126
554k
        return _approximate_data_size.load(std::memory_order_relaxed);
127
554k
    }
128
129
    std::string tablet_path() const override;
130
131
    // clang-format off
132
5.65M
    int64_t fetch_add_approximate_num_rowsets (int64_t x) { return _approximate_num_rowsets .fetch_add(x, std::memory_order_relaxed); }
133
183k
    int64_t fetch_add_approximate_num_segments(int64_t x) { return _approximate_num_segments.fetch_add(x, std::memory_order_relaxed); }
134
183k
    int64_t fetch_add_approximate_num_rows    (int64_t x) { return _approximate_num_rows    .fetch_add(x, std::memory_order_relaxed); }
135
183k
    int64_t fetch_add_approximate_data_size   (int64_t x) { return _approximate_data_size   .fetch_add(x, std::memory_order_relaxed); }
136
206k
    int64_t fetch_add_approximate_cumu_num_rowsets (int64_t x) { return _approximate_cumu_num_rowsets.fetch_add(x, std::memory_order_relaxed); }
137
186k
    int64_t fetch_add_approximate_cumu_num_deltas   (int64_t x) { return _approximate_cumu_num_deltas.fetch_add(x, std::memory_order_relaxed); }
138
    // clang-format on
139
140
    // meta lock must be held when calling this function
141
    void reset_approximate_stats(int64_t num_rowsets, int64_t num_segments, int64_t num_rows,
142
                                 int64_t data_size);
143
144
    // return a json string to show the compaction status of this tablet
145
    void get_compaction_status(std::string* json_result);
146
147
    // Synchronize the rowsets from meta service.
148
    // If tablet state is not `TABLET_RUNNING`, sync tablet meta and all visible rowsets.
149
    // If `query_version` > 0 and local max_version of the tablet >= `query_version`, do nothing.
150
    // If 'need_download_data_async' is true, it means that we need to download the new version
151
    // rowsets datum async.
152
    Status sync_rowsets(const SyncOptions& options = {}, SyncRowsetStats* stats = nullptr);
153
154
    // Synchronize the tablet meta from meta service.
155
    Status sync_meta();
156
157
    // If `version_overlap` is true, function will delete rowsets with overlapped version in this tablet.
158
    // If 'warmup_delta_data' is true, download the new version rowset data in background.
159
    // MUST hold EXCLUSIVE `_meta_lock`.
160
    // If 'need_download_data_async' is true, it means that we need to download the new version
161
    // rowsets datum async.
162
    void add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
163
                     std::unique_lock<BthreadSharedMutex>& meta_lock,
164
                     bool warmup_delta_data = false);
165
166
    // MUST hold EXCLUSIVE `_meta_lock`.
167
    void delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
168
                        std::unique_lock<BthreadSharedMutex>& meta_lock);
169
170
    // Like delete_rowsets, but also removes edges from the version graph.
171
    // Used by schema change to prevent the greedy capture algorithm from
172
    // preferring stale compaction rowsets over individual SC output rowsets.
173
    // MUST hold EXCLUSIVE `_meta_lock`.
174
    void delete_rowsets_for_schema_change(const std::vector<RowsetSharedPtr>& to_delete,
175
                                          std::unique_lock<BthreadSharedMutex>& meta_lock,
176
                                          bool recycle_deleted_rowsets = true);
177
178
    // Replace local rowsets in [2, alter_version] with schema change output rowsets.
179
    // Existing SC output rowsets are kept; other local/double-write/compaction rowsets
180
    // in this version range are removed from both _rs_version_map and version graph.
181
    // recycle_deleted_rowsets should only be true for the real tablet; temporary
182
    // schema-change delete-bitmap tablets only need to normalize their local graph.
183
    // MUST hold EXCLUSIVE `_meta_lock`.
184
    void replace_rowsets_with_schema_change_output(
185
            const std::vector<RowsetSharedPtr>& output_rowsets, int64_t alter_version,
186
            std::unique_lock<BthreadSharedMutex>& meta_lock, const char* stage,
187
            bool recycle_deleted_rowsets);
188
189
    // When the tablet is dropped, we need to recycle cached data:
190
    // 1. The data in file cache
191
    // 2. The memory in tablet cache
192
    void clear_cache() override;
193
194
    // Return number of deleted stale rowsets
195
    uint64_t delete_expired_stale_rowsets();
196
197
999k
    bool has_stale_rowsets() const { return !_stale_rs_version_map.empty(); }
198
199
    int64_t get_cloud_base_compaction_score() const;
200
    int64_t get_cloud_cumu_compaction_score() const;
201
202
667k
    int64_t max_version_unlocked() const override { return _max_version; }
203
962k
    int64_t base_compaction_cnt() const { return _base_compaction_cnt; }
204
961k
    int64_t cumulative_compaction_cnt() const { return _cumulative_compaction_cnt; }
205
300k
    int64_t full_compaction_cnt() const { return _full_compaction_cnt; }
206
989k
    int64_t cumulative_layer_point() const {
207
989k
        return _cumulative_point.load(std::memory_order_relaxed);
208
989k
    }
209
210
301k
    void set_base_compaction_cnt(int64_t cnt) { _base_compaction_cnt = cnt; }
211
311k
    void set_cumulative_compaction_cnt(int64_t cnt) { _cumulative_compaction_cnt = cnt; }
212
301k
    void set_full_compaction_cnt(int64_t cnt) { _full_compaction_cnt = cnt; }
213
    void set_cumulative_layer_point(int64_t new_point);
214
215
871M
    int64_t last_cumu_compaction_failure_time() { return _last_cumu_compaction_failure_millis; }
216
370
    void set_last_cumu_compaction_failure_time(int64_t millis) {
217
370
        _last_cumu_compaction_failure_millis = millis;
218
370
    }
219
220
114M
    int64_t last_base_compaction_failure_time() { return _last_base_compaction_failure_millis; }
221
7.74k
    void set_last_base_compaction_failure_time(int64_t millis) {
222
7.74k
        _last_base_compaction_failure_millis = millis;
223
7.74k
    }
224
225
0
    int64_t last_full_compaction_failure_time() { return _last_full_compaction_failure_millis; }
226
0
    void set_last_full_compaction_failure_time(int64_t millis) {
227
0
        _last_full_compaction_failure_millis = millis;
228
0
    }
229
230
61.5k
    int64_t last_cumu_compaction_success_time() { return _last_cumu_compaction_success_millis; }
231
19.3k
    void set_last_cumu_compaction_success_time(int64_t millis) {
232
19.3k
        _last_cumu_compaction_success_millis = millis;
233
19.3k
    }
234
235
9.99k
    int64_t last_base_compaction_success_time() { return _last_base_compaction_success_millis; }
236
10.0k
    void set_last_base_compaction_success_time(int64_t millis) {
237
10.0k
        _last_base_compaction_success_millis = millis;
238
10.0k
    }
239
240
9.84k
    int64_t last_full_compaction_success_time() { return _last_full_compaction_success_millis; }
241
9.90k
    void set_last_full_compaction_success_time(int64_t millis) {
242
9.90k
        _last_full_compaction_success_millis = millis;
243
9.90k
    }
244
245
0
    int64_t last_cumu_compaction_schedule_time() { return _last_cumu_compaction_schedule_millis; }
246
373
    void set_last_cumu_compaction_schedule_time(int64_t millis) {
247
373
        _last_cumu_compaction_schedule_millis = millis;
248
373
    }
249
250
0
    int64_t last_base_compaction_schedule_time() { return _last_base_compaction_schedule_millis; }
251
2
    void set_last_base_compaction_schedule_time(int64_t millis) {
252
2
        _last_base_compaction_schedule_millis = millis;
253
2
    }
254
255
0
    int64_t last_full_compaction_schedule_time() { return _last_full_compaction_schedule_millis; }
256
65
    void set_last_full_compaction_schedule_time(int64_t millis) {
257
65
        _last_full_compaction_schedule_millis = millis;
258
65
    }
259
260
204k
    void set_last_cumu_compaction_status(std::string status) {
261
204k
        _last_cumu_compaction_status = std::move(status);
262
204k
    }
263
264
0
    std::string get_last_cumu_compaction_status() { return _last_cumu_compaction_status; }
265
266
4.00k
    void set_last_base_compaction_status(std::string status) {
267
4.00k
        _last_base_compaction_status = std::move(status);
268
4.00k
    }
269
270
0
    std::string get_last_base_compaction_status() { return _last_base_compaction_status; }
271
272
130
    void set_last_full_compaction_status(std::string status) {
273
130
        _last_full_compaction_status = std::move(status);
274
130
    }
275
276
0
    std::string get_last_full_compaction_status() { return _last_full_compaction_status; }
277
278
268k
    int64_t alter_version() const { return _alter_version; }
279
86.6k
    void set_alter_version(int64_t alter_version) { _alter_version = alter_version; }
280
281
    // Whether this tablet still has an active alter (schema change) job registered
282
    // in meta-service. Only meaningful for NOT_READY tablets; refreshed on each
283
    // rowset sync (see GetRowsetResponse.has_alter_job). Defaults to true so that
284
    // a tablet is treated as alter-in-progress until the first sync tells otherwise
285
    // (also keeps behavior unchanged when MS has not been upgraded to fill it).
286
1.83M
    bool has_active_alter_job() const {
287
1.83M
        return _has_active_alter_job.load(std::memory_order_relaxed);
288
1.83M
    }
289
24.5k
    void set_has_active_alter_job(bool v) {
290
24.5k
        _has_active_alter_job.store(v, std::memory_order_relaxed);
291
24.5k
    }
292
293
    // A NOT_READY tablet whose schema change job has been cancelled or removed in
294
    // meta-service. It will never be converted or compacted, does not serve reads
295
    // or writes, and just stays in the local tablet cache until the recycler
296
    // permanently deletes its meta. It should be invisible to compaction score
297
    // stats, /api/compaction_score and compaction scheduling.
298
2.81G
    bool is_zombie_tablet() const {
299
2.81G
        return tablet_state() == TABLET_NOTREADY && !has_active_alter_job();
300
2.81G
    }
301
302
    // Why this tablet is currently not scheduled for compaction, empty if it is
303
    // schedulable. Display only (/api/compaction_score, /api/compaction/show),
304
    // evaluated on demand, never in the scheduling loop.
305
    std::string compaction_not_scheduled_reason();
306
307
    // Last active cluster info for compaction read-write separation
308
986M
    std::string last_active_cluster_id() const {
309
986M
        std::shared_lock lock(_cluster_info_mutex);
310
986M
        return _last_active_cluster_id;
311
986M
    }
312
    int64_t last_active_time_ms() const {
313
        std::shared_lock lock(_cluster_info_mutex);
314
        return _last_active_time_ms;
315
    }
316
55.4k
    void set_last_active_cluster_info(const std::string& cluster_id, int64_t time_ms) {
317
55.4k
        std::unique_lock lock(_cluster_info_mutex);
318
55.4k
        _last_active_cluster_id = cluster_id;
319
55.4k
        _last_active_time_ms = time_ms;
320
55.4k
    }
321
322
    // MUST hold SHARED `_meta_lock`.
323
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction_unlocked();
324
325
1.26M
    inline Version max_version() const {
326
1.26M
        std::shared_lock rdlock(_meta_lock);
327
1.26M
        return _tablet_meta->max_version();
328
1.26M
    }
329
330
140k
    int64_t base_size() const { return _base_size; }
331
332
    // MUST hold SHARED `_meta_lock`.
333
    std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction_unlocked();
334
    Result<RowsetSharedPtr> pick_a_rowset_for_index_change(int schema_version,
335
                                                           bool& is_base_rowset);
336
    Status check_rowset_schema_for_build_index(std::vector<TColumn>& columns, int schema_version);
337
338
0
    std::mutex& get_base_compaction_lock() { return _base_compaction_lock; }
339
0
    std::mutex& get_cumulative_compaction_lock() { return _cumulative_compaction_lock; }
340
341
    Result<std::unique_ptr<RowsetWriter>> create_transient_rowset_writer(
342
            const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
343
            int64_t txn_expiration = 0) override;
344
345
    CalcDeleteBitmapExecutor* calc_delete_bitmap_executor() override;
346
347
    Status save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
348
                              DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
349
                              const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id = -1,
350
                              int64_t next_visible_version = -1) override;
351
352
    Status save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id,
353
                                    DeleteBitmapPtr delete_bitmap, int64_t lock_id,
354
                                    int64_t next_visible_version, RowsetSharedPtr rowset);
355
356
    Status calc_delete_bitmap_for_compaction(const std::vector<RowsetSharedPtr>& input_rowsets,
357
                                             const RowsetSharedPtr& output_rowset,
358
                                             const RowIdConversion& rowid_conversion,
359
                                             ReaderType compaction_type, int64_t merged_rows,
360
                                             int64_t filtered_rows, int64_t initiator,
361
                                             DeleteBitmapPtr& output_rowset_delete_bitmap,
362
                                             bool allow_delete_in_cumu_compaction,
363
                                             int64_t& get_delete_bitmap_lock_start_time);
364
365
    // Find the missed versions until the spec_version.
366
    //
367
    // for example:
368
    //     [0-4][5-5][8-8][9-9][14-14]
369
    // if spec_version = 12, it will return [6-7],[10-12]
370
    Versions calc_missed_versions(int64_t spec_version, Versions existing_versions) const override;
371
372
56.6k
    std::mutex& get_rowset_update_lock() { return _rowset_update_lock; }
373
374
166k
    bthread::Mutex& get_sync_meta_lock() { return _sync_meta_lock; }
375
376
1.89M
    const auto& rowset_map() const { return _rs_version_map; }
377
378
    // How long since this tablet's ROWSETS were pulled from MS. Only sync_rowsets() advances
379
    // it, and only when it actually issues the RPC.
380
    int64_t last_sync_rowsets_time_s = 0;
381
    // How long since this tablet's META was pulled from MS, which is what carries properties
382
    // such as the file cache TTL. Only sync_meta() advances it. Tracked separately on
383
    // purpose: a tablet under continuous ingest keeps the rowsets clock permanently fresh,
384
    // so gating meta work on that one starves the meta refresh entirely.
385
    int64_t last_sync_tablet_meta_time_s = 0;
386
    int64_t last_load_time_ms = 0;
387
    int64_t last_base_compaction_success_time_ms = 0;
388
    int64_t last_cumu_compaction_success_time_ms = 0;
389
    int64_t last_cumu_no_suitable_version_ms = 0;
390
    int64_t last_access_time_ms = 0;
391
392
    std::atomic<int64_t> local_read_time_us = 0;
393
    std::atomic<int64_t> remote_read_time_us = 0;
394
    std::atomic<int64_t> exec_compaction_time_us = 0;
395
396
    void build_tablet_report_info(TTabletInfo* tablet_info);
397
398
    // check that if the delete bitmap in delete bitmap cache has the same cardinality with the expected_delete_bitmap's
399
    Status check_delete_bitmap_cache(int64_t txn_id, DeleteBitmap* expected_delete_bitmap) override;
400
401
    void agg_delete_bitmap_for_compaction(
402
            int64_t start_version, int64_t end_version,
403
            const std::vector<RowsetSharedPtr>& pre_rowsets, DeleteBitmapPtr& new_delete_bitmap,
404
            std::map<std::string, int64_t>& pre_rowset_to_versions,
405
            PreRowsetDeleteBitmapStats* pre_rowset_delete_bitmap_stats);
406
407
    bool need_remove_unused_rowsets();
408
409
    void add_unused_rowsets(const std::vector<RowsetSharedPtr>& rowsets);
410
    void remove_unused_rowsets();
411
412
    // For each given rowset not in active use, clears its file cache and returns its
413
    // ID, segment count, and index file names as RecycledRowsets entries.
414
    static std::vector<RecycledRowsets> recycle_cached_data(
415
            const std::vector<RowsetSharedPtr>& rowsets);
416
417
    // Add warmup state management
418
    WarmUpState get_rowset_warmup_state(RowsetId rowset_id);
419
    std::optional<uint64_t> add_rowset_warmup_state(
420
            const RowsetMeta& rowset, WarmUpTriggerSource source,
421
            std::chrono::steady_clock::time_point start_tp = std::chrono::steady_clock::now());
422
    bool update_rowset_warmup_state_inverted_idx_num(WarmUpTriggerSource source, RowsetId rowset_id,
423
                                                     int64_t delta, uint64_t generation);
424
    bool update_rowset_warmup_state_inverted_idx_num_unlocked(WarmUpTriggerSource source,
425
                                                              RowsetId rowset_id, int64_t delta,
426
                                                              uint64_t generation);
427
    WarmUpState complete_rowset_segment_warmup(WarmUpTriggerSource trigger_source,
428
                                               RowsetId rowset_id, Status status,
429
                                               int64_t segment_num, int64_t inverted_idx_num,
430
                                               uint64_t generation);
431
432
    // Warmup can arrive before metadata sync and never enter the local Rowset GC lifecycle.
433
    size_t gc_rowset_warmup_states(
434
            std::chrono::seconds grace_period,
435
            std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now());
436
437
    bool is_rowset_warmed_up(const RowsetId& rowset_id) const;
438
439
    void add_warmed_up_rowset(const RowsetId& rowset_id);
440
    // Test helper: add a rowset to the warmup state map with DOING progress,
441
    // so that is_rowset_warmed_up() returns false for it.
442
    void add_not_warmed_up_rowset(const RowsetId& rowset_id);
443
444
    // Try to apply visible pending rowsets to tablet meta in version order
445
    // This should be called after receiving FE notification or when new rowsets are added
446
    // @return Status::OK() if successfully applied, error otherwise
447
    void apply_visible_pending_rowsets();
448
449
    void try_make_committed_rs_visible(int64_t txn_id, int64_t visible_version,
450
                                       int64_t version_update_time_ms);
451
    void try_make_committed_rs_visible_for_mow(int64_t txn_id, int64_t visible_version,
452
                                               int64_t version_update_time_ms);
453
454
    void clear_unused_visible_pending_rowsets();
455
456
5
    std::string rowset_warmup_digest() const {
457
5
        std::string res;
458
104
        auto add_log = [&](const RowsetSharedPtr& rs) {
459
104
            auto tmp = fmt::format("{}{}", rs->rowset_id().to_string(), rs->version().to_string());
460
104
            if (_rowset_warm_up_states.contains(rs->rowset_id())) {
461
104
                tmp += fmt::format(
462
104
                        ", progress={}, segments_warmed_up={}/{}, inverted_idx_warmed_up={}/{}",
463
104
                        _rowset_warm_up_states.at(rs->rowset_id()).state.progress,
464
104
                        _rowset_warm_up_states.at(rs->rowset_id()).num_segments_warmed_up,
465
104
                        _rowset_warm_up_states.at(rs->rowset_id()).num_segments,
466
104
                        _rowset_warm_up_states.at(rs->rowset_id()).num_inverted_idx_warmed_up,
467
104
                        _rowset_warm_up_states.at(rs->rowset_id()).num_inverted_idx);
468
104
            }
469
104
            res += fmt::format("[{}],", tmp);
470
104
        };
471
5
        traverse_rowsets_unlocked(add_log, true);
472
5
        return res;
473
5
    }
474
475
private:
476
    // FIXME(plat1ko): No need to record base size if rowsets are ordered by version
477
    void update_base_size(const Rowset& rs);
478
479
    Status sync_if_not_running(SyncRowsetStats* stats = nullptr);
480
481
    std::optional<uint64_t> add_rowset_warmup_state_unlocked(
482
            const RowsetMeta& rowset, WarmUpTriggerSource source,
483
            std::chrono::steady_clock::time_point start_tp = std::chrono::steady_clock::now());
484
485
    // used by capture_rs_reader_xxx functions
486
    bool rowset_is_warmed_up_unlocked(int64_t start_version, int64_t end_version) const;
487
488
    // Check if a rowset should be visible but not warmed up within freshness tolerance
489
    bool _check_rowset_should_be_visible_but_not_warmed_up(
490
            const RowsetMetaSharedPtr& rs_meta, int64_t path_max_version,
491
            std::chrono::system_clock::time_point freshness_limit_tp) const;
492
493
    // Submit a segment download task for warming up
494
    void _submit_segment_download_task(const RowsetSharedPtr& rs, io::Path segment_path,
495
                                       int64_t segment_file_size, int64_t expiration_time,
496
                                       uint64_t generation);
497
498
    // Submit an inverted index download task for warming up
499
    void _submit_inverted_index_download_task(const RowsetSharedPtr& rs,
500
                                              const StorageResource* storage_resource,
501
                                              const io::Path& idx_path, int64_t idx_size,
502
                                              int64_t expiration_time, uint64_t generation);
503
504
    // Add rowsets directly with warmup
505
    void _add_rowsets_directly(std::vector<RowsetSharedPtr>& rowsets, bool warmup_delta_data);
506
507
    CloudStorageEngine& _engine;
508
509
    // this mutex MUST ONLY be used when sync meta
510
    bthread::Mutex _sync_meta_lock;
511
    // ATTENTION: lock order should be: _sync_meta_lock -> _meta_lock
512
513
    std::atomic<int64_t> _cumulative_point {-1};
514
    std::atomic<int64_t> _approximate_num_rowsets {-1};
515
    std::atomic<int64_t> _approximate_num_segments {-1};
516
    std::atomic<int64_t> _approximate_num_rows {-1};
517
    std::atomic<int64_t> _approximate_data_size {-1};
518
    std::atomic<int64_t> _approximate_cumu_num_rowsets {-1};
519
    // Number of sorted arrays (e.g. for rowset with N segments, if rowset is overlapping, delta is N, otherwise 1) after cumu point
520
    std::atomic<int64_t> _approximate_cumu_num_deltas {-1};
521
522
    // timestamp of last cumu compaction failure
523
    std::atomic<int64_t> _last_cumu_compaction_failure_millis;
524
    // timestamp of last base compaction failure
525
    std::atomic<int64_t> _last_base_compaction_failure_millis;
526
    // timestamp of last full compaction failure
527
    std::atomic<int64_t> _last_full_compaction_failure_millis;
528
    // timestamp of last cumu compaction success
529
    std::atomic<int64_t> _last_cumu_compaction_success_millis;
530
    // timestamp of last base compaction success
531
    std::atomic<int64_t> _last_base_compaction_success_millis;
532
    // timestamp of last full compaction success
533
    std::atomic<int64_t> _last_full_compaction_success_millis;
534
    // timestamp of last cumu compaction schedule time
535
    std::atomic<int64_t> _last_cumu_compaction_schedule_millis;
536
    // timestamp of last base compaction schedule time
537
    std::atomic<int64_t> _last_base_compaction_schedule_millis;
538
    // timestamp of last full compaction schedule time
539
    std::atomic<int64_t> _last_full_compaction_schedule_millis;
540
    std::string _last_cumu_compaction_status;
541
    std::string _last_base_compaction_status;
542
    std::string _last_full_compaction_status;
543
    int64_t _base_compaction_cnt = 0;
544
    int64_t _cumulative_compaction_cnt = 0;
545
    int64_t _full_compaction_cnt = 0;
546
    int64_t _max_version = -1;
547
    int64_t _base_size = 0;
548
    int64_t _alter_version = -1;
549
    std::atomic<bool> _has_active_alter_job {true};
550
551
    std::mutex _base_compaction_lock;
552
    std::mutex _cumulative_compaction_lock;
553
554
    // To avoid multiple calc delete bitmap tasks on same (txn_id, tablet_id) with different
555
    // signatures being executed concurrently, we use _rowset_update_lock to serialize them
556
    mutable std::mutex _rowset_update_lock;
557
558
    // unused_rowsets, [start_version, end_version]
559
    std::mutex _gc_mutex;
560
    std::unordered_map<RowsetId, RowsetSharedPtr> _unused_rowsets;
561
    std::vector<std::pair<std::vector<RowsetId>, DeleteBitmapKeyRanges>> _unused_delete_bitmap;
562
563
    // for warm up states management
564
    struct RowsetWarmUpInfo {
565
        WarmUpState state;
566
        uint64_t generation = 0;
567
        int64_t num_segments = 0;
568
        int64_t num_inverted_idx = 0;
569
        int64_t num_segments_warmed_up = 0;
570
        int64_t num_inverted_idx_warmed_up = 0;
571
        std::chrono::steady_clock::time_point start_tp;
572
573
61.7k
        void done(int64_t input_num_segments, int64_t input_num_inverted_idx) {
574
61.7k
            num_segments_warmed_up += input_num_segments;
575
61.7k
            num_inverted_idx_warmed_up += input_num_inverted_idx;
576
61.7k
            update_state();
577
61.7k
        }
578
579
361k
        bool has_finished() const {
580
361k
            return (num_segments_warmed_up >= num_segments) &&
581
361k
                   (num_inverted_idx_warmed_up >= num_inverted_idx);
582
361k
        }
583
584
        void update_state();
585
    };
586
    uint64_t _next_warmup_generation = 0;
587
    std::unordered_map<RowsetId, RowsetWarmUpInfo> _rowset_warm_up_states;
588
589
    mutable std::shared_mutex _warmed_up_rowsets_mutex;
590
    std::unordered_set<RowsetId> _warmed_up_rowsets;
591
592
    // Cluster info for compaction read-write separation
593
    mutable std::shared_mutex _cluster_info_mutex;
594
    std::string _last_active_cluster_id;
595
    int64_t _last_active_time_ms {0};
596
597
    // Map: version -> <rowset_meta, expiration_time>
598
    // Stores rowsets that have been notified by FE but not yet added to tablet meta
599
    // due to out-of-order notification or version discontinuity
600
    struct VisiblePendingRowset {
601
        const bool is_empty_rowset;
602
        const int64_t expiration_time; // seconds since epoch
603
        RowsetMetaSharedPtr rowset_meta;
604
605
        VisiblePendingRowset(RowsetMetaSharedPtr rowset_meta_, int64_t expiration_time_,
606
                             bool is_empty_rowset_ = false)
607
121k
                : is_empty_rowset(is_empty_rowset_),
608
121k
                  expiration_time(expiration_time_),
609
121k
                  rowset_meta(std::move(rowset_meta_)) {}
610
    };
611
    mutable std::mutex _visible_pending_rs_lock;
612
    std::map<int64_t, VisiblePendingRowset> _visible_pending_rs_map;
613
};
614
615
using CloudTabletSPtr = std::shared_ptr<CloudTablet>;
616
617
} // namespace doris