Coverage Report

Created: 2026-10-29 14:41

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/cache/file_cache_common.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
// This file is copied from
18
// https://github.com/ClickHouse/ClickHouse/blob/master/src/Interpreters/Cache/FileCache_fwd.h
19
// and modified by Doris
20
21
#pragma once
22
#include <cstdint>
23
#include <vector>
24
25
#include "common/config.h"
26
#include "core/uint128.h"
27
#include "io/io_common.h"
28
29
namespace doris::io {
30
31
inline static constexpr size_t REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS = 100 * 1024;
32
inline static constexpr size_t FILE_CACHE_MAX_FILE_BLOCK_SIZE = 1 * 1024 * 1024;
33
inline static constexpr size_t DEFAULT_NORMAL_PERCENT = 40;
34
inline static constexpr size_t DEFAULT_DISPOSABLE_PERCENT = 5;
35
inline static constexpr size_t DEFAULT_INDEX_PERCENT = 5;
36
inline static constexpr size_t DEFAULT_TTL_PERCENT = 50;
37
38
using uint128_t = UInt128;
39
40
enum FileCacheType {
41
    INDEX = 2,
42
    NORMAL = 1,
43
    DISPOSABLE = 0,
44
    TTL = 3,
45
};
46
47
47.7M
inline size_t file_cache_type_index(FileCacheType type) {
48
47.7M
    return static_cast<size_t>(type);
49
47.7M
}
50
51
std::string cache_type_to_surfix(FileCacheType type);
52
FileCacheType surfix_to_cache_type(const std::string& str);
53
54
FileCacheType string_to_cache_type(const std::string& str);
55
std::string cache_type_to_string(FileCacheType type);
56
57
struct UInt128Wrapper {
58
    uint128_t value_;
59
    [[nodiscard]] std::string to_string() const;
60
61
    UInt128Wrapper() = default;
62
2.31M
    explicit UInt128Wrapper(const uint128_t& value) : value_(value) {}
63
64
17.0M
    bool operator==(const UInt128Wrapper& other) const { return value_ == other.value_; }
65
66
546k
    uint64_t high() const { return static_cast<uint64_t>(value_ >> 64); }
67
546k
    uint64_t low() const { return static_cast<uint64_t>(value_); }
68
69
    friend std::ostream& operator<<(std::ostream& os, const UInt128Wrapper& wrapper) {
70
        os << "UInt128Wrapper(" << wrapper.high() << ", " << wrapper.low() << ")";
71
        return os;
72
    }
73
};
74
75
struct ReadStatistics {
76
    bool hit_cache = true;
77
    bool from_peer_cache = false;
78
    bool skip_cache = false;
79
    int64_t bytes_read = 0;
80
    int64_t bytes_read_from_local = 0;
81
    int64_t bytes_read_from_remote = 0;
82
    int64_t bytes_read_from_peer = 0;
83
    int64_t remote_physical_read_bytes = 0;
84
    int64_t bytes_write_into_file_cache = 0;
85
    int64_t remote_read_timer = 0;
86
    int64_t peer_read_timer = 0;
87
    int64_t remote_wait_timer = 0; // wait for other downloader
88
    int64_t local_read_timer = 0;
89
    int64_t local_write_timer = 0;
90
    int64_t read_cache_file_directly_timer = 0;
91
    int64_t cache_get_or_set_timer = 0;
92
    int64_t lock_wait_timer = 0;
93
    int64_t get_timer = 0;
94
    int64_t set_timer = 0;
95
    int64_t async_cache_write_submitted = 0;
96
    int64_t async_cache_write_rejected = 0;
97
    int64_t async_cache_write_buffer_alloc_fail = 0;
98
    int64_t async_cache_write_drop_stale_epoch = 0;
99
    // Async-write read path, query thread only: copying remote/peer data into the user buffer, and
100
    // building plus enqueueing write tasks. Remote IO itself is in remote/peer_read_timer.
101
    int64_t async_copy_to_user_timer = 0;
102
    int64_t async_submit_write_timer = 0;
103
    int64_t inflight_write_buffer_index_hit = 0;
104
    int64_t inflight_write_buffer_index_miss = 0;
105
    int64_t probe_downloaded_hit = 0;
106
    int64_t probe_downloading_hit = 0;
107
    int64_t probe_miss = 0;
108
    int64_t block_wait_success = 0;
109
    int64_t block_wait_timeout = 0;
110
};
111
112
class BlockFileCache;
113
struct FileBlocksHolder;
114
using FileBlocksHolderPtr = std::unique_ptr<FileBlocksHolder>;
115
116
struct FileCacheAllocatorBuilder {
117
    bool _is_cold_data;
118
    uint64_t _expiration_time;
119
    UInt128Wrapper _cache_hash;
120
    BlockFileCache* _cache; // Only one ref, the lifetime is owned by FileCache
121
    FileBlocksHolderPtr allocate_cache_holder(size_t offset, size_t size, int64_t tablet_id) const;
122
};
123
124
struct KeyHash {
125
21.5M
    std::size_t operator()(const UInt128Wrapper& w) const {
126
21.5M
        return util_hash::HashLen16(w.value_.low(), w.value_.high());
127
21.5M
    }
128
};
129
130
using AccessKeyAndOffset = std::pair<UInt128Wrapper, size_t>;
131
struct KeyAndOffsetHash {
132
4.20M
    std::size_t operator()(const AccessKeyAndOffset& key) const {
133
4.20M
        return KeyHash()(key.first) ^ std::hash<uint64_t>()(key.second);
134
4.20M
    }
135
};
136
137
struct KeyMeta {
138
    uint64_t expiration_time; // absolute time
139
    FileCacheType type;
140
    int64_t tablet_id {0};
141
};
142
143
struct FileCacheKey {
144
    UInt128Wrapper hash;
145
    size_t offset;
146
    KeyMeta meta;
147
};
148
149
struct FileCacheSettings {
150
    size_t capacity {0};
151
    size_t disposable_queue_size {0};
152
    size_t disposable_queue_elements {0};
153
    size_t index_queue_size {0};
154
    size_t index_queue_elements {0};
155
    size_t query_queue_size {0};
156
    size_t query_queue_elements {0};
157
    size_t ttl_queue_size {0};
158
    size_t ttl_queue_elements {0};
159
    size_t max_file_block_size {0};
160
    size_t max_query_cache_size {0};
161
    std::string storage;
162
163
    // to string
164
    std::string to_string() const;
165
};
166
167
FileCacheSettings get_file_cache_settings(size_t capacity, size_t max_query_cache_size,
168
                                          size_t normal_percent = DEFAULT_NORMAL_PERCENT,
169
                                          size_t disposable_percent = DEFAULT_DISPOSABLE_PERCENT,
170
                                          size_t index_percent = DEFAULT_INDEX_PERCENT,
171
                                          size_t ttl_percent = DEFAULT_TTL_PERCENT,
172
                                          const std::string& storage = "disk");
173
174
struct CacheContext {
175
984k
    CacheContext(const IOContext* io_context) {
176
984k
        if (io_context->expiration_time != 0) {
177
2
            cache_type = FileCacheType::TTL;
178
2
            expiration_time = io_context->expiration_time;
179
984k
        } else if (io_context->is_index_data) {
180
233k
            cache_type = FileCacheType::INDEX;
181
751k
        } else if (io_context->is_disposable) {
182
6.90k
            cache_type = FileCacheType::DISPOSABLE;
183
744k
        } else {
184
744k
            cache_type = FileCacheType::NORMAL;
185
744k
        }
186
984k
        query_id = io_context->query_id ? *io_context->query_id : TUniqueId();
187
984k
        is_warmup = io_context->is_warmup;
188
984k
        remote_scan_cache_write_limiter = io_context->remote_scan_cache_write_limiter;
189
984k
        admit_cache_write_by_remote_scan_limiter =
190
984k
                remote_scan_cache_write_limiter != nullptr &&
191
984k
                io_context->reader_type == ReaderType::READER_QUERY &&
192
984k
                (!io_context->is_index_data || io_context->is_inverted_index ||
193
9
                 config::enable_file_cache_query_limit_segment_meta) &&
194
984k
                !io_context->is_warmup;
195
984k
    }
196
68.7k
    CacheContext() = default;
197
    bool operator==(const CacheContext& rhs) const {
198
        return query_id == rhs.query_id && cache_type == rhs.cache_type &&
199
               expiration_time == rhs.expiration_time && is_cold_data == rhs.is_cold_data;
200
    }
201
    TUniqueId query_id {};
202
    FileCacheType cache_type {FileCacheType::NORMAL};
203
    int64_t expiration_time {0};
204
    bool is_cold_data {false};
205
    ReadStatistics* stats {nullptr};
206
    bool is_warmup {false};
207
    int64_t tablet_id {0};
208
    RemoteScanCacheWriteLimiter* remote_scan_cache_write_limiter = nullptr;
209
    bool admit_cache_write_by_remote_scan_limiter {false};
210
};
211
212
template <class Lock>
213
concept IsXLock = std::same_as<Lock, std::lock_guard<std::mutex>> ||
214
                  std::same_as<Lock, std::unique_lock<std::mutex>>;
215
216
class LRUQueue {
217
public:
218
3.21k
    LRUQueue() = default;
219
    LRUQueue(size_t max_size, size_t max_element_size, int64_t hot_data_interval)
220
1.58k
            : max_size(max_size),
221
1.58k
              max_element_size(max_element_size),
222
1.58k
              hot_data_interval(hot_data_interval) {}
223
224
    struct HashFileKeyAndOffset {
225
1.30M
        std::size_t operator()(const std::pair<UInt128Wrapper, size_t>& pair) const {
226
1.30M
            return KeyHash()(pair.first) + pair.second;
227
1.30M
        }
228
    };
229
230
    struct FileKeyAndOffset {
231
        UInt128Wrapper hash;
232
        size_t offset;
233
        size_t size;
234
235
        FileKeyAndOffset(const UInt128Wrapper& hash, size_t offset, size_t size)
236
630k
                : hash(hash), offset(offset), size(size) {}
237
    };
238
239
    using Iterator = typename std::list<FileKeyAndOffset>::iterator;
240
241
76.9k
    size_t get_max_size() const { return max_size; }
242
724
    size_t get_max_element_size() const { return max_element_size; }
243
244
    template <class T>
245
        requires IsXLock<T>
246
85.1k
    size_t get_capacity(T& /* cache_lock */) const {
247
85.1k
        return cache_size;
248
85.1k
    }
249
250
721
    size_t get_capacity_unsafe() const { return cache_size; }
251
252
753
    size_t get_elements_num_unsafe() const { return queue.size(); }
253
254
5.25k
    size_t get_elements_num(std::lock_guard<std::mutex>& /* cache_lock */) const {
255
5.25k
        return queue.size();
256
5.25k
    }
257
258
    Iterator add(const UInt128Wrapper& hash, size_t offset, size_t size,
259
                 std::lock_guard<std::mutex>& cache_lock);
260
    template <class T>
261
        requires IsXLock<T>
262
612k
    void remove(Iterator queue_it, T& /* cache_lock */) {
263
612k
        cache_size -= queue_it->size;
264
612k
        map.erase(std::make_pair(queue_it->hash, queue_it->offset));
265
612k
        queue.erase(queue_it);
266
612k
    }
267
268
    void move_to_end(Iterator queue_it, std::lock_guard<std::mutex>& cache_lock);
269
270
    void resize(Iterator queue_it, size_t new_size, std::lock_guard<std::mutex>& cache_lock);
271
272
    std::string to_string(std::lock_guard<std::mutex>& cache_lock) const;
273
274
    bool contains(const UInt128Wrapper& hash, size_t offset,
275
                  std::lock_guard<std::mutex>& cache_lock) const;
276
277
1.21M
    Iterator begin() { return queue.begin(); }
278
279
1.21M
    Iterator end() { return queue.end(); }
280
281
    void remove_all(std::lock_guard<std::mutex>& cache_lock);
282
283
    bool pop_front(std::lock_guard<std::mutex>& cache_lock);
284
285
    Iterator get(const UInt128Wrapper& hash, size_t offset,
286
                 std::lock_guard<std::mutex>& /* cache_lock */) const;
287
288
119k
    int64_t get_hot_data_interval() const { return hot_data_interval; }
289
290
0
    void clear(std::lock_guard<std::mutex>& cache_lock) {
291
0
        queue.clear();
292
0
        map.clear();
293
0
        cache_size = 0;
294
0
    }
295
296
    size_t levenshtein_distance_from(LRUQueue& base, std::lock_guard<std::mutex>& cache_lock);
297
298
    size_t max_size;
299
    size_t max_element_size;
300
    std::list<FileKeyAndOffset> queue;
301
    std::unordered_map<std::pair<UInt128Wrapper, size_t>, Iterator, HashFileKeyAndOffset> map;
302
    size_t cache_size = 0;
303
    int64_t hot_data_interval {0};
304
};
305
struct FileCacheInfo {
306
    UInt128Wrapper hash {0};
307
    uint64_t expiration_time {0};
308
    uint64_t size {0};
309
    size_t offset {0};
310
    bool is_tmp {false};
311
    FileCacheType cache_type {NORMAL};
312
313
    std::string to_string() const;
314
};
315
316
class InconsistencyType {
317
    uint32_t type;
318
319
public:
320
    enum : uint32_t {
321
        // No anomaly
322
        NONE = 0,
323
        // Missing a block cache metadata in _files
324
        NOT_LOADED = 1 << 0,
325
        // A block cache is missing in storage
326
        MISSING_IN_STORAGE = 1 << 1,
327
        // Size of a block cache recorded in _files is inconsistent with the storage
328
        SIZE_INCONSISTENT = 1 << 2,
329
        // Cache type of a block cache recorded in _files is inconsistent with the storage
330
        CACHE_TYPE_INCONSISTENT = 1 << 3,
331
        // Expiration time of a block cache recorded in _files is inconsistent with the storage
332
        EXPIRATION_TIME_INCONSISTENT = 1 << 4,
333
        // File in storage has a _tmp suffix, but the state of block cache in _files is not set to downloading
334
        TMP_FILE_EXPECT_DOWNLOADING_STATE = 1 << 5
335
    };
336
0
    InconsistencyType(uint32_t t = 0) : type(t) {}
337
0
    operator uint32_t&() { return type; }
338
339
    std::string to_string() const;
340
};
341
342
struct InconsistencyContext {
343
    // The infos in _files of BlockFileCache.
344
    std::vector<FileCacheInfo> infos_in_manager;
345
    std::vector<FileCacheInfo> infos_in_storage;
346
    std::vector<InconsistencyType> types;
347
};
348
349
std::optional<int64_t> get_tablet_id(std::string file_path);
350
351
} // namespace doris::io