Coverage Report

Created: 2026-08-06 11:40

/root/doris/cloud/src/recycler/checker.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 "resource-manager/resource_manager.h"
21
#if defined(USE_LIBCPP) && _LIBCPP_ABI_VERSION <= 1
22
#define _LIBCPP_ABI_INCOMPLETE_TYPES_IN_DEQUE
23
#endif
24
#include <atomic>
25
#include <condition_variable>
26
#include <cstdint>
27
#include <deque>
28
#include <functional>
29
#include <memory>
30
#include <string_view>
31
#include <thread>
32
#include <unordered_map>
33
#include <unordered_set>
34
35
#include "meta-store/txn_kv_error.h"
36
#include "recycler/storage_vault_accessor.h"
37
#include "recycler/white_black_list.h"
38
#include "snapshot/snapshot_manager.h"
39
40
namespace doris {
41
class RowsetMetaCloudPB;
42
} // namespace doris
43
44
namespace doris::cloud {
45
class StorageVaultAccessor;
46
class InstanceChecker;
47
class TxnKv;
48
class InstanceInfoPB;
49
50
struct PendingTableStreamDrop {
51
    int64_t base_db_id;
52
    int64_t base_table_id;
53
    int64_t stream_db_id;
54
55
    bool matches(int64_t offset_base_db_id, int64_t offset_base_table_id,
56
2
                 int64_t offset_stream_db_id) const {
57
2
        return base_db_id == offset_base_db_id && base_table_id == offset_base_table_id &&
58
2
               stream_db_id == offset_stream_db_id;
59
2
    }
60
};
61
62
TxnErrorCode collect_pending_table_stream_drops(
63
        const std::shared_ptr<TxnKv>& txn_kv, std::string_view instance_id,
64
        std::unordered_map<int64_t, PendingTableStreamDrop>* pending_drops);
65
66
class Checker {
67
public:
68
    explicit Checker(std::shared_ptr<TxnKv> txn_kv);
69
    ~Checker();
70
71
    int start();
72
73
    void stop();
74
323
    bool stopped() const { return stopped_.load(std::memory_order_acquire); }
75
76
private:
77
    void lease_check_jobs();
78
    void inspect_instance_check_interval();
79
    void do_inspect(const InstanceInfoPB& instance);
80
81
private:
82
    friend class RecyclerServiceImpl;
83
84
    std::shared_ptr<TxnKv> txn_kv_;
85
    std::atomic_bool stopped_ {false};
86
    std::string ip_port_;
87
    std::vector<std::thread> workers_;
88
89
    std::mutex mtx_;
90
    // notify check workers
91
    std::condition_variable pending_instance_cond_;
92
    std::deque<InstanceInfoPB> pending_instance_queue_;
93
    // instance_id -> enqueue_timestamp
94
    std::unordered_map<std::string, long> pending_instance_map_;
95
    std::unordered_map<std::string, std::shared_ptr<InstanceChecker>> working_instance_map_;
96
    // notify instance scanner and lease thread
97
    std::condition_variable notifier_;
98
99
    WhiteBlackList instance_filter_;
100
};
101
102
class InstanceChecker {
103
public:
104
    explicit InstanceChecker(std::shared_ptr<TxnKv> txn_kv, const std::string& instance_id);
105
    // Return 0 if success, otherwise error
106
    int init(const InstanceInfoPB& instance);
107
    // Return 0 if success.
108
    // Return 1 if data leak is identified.
109
    // Return negative if a temporary error occurred during the check process.
110
    int do_inverted_check();
111
112
    // Return 0 if success.
113
    // Return 1 if data loss is identified.
114
    // Return negative if a temporary error occurred during the check process.
115
    int do_check();
116
117
    // Return 0 if success.
118
    // Return 1 if delete bitmap leak is identified.
119
    // Return negative if a temporary error occurred during the check process.
120
    int do_delete_bitmap_inverted_check();
121
122
    // version = 1 : https://github.com/apache/doris/pull/40204
123
    // checks if https://github.com/apache/doris/pull/40204 works as expected
124
    // the stale delete bitmap will be cleared in MS when BE delete expired stale rowsets
125
    // NOTE: stale rowsets will be lost after BE restarts, so there may be some stale delete bitmaps
126
    // which will not be cleared.
127
    // version = 2 : https://github.com/apache/doris/pull/49822
128
    int do_delete_bitmap_storage_optimize_check(int version = 2);
129
130
    int do_mow_job_key_check();
131
132
    int do_tablet_stats_key_check();
133
134
    int do_restore_job_check();
135
136
    int do_txn_key_check();
137
138
    // check table and partition version key
139
    // table version should be greater than the versions of all its partitions
140
    // Return 0 if success, otherwise error
141
    int do_version_key_check();
142
143
    // Return 0 if success.
144
    // Return 1 if meta rowset key leak or loss is identified.
145
    // Return negative if a temporary error occurred during the check process.
146
    int do_meta_rowset_key_check();
147
148
    // Return 0 if success.
149
    // Return 1 if snapshot key and file leak or loss is identified.
150
    // Return negative if a temporary error occurred during the check process.
151
    int do_snapshots_check();
152
153
    // Return 0 if success.
154
    // Return 1 if mvcc meta key and data leak or loss is identified.
155
    // Return negative if a temporary error occurred during the check process.
156
    int do_mvcc_meta_key_check();
157
158
    // Return 0 if all Table Stream mappings and offsets are consistent.
159
    // Return 1 if an inconsistent mapping or offset is identified.
160
    // Return negative if a temporary error occurred during the check process.
161
    int do_table_stream_check();
162
163
    // Return 0 if success.
164
    // Return 1 if packed file metadata leak or loss is identified.
165
    // Return negative if a temporary error occurred during the check process.
166
    int do_packed_file_check();
167
168
    StorageVaultAccessor* get_accessor(const std::string& id);
169
170
0
    ResourceManager* resource_mgr() const { return resource_mgr_.get(); }
171
172
    void get_all_accessor(std::vector<StorageVaultAccessor*>* accessors);
173
174
0
    std::string_view instance_id() const { return instance_id_; }
175
176
0
    void TEST_add_accessor(std::string_view id, std::shared_ptr<StorageVaultAccessor> accessor) {
177
0
        accessor_map_.insert({std::string(id), std::move(accessor)});
178
0
    }
179
180
    // If there are multiple buckets, return the minimum lifecycle; if there are no buckets (i.e.
181
    // all accessors are HdfsAccessor), return INT64_MAX.
182
    // Return 0 if success, otherwise error
183
    int get_bucket_lifecycle(int64_t* lifecycle_days);
184
0
    void stop() { stopped_.store(true, std::memory_order_release); }
185
6.02k
    bool stopped() const { return stopped_.load(std::memory_order_acquire); }
186
187
private:
188
    struct RowsetIndexesFormatV1 {
189
        std::string rowset_id;
190
        std::unordered_set<int64_t> segment_ids;
191
        std::unordered_set<std::string> index_ids;
192
    };
193
194
    struct RowsetIndexesFormatV2 {
195
        std::string rowset_id;
196
        std::unordered_set<int64_t> segment_ids;
197
    };
198
199
private:
200
    // returns 0 for success otherwise error
201
    int init_obj_store_accessors(const InstanceInfoPB& instance);
202
203
    // returns 0 for success otherwise error
204
    int init_storage_vault_accessors(const InstanceInfoPB& instance);
205
206
    int traverse_mow_tablet(const std::function<int(int64_t, bool)>& check_func);
207
    int traverse_rowset_delete_bitmaps(
208
            int64_t tablet_id, std::string rowset_id,
209
            const std::function<int(int64_t, std::string_view, int64_t, int64_t)>& callback);
210
    int collect_tablet_rowsets(
211
            int64_t tablet_id,
212
            const std::function<void(const doris::RowsetMetaCloudPB&)>& collect_cb);
213
    int collect_unexpired_job_tmp_rowsets(
214
            std::unordered_map<int64_t, std::unordered_set<std::string>>& tmp_rowsets);
215
    int get_pending_delete_bitmap_keys(int64_t tablet_id,
216
                                       std::unordered_set<std::string>& pending_delete_bitmaps);
217
    int check_delete_bitmap_storage_optimize_v2(int64_t tablet_id, bool has_sequence_col,
218
                                                int64_t& abnormal_rowsets_num);
219
220
    int check_inverted_index_file_storage_format_v1(int64_t tablet_id, const std::string& file_path,
221
                                                    const std::string& rowset_info,
222
                                                    RowsetIndexesFormatV1& rowset_index_cache_v1);
223
224
    int check_inverted_index_file_storage_format_v2(int64_t tablet_id, const std::string& file_path,
225
                                                    const std::string& rowset_info,
226
                                                    RowsetIndexesFormatV2& rowset_index_cache_v2);
227
228
    // Return 0 if success.
229
    // Return 1 if key loss is abnormal.
230
    // Return negative if a temporary error occurred during the check process.
231
    int check_stats_tablet_key(std::string_view key, std::string_view value);
232
233
    // Return 0 if success.
234
    // Return 1 if key loss is identified.
235
    // Return negative if a temporary error occurred during the check process.
236
    int check_stats_tablet_key_exists(std::string_view key, std::string_view value);
237
238
    // Return 0 if success.
239
    // Return 1 if key leak is identified.
240
    // Return negative if a temporary error occurred during the check process.
241
    int check_stats_tablet_key_leaked(std::string_view key, std::string_view value);
242
    int check_txn_info_key(std::string_view key, std::string_view value);
243
244
    int check_txn_label_key(std::string_view key, std::string_view value);
245
246
    int check_txn_index_key(std::string_view key, std::string_view value);
247
248
    int check_txn_running_key(std::string_view key, std::string_view value);
249
250
    // Only check whether the meta rowset key is leak
251
    // in do_inverted_check() function, check whether the key is lost by comparing data file with key
252
    // Return 0 if success.
253
    // Return 1 if meta rowset key leak is identified.
254
    // Return negative if a temporary error occurred during the check process.
255
    int check_meta_rowset_key(std::string_view key, std::string_view value);
256
257
    // if TxnInfoKey's finish time > current time, it should not find tmp rowset
258
    // Return 0 if success.
259
    // Return 1 if meta tmp rowset key is abnormal.
260
    // Return negative if a temporary error occurred during the check process.
261
    int check_meta_tmp_rowset_key(std::string_view key, std::string_view value);
262
263
    /**
264
     * It is used to scan the key in the range from start_key to end_key 
265
     * and then perform handle operations on each group of kv
266
     * 
267
     * @param start_key Range begining. Note that this function will modify the `start_key`
268
     * @param end_key Range ending
269
     * @param handle_kv Operations on kv
270
     * @return code int 0 for success to scan and hanle, 1 for success to scan but handle abnormally, -1 for failed to handle 
271
     */
272
    int scan_and_handle_kv(std::string& start_key, const std::string& end_key,
273
                           std::function<int(std::string_view, std::string_view)> handle_kv);
274
275
    std::atomic_bool stopped_ {false};
276
    std::shared_ptr<TxnKv> txn_kv_;
277
    std::string instance_id_;
278
    // id -> accessor
279
    std::unordered_map<std::string, std::shared_ptr<StorageVaultAccessor>> accessor_map_;
280
    std::shared_ptr<SnapshotManager> snapshot_manager_;
281
    std::shared_ptr<ResourceManager> resource_mgr_;
282
    bool table_stream_versioned_write_ {false};
283
};
284
285
} // namespace doris::cloud