Coverage Report

Created: 2026-08-03 14:48

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/cloud/cloud_meta_mgr.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
#pragma once
18
19
#include <gen_cpp/olap_file.pb.h>
20
21
#include <future>
22
#include <memory>
23
#include <string>
24
#include <tuple>
25
#include <variant>
26
#include <vector>
27
28
#include "cloud/cloud_tablet.h"
29
#include "common/status.h"
30
#include "storage/rowset/rowset_fwd.h"
31
#include "storage/rowset/rowset_meta.h"
32
#include "util/s3_util.h"
33
34
namespace doris {
35
36
class DeleteBitmap;
37
class StreamLoadContext;
38
class CloudTablet;
39
class TabletMeta;
40
class TabletSchema;
41
class TabletMetaPB;
42
class RowsetMeta;
43
44
namespace cloud {
45
46
class FinishTabletJobResponse;
47
class StartTabletJobResponse;
48
class TabletJobInfoPB;
49
class TabletStatsPB;
50
class TabletIndexPB;
51
class HostLevelMSRpcRateLimiters;
52
class MSBackpressureHandler;
53
54
using StorageVaultInfos = std::vector<
55
        std::tuple<std::string, std::variant<S3Conf, HdfsVaultInfo>, StorageVaultPB_PathFormat>>;
56
57
// run tasks in bthread with concurrency and wait until all tasks done
58
// it stops running tasks if there are any tasks return !ok, leaving some tasks untouched
59
// return OK if all tasks successfully done, otherwise return the result of the failed task
60
Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int concurrency);
61
62
// An async wrap of `bthread_fork_join` declared previously using promise-future
63
// return OK if fut successfully created, otherwise return error
64
Status bthread_fork_join(std::vector<std::function<Status()>>&& tasks, int concurrency,
65
                         std::future<Status>* fut);
66
67
// Returns the exact actual_code when recognized, otherwise the legacy-compatible code.
68
// Exposed for unit tests.
69
MetaServiceCode get_response_code(const MetaServiceResponseStatus& status);
70
71
class CloudMetaMgr {
72
public:
73
248
    CloudMetaMgr() = default;
74
    ~CloudMetaMgr() = default;
75
    CloudMetaMgr(const CloudMetaMgr&) = delete;
76
    CloudMetaMgr& operator=(const CloudMetaMgr&) = delete;
77
78
    Status get_tablet_meta(int64_t tablet_id, std::shared_ptr<TabletMeta>* tablet_meta);
79
80
    Status sync_tablet_rowsets(CloudTablet* tablet, const SyncOptions& options = {},
81
                               SyncRowsetStats* sync_stats = nullptr);
82
    Status sync_tablet_rowsets_unlocked(
83
            CloudTablet* tablet, std::unique_lock<bthread::Mutex>& lock /* _sync_meta_lock */,
84
            const SyncOptions& options = {}, SyncRowsetStats* sync_stats = nullptr);
85
86
    Status prepare_rowset(const RowsetMeta& rs_meta, const std::string& job_id, int64_t table_id,
87
                          std::shared_ptr<RowsetMeta>* existed_rs_meta = nullptr);
88
89
    Status commit_rowset(RowsetMeta& rs_meta, const std::string& job_id, int64_t table_id,
90
                         std::shared_ptr<RowsetMeta>* existed_rs_meta = nullptr);
91
    void cache_committed_rowset(RowsetMetaSharedPtr rs_meta, int64_t expiration_time);
92
93
    Status update_tmp_rowset(const RowsetMeta& rs_meta, int64_t table_id);
94
95
    Status update_packed_file_info(const std::string& packed_file_path,
96
                                   const cloud::PackedFileInfoPB& packed_file_info,
97
                                   int64_t table_id);
98
99
    Status commit_txn(const StreamLoadContext& ctx, bool is_2pc);
100
101
    Status abort_txn(const StreamLoadContext& ctx);
102
103
    Status precommit_txn(const StreamLoadContext& ctx);
104
105
    /**
106
     * Prepares a restore job for a tablet to meta-service
107
     * Change the state to PREPARED
108
     * PREPARED state means the meta of tablet has been uploaded but not finalized.
109
     */
110
    Status prepare_restore_job(const TabletMetaPB& tablet_meta);
111
112
    /**
113
     * Commits a restore job for a tablet to meta-service
114
     * Change the state from PREPARED to COMMITTED
115
     * COMMITTED state means the meta of tablet has been finalized.
116
     */
117
    Status commit_restore_job(const int64_t tablet_id);
118
119
    /**
120
     * Finish a restore job for a tablet from meta-service
121
     * Change the state to final state.
122
     * If is_completed = true, change the state from COMMITTED to COMPLETED
123
     * If is_completed = false, change the state to from PREPARED/COMMITTED to DROPPED
124
     * COMPLETED state means the job is finished, the restored data should be visible.
125
     * DROPPED state means the job is aborted.
126
     * COMPLETED/DROPPED are the final states, jobs with final states will be recycled.
127
     */
128
    Status finish_restore_job(const int64_t tablet_id, bool is_completed);
129
130
    /**
131
     * Gets storage vault (storage backends) from meta-service
132
     * 
133
     * @param vault_info output param, all storage backends
134
     * @param is_vault_mode output param, true for pure vault mode, false for legacy mode
135
     * @return status
136
     */
137
    Status get_storage_vault_info(StorageVaultInfos* vault_infos, bool* is_vault_mode);
138
139
    Status prepare_tablet_job(const TabletJobInfoPB& job, StartTabletJobResponse* res);
140
141
    Status commit_tablet_job(const TabletJobInfoPB& job, FinishTabletJobResponse* res);
142
143
    Status abort_tablet_job(const TabletJobInfoPB& job);
144
145
    Status lease_tablet_job(const TabletJobInfoPB& job);
146
147
    Status update_delete_bitmap(const CloudTablet& tablet, int64_t lock_id, int64_t initiator,
148
                                DeleteBitmap* delete_bitmap, DeleteBitmap* delete_bitmap_v2,
149
                                std::string rowset_id,
150
                                std::optional<StorageResource> storage_resource,
151
                                int64_t store_version, int64_t table_id, int64_t txn_id = -1,
152
                                bool is_explicit_txn = false, int64_t next_visible_version = -1);
153
154
    Status cloud_update_delete_bitmap_without_lock(
155
            const CloudTablet& tablet, DeleteBitmap* delete_bitmap,
156
            std::map<std::string, int64_t>& rowset_to_versions, int64_t table_id,
157
            int64_t pre_rowset_agg_start_version = 0, int64_t pre_rowset_agg_end_version = 0);
158
159
    Status get_delete_bitmap_update_lock(const CloudTablet& tablet, int64_t lock_id,
160
                                         int64_t initiator);
161
162
    void remove_delete_bitmap_update_lock(int64_t table_id, int64_t lock_id, int64_t initiator,
163
                                          int64_t tablet_id);
164
165
    // Fill version holes by creating empty rowsets for missing versions
166
    Status fill_version_holes(CloudTablet* tablet, int64_t max_version,
167
                              std::unique_lock<BthreadSharedMutex>& wlock);
168
169
    // Create an empty rowset to fill a version hole
170
    Status create_empty_rowset_for_hole(CloudTablet* tablet, int64_t version,
171
                                        RowsetMetaSharedPtr prev_rowset_meta,
172
                                        RowsetSharedPtr* rowset);
173
174
    Status list_snapshot(std::vector<SnapshotInfoPB>& snapshots);
175
    Status get_snapshot_properties(SnapshotSwitchStatus& switch_status,
176
                                   int64_t& max_reserved_snapshots,
177
                                   int64_t& snapshot_interval_seconds);
178
179
    // Get all cluster status for the instance
180
    // Returns cluster_id -> (status, mtime_ms)
181
    // If my_cluster_id is not null, also returns the requesting node's cluster_id
182
    Status get_cluster_status(std::unordered_map<std::string, std::pair<int32_t, int64_t>>* result,
183
                              std::string* my_cluster_id = nullptr);
184
185
1
    void set_host_level_ms_rpc_rate_limiters(HostLevelMSRpcRateLimiters* limiters) {
186
1
        host_level_ms_rpc_rate_limiters_ = limiters;
187
1
    }
188
189
1
    void set_ms_backpressure_handler(MSBackpressureHandler* handler) {
190
1
        ms_backpressure_handler_ = handler;
191
1
    }
192
193
private:
194
    bool sync_tablet_delete_bitmap_by_cache(CloudTablet* tablet, std::ranges::range auto&& rs_metas,
195
                                            DeleteBitmap* delete_bitmap);
196
197
    Status sync_tablet_delete_bitmap(CloudTablet* tablet, int64_t old_max_version,
198
                                     std::ranges::range auto&& rs_metas, const TabletStatsPB& stats,
199
                                     const TabletIndexPB& idx, DeleteBitmap* delete_bitmap,
200
                                     bool full_sync, SyncRowsetStats* sync_stats,
201
                                     int32_t read_version, bool full_sync_v2);
202
    Status _read_tablet_delete_bitmap_v2(CloudTablet* tablet, int64_t old_max_version,
203
                                         std::ranges::range auto&& rs_metas,
204
                                         DeleteBitmap* delete_bitmap, GetDeleteBitmapResponse& res,
205
                                         int64_t& remote_delete_bitmap_bytes, bool full_sync_v2);
206
    Status _log_mow_delete_bitmap(CloudTablet* tablet, GetRowsetResponse& resp,
207
                                  DeleteBitmap& delete_bitmap, int64_t old_max_version,
208
                                  bool full_sync, int32_t read_version);
209
    Status _check_delete_bitmap_v2_correctness(CloudTablet* tablet, GetRowsetRequest& req,
210
                                               GetRowsetResponse& resp, int64_t old_max_version);
211
212
    Status _get_delete_bitmap_from_ms(GetDeleteBitmapRequest& req, GetDeleteBitmapResponse& res);
213
    Status _get_delete_bitmap_from_ms_by_batch(GetDeleteBitmapRequest& req,
214
                                               GetDeleteBitmapResponse& res,
215
                                               int64_t bytes_threadhold);
216
217
    void check_table_size_correctness(RowsetMeta& rs_meta);
218
    int64_t get_segment_file_size(RowsetMeta& rs_meta);
219
    int64_t get_inverted_index_file_size(RowsetMeta& rs_meta);
220
221
    HostLevelMSRpcRateLimiters* host_level_ms_rpc_rate_limiters_ {nullptr};
222
    MSBackpressureHandler* ms_backpressure_handler_ {nullptr};
223
};
224
225
} // namespace cloud
226
} // namespace doris