This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new e9b337360c8 [fix](be) Restore file-level cache metadata loading during
startup (#69116)
e9b337360c8 is described below
commit e9b337360c87917a57c49ca5bbe696f0de526669
Author: deardeng <[email protected]>
AuthorDate: Fri Oct 16 16:15:53 2026 +0800
[fix](be) Restore file-level cache metadata loading during startup (#69116)
Related PR: #57072, #65658
Problem Summary: During BE startup, get_or_set, the fully-covered
lookup, and probe attempt to restore an absent file before background
metadata loading completes. Their FileCacheKey leaves offset
uninitialized, while the storage implementation now performs an exact
RocksDB lookup for one block. A file with multiple persisted blocks can
therefore miss existing cache data or restore only one block. Query
metadata by tablet and file hash prefix, restore all blocks, and assert
that the file hash is absent when entering synchronous restoration. Keep
caller trigger conditions, key encoding, and existing tablet-range
queries unchanged.
Fix reuse of persistent file cache blocks while BE startup metadata
loading is in progress.
---
be/src/cloud/cloud_internal_service.cpp | 4 +
be/src/io/cache/block_file_cache.cpp | 21 +--
be/src/io/cache/cache_block_meta_store.cpp | 54 ++++---
be/src/io/cache/cache_block_meta_store.h | 4 +
be/src/io/cache/file_cache_storage.h | 7 +-
be/src/io/cache/fs_file_cache_storage.cpp | 41 +++--
be/src/io/cache/fs_file_cache_storage.h | 5 +-
be/src/io/cache/mem_file_cache_storage.cpp | 6 +-
be/src/io/cache/mem_file_cache_storage.h | 5 +-
.../io/cache/block_file_cache_test_meta_store.cpp | 180 ++++++++++++++++++++-
be/test/io/cache/cache_block_meta_store_test.cpp | 37 ++++-
.../cache/cached_remote_file_reader_peer_test.cpp | 84 +++++++++-
12 files changed, 379 insertions(+), 69 deletions(-)
diff --git a/be/src/cloud/cloud_internal_service.cpp
b/be/src/cloud/cloud_internal_service.cpp
index e8057e6fe20..24274a49451 100644
--- a/be/src/cloud/cloud_internal_service.cpp
+++ b/be/src/cloud/cloud_internal_service.cpp
@@ -724,6 +724,10 @@ Status handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* request
io::CacheContext ctx {};
io::ReadStatistics local_stats;
ctx.stats = &local_stats;
+ // Startup restore needs the caller's tablet; older clients omit
rowset_meta.
+ if (request->has_rowset_meta()) {
+ ctx.tablet_id = request->rowset_meta().tablet_id();
+ }
const size_t file_size =
request->has_file_size()
? static_cast<size_t>(std::max<int64_t>(0,
request->file_size()))
diff --git a/be/src/io/cache/block_file_cache.cpp
b/be/src/io/cache/block_file_cache.cpp
index 0e4b51ad161..f0b9164d93d 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -705,12 +705,7 @@ FileBlocks BlockFileCache::get_impl(const UInt128Wrapper&
hash, const CacheConte
if (_async_open_done) {
return {};
}
- FileCacheKey key;
- key.hash = hash;
- key.meta.type = context.cache_type;
- key.meta.expiration_time = context.expiration_time;
- key.meta.tablet_id = context.tablet_id;
- _storage->load_blocks_directly_unlocked(this, key, cache_lock);
+ _storage->load_file_blocks_directly_unlocked(this, hash,
context.tablet_id, cache_lock);
it = _files.find(hash);
if (it == _files.end()) [[unlikely]] {
@@ -816,12 +811,7 @@ Status
BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
if (_async_open_done) {
return Status::OK();
}
- FileCacheKey key;
- key.hash = hash;
- key.meta.type = context.cache_type;
- key.meta.expiration_time = context.expiration_time;
- key.meta.tablet_id = context.tablet_id;
- _storage->load_blocks_directly_unlocked(this, key, cache_lock);
+ _storage->load_file_blocks_directly_unlocked(this, hash,
context.tablet_id, cache_lock);
it = _files.find(hash);
if (it == _files.end()) {
@@ -897,12 +887,7 @@ FileBlocksProbeResult BlockFileCache::probe(const
UInt128Wrapper& hash, size_t o
auto file_iterator = _files.find(hash);
if (file_iterator == _files.end() && !_async_open_done) {
- FileCacheKey key;
- key.hash = hash;
- key.meta.type = context.cache_type;
- key.meta.expiration_time = context.expiration_time;
- key.meta.tablet_id = context.tablet_id;
- _storage->load_blocks_directly_unlocked(this, key, cache_lock);
+ _storage->load_file_blocks_directly_unlocked(this, hash,
context.tablet_id, cache_lock);
file_iterator = _files.find(hash);
}
diff --git a/be/src/io/cache/cache_block_meta_store.cpp
b/be/src/io/cache/cache_block_meta_store.cpp
index fa82cee3ce9..2e1f644dd19 100644
--- a/be/src/io/cache/cache_block_meta_store.cpp
+++ b/be/src/io/cache/cache_block_meta_store.cpp
@@ -38,6 +38,26 @@
#include "util/threadpool.h"
namespace doris::io {
+namespace {
+
+std::string serialize_key_prefix(int64_t tablet_id) {
+ std::string prefix;
+ prefix.push_back(0x1); // version byte
+
get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT)->full_encode_ascending(&tablet_id,
&prefix);
+ return prefix;
+}
+
+std::string serialize_key_prefix(int64_t tablet_id, const UInt128Wrapper&
hash) {
+ auto prefix = serialize_key_prefix(tablet_id);
+ const auto* coder = get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT);
+ uint64_t hash_high = hash.high();
+ uint64_t hash_low = hash.low();
+ coder->full_encode_ascending(&hash_high, &prefix);
+ coder->full_encode_ascending(&hash_low, &prefix);
+ return prefix;
+}
+
+} // namespace
const std::string FILE_CACHE_META_COLUMN_FAMILY = "file_cache_meta";
@@ -170,12 +190,16 @@ std::optional<BlockMeta> CacheBlockMetaStore::get(const
BlockMetaKey& key) {
}
std::unique_ptr<BlockMetaIterator> CacheBlockMetaStore::range_get(int64_t
tablet_id) {
- // Generate prefix using new serialization format
- std::string prefix;
- prefix.push_back(0x1); // version byte
- auto* tablet_id_coder = get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT);
- tablet_id_coder->full_encode_ascending(&tablet_id, &prefix);
+ return range_get_by_prefix(serialize_key_prefix(tablet_id));
+}
+std::unique_ptr<BlockMetaIterator> CacheBlockMetaStore::range_get(int64_t
tablet_id,
+ const
UInt128Wrapper& hash) {
+ return range_get_by_prefix(serialize_key_prefix(tablet_id, hash));
+}
+
+std::unique_ptr<BlockMetaIterator> CacheBlockMetaStore::range_get_by_prefix(
+ const std::string& prefix) {
class RocksDBIterator : public BlockMetaIterator {
public:
RocksDBIterator(rocksdb::Iterator* iter, const std::string& prefix)
@@ -459,22 +483,8 @@ void CacheBlockMetaStore::async_write_worker() {
}
std::string serialize_key(const BlockMetaKey& key) {
- std::string result;
- // Add version byte
- result.push_back(0x1);
-
- // Encode tablet_id using KeyCoderTraits
- auto* tablet_id_coder = get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT);
- tablet_id_coder->full_encode_ascending(&key.tablet_id, &result);
-
- // Encode hash high and low parts
- uint64_t hash_high = key.hash.high();
- uint64_t hash_low = key.hash.low();
- tablet_id_coder->full_encode_ascending(&hash_high, &result);
- tablet_id_coder->full_encode_ascending(&hash_low, &result);
-
- // Encode offset
- tablet_id_coder->full_encode_ascending(&key.offset, &result);
+ auto result = serialize_key_prefix(key.tablet_id, key.hash);
+
get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT)->full_encode_ascending(&key.offset,
&result);
return result;
}
@@ -609,4 +619,4 @@ std::optional<BlockMeta> deserialize_value(std::string_view
value_view, Status*
return std::nullopt;
}
-} // namespace doris::io
\ No newline at end of file
+} // namespace doris::io
diff --git a/be/src/io/cache/cache_block_meta_store.h
b/be/src/io/cache/cache_block_meta_store.h
index d480b4a9d78..9931c5c5b61 100644
--- a/be/src/io/cache/cache_block_meta_store.h
+++ b/be/src/io/cache/cache_block_meta_store.h
@@ -99,6 +99,9 @@ public:
// Range query all BlockMeta for specified tablet_id
std::unique_ptr<BlockMetaIterator> range_get(int64_t tablet_id);
+ // Range query all BlockMeta for a file in the specified tablet.
+ std::unique_ptr<BlockMetaIterator> range_get(int64_t tablet_id, const
UInt128Wrapper& hash);
+
// Get iterator for all BlockMeta records
std::unique_ptr<BlockMetaIterator> get_all();
@@ -116,6 +119,7 @@ public:
private:
void async_write_worker();
+ std::unique_ptr<BlockMetaIterator> range_get_by_prefix(const std::string&
prefix);
std::string _db_path;
std::unique_ptr<rocksdb::DB> _db;
diff --git a/be/src/io/cache/file_cache_storage.h
b/be/src/io/cache/file_cache_storage.h
index 04dd1398414..95b6fc59f6b 100644
--- a/be/src/io/cache/file_cache_storage.h
+++ b/be/src/io/cache/file_cache_storage.h
@@ -79,9 +79,10 @@ public:
// change the block meta
virtual Status change_key_meta_type(const FileCacheKey& key, const
FileCacheType type,
const size_t size) = 0;
- // use when lazy load cache
- virtual void load_blocks_directly_unlocked(BlockFileCache* _mgr, const
FileCacheKey& key,
- std::lock_guard<std::mutex>&
cache_lock) {}
+ // Restore a file while lazy loading, with its hash absent from the cache.
+ virtual void load_file_blocks_directly_unlocked(BlockFileCache* mgr, const
UInt128Wrapper& hash,
+ int64_t tablet_id,
+
std::lock_guard<std::mutex>& cache_lock) {}
// force clear all current data in the cache
virtual Status clear(std::string& msg) = 0;
virtual FileCacheStorageType get_type() = 0;
diff --git a/be/src/io/cache/fs_file_cache_storage.cpp
b/be/src/io/cache/fs_file_cache_storage.cpp
index 4ca6f13462a..289aed50d70 100644
--- a/be/src/io/cache/fs_file_cache_storage.cpp
+++ b/be/src/io/cache/fs_file_cache_storage.cpp
@@ -42,6 +42,7 @@
#include <unordered_set>
#include <vector>
+#include "common/check.h"
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
@@ -1157,26 +1158,34 @@ void
FSFileCacheStorage::load_cache_info_into_memory(BlockFileCache* mgr) const
}
}
-void FSFileCacheStorage::load_blocks_directly_unlocked(BlockFileCache* mgr,
const FileCacheKey& key,
-
std::lock_guard<std::mutex>& cache_lock) {
- BlockMetaKey mkey(key.meta.tablet_id, UInt128Wrapper(key.hash),
key.offset);
- auto block_meta = _meta_store->get(mkey);
- if (!block_meta.has_value()) {
- // cache miss
+void FSFileCacheStorage::load_file_blocks_directly_unlocked(
+ BlockFileCache* mgr, const UInt128Wrapper& hash, int64_t tablet_id,
+ std::lock_guard<std::mutex>& cache_lock) {
+ DORIS_CHECK(!mgr->_files.contains(hash));
+ auto iterator = _meta_store->range_get(tablet_id, hash);
+ // Meta-store initialization failure leaves no database to query.
+ if (!iterator) {
+ LOG(WARNING) << "Failed to create iterator for meta store";
return;
}
- CacheContext context_original;
- context_original.query_id = TUniqueId();
- context_original.expiration_time = block_meta->ttl;
- context_original.cache_type = block_meta->type;
- context_original.tablet_id = key.meta.tablet_id;
+ for (; iterator->valid(); iterator->next()) {
+ const auto meta_key = iterator->key();
+ const auto block_meta = iterator->value();
+ // Malformed persisted records cannot describe a downloaded cache
block.
+ if (!iterator->get_last_key_error().ok() ||
!iterator->get_last_value_error().ok()) {
+ LOG(WARNING) << "Failed to deserialize cache block metadata: "
+ << "key_error=" <<
iterator->get_last_key_error().to_string()
+ << ", value_error=" <<
iterator->get_last_value_error().to_string();
+ continue;
+ }
- if (handle_already_loaded_block(mgr, key.hash, key.offset,
block_meta->size, key.meta.tablet_id,
- cache_lock)) {
- return;
- } else {
- mgr->add_cell(key.hash, context_original, key.offset, block_meta->size,
+ CacheContext context_original;
+ context_original.query_id = TUniqueId();
+ context_original.expiration_time = block_meta.ttl;
+ context_original.cache_type = block_meta.type;
+ context_original.tablet_id = tablet_id;
+ mgr->add_cell(hash, context_original, meta_key.offset, block_meta.size,
FileBlock::State::DOWNLOADED, cache_lock);
}
}
diff --git a/be/src/io/cache/fs_file_cache_storage.h
b/be/src/io/cache/fs_file_cache_storage.h
index 98a600c04b1..aff42abc53b 100644
--- a/be/src/io/cache/fs_file_cache_storage.h
+++ b/be/src/io/cache/fs_file_cache_storage.h
@@ -87,8 +87,9 @@ public:
Status remove(const FileCacheKey& key) override;
Status change_key_meta_type(const FileCacheKey& key, const FileCacheType
type,
const size_t size) override;
- void load_blocks_directly_unlocked(BlockFileCache* _mgr, const
FileCacheKey& key,
- std::lock_guard<std::mutex>&
cache_lock) override;
+ void load_file_blocks_directly_unlocked(BlockFileCache* mgr, const
UInt128Wrapper& hash,
+ int64_t tablet_id,
+ std::lock_guard<std::mutex>&
cache_lock) override;
Status clear(std::string& msg) override;
std::string get_local_file(const FileCacheKey& key) override;
diff --git a/be/src/io/cache/mem_file_cache_storage.cpp
b/be/src/io/cache/mem_file_cache_storage.cpp
index a6d05438a86..6afdd6c0db7 100644
--- a/be/src/io/cache/mem_file_cache_storage.cpp
+++ b/be/src/io/cache/mem_file_cache_storage.cpp
@@ -106,9 +106,9 @@ Status MemFileCacheStorage::change_key_meta_type(const
FileCacheKey& key, const
return Status::OK();
}
-void MemFileCacheStorage::load_blocks_directly_unlocked(BlockFileCache* _mgr,
- const FileCacheKey&
key,
-
std::lock_guard<std::mutex>& cache_lock) {
+void MemFileCacheStorage::load_file_blocks_directly_unlocked(
+ BlockFileCache* mgr, const UInt128Wrapper& hash, int64_t tablet_id,
+ std::lock_guard<std::mutex>& cache_lock) {
// load nothing for in memory cache coz nothing is persisted
}
diff --git a/be/src/io/cache/mem_file_cache_storage.h
b/be/src/io/cache/mem_file_cache_storage.h
index 06fffb6e0b0..4142cc479d7 100644
--- a/be/src/io/cache/mem_file_cache_storage.h
+++ b/be/src/io/cache/mem_file_cache_storage.h
@@ -43,8 +43,9 @@ public:
Status remove(const FileCacheKey& key) override;
Status change_key_meta_type(const FileCacheKey& key, const FileCacheType
type,
const size_t size) override;
- void load_blocks_directly_unlocked(BlockFileCache* _mgr, const
FileCacheKey& key,
- std::lock_guard<std::mutex>&
cache_lock) override;
+ void load_file_blocks_directly_unlocked(BlockFileCache* mgr, const
UInt128Wrapper& hash,
+ int64_t tablet_id,
+ std::lock_guard<std::mutex>&
cache_lock) override;
Status clear(std::string& msg) override;
std::string get_local_file(const FileCacheKey& key) override;
diff --git a/be/test/io/cache/block_file_cache_test_meta_store.cpp
b/be/test/io/cache/block_file_cache_test_meta_store.cpp
index fe897b542bb..b504f7d7329 100644
--- a/be/test/io/cache/block_file_cache_test_meta_store.cpp
+++ b/be/test/io/cache/block_file_cache_test_meta_store.cpp
@@ -23,6 +23,8 @@
#pragma clang diagnostic ignored "-Wkeyword-macro"
#endif
+#include <tuple>
+
#include "io/cache/block_file_cache_test_common.h"
#include "util/defer_op.h"
@@ -738,7 +740,183 @@ TEST_F(BlockFileCacheTest,
handle_already_loaded_block_updates_size_and_tablet)
}
}
-//TODO(zhengyu): check lazy load
+namespace {
+
+class CountingFSFileCacheStorage : public FSFileCacheStorage {
+public:
+ void load_file_blocks_directly_unlocked(BlockFileCache* mgr, const
UInt128Wrapper& hash,
+ int64_t tablet_id,
+ std::lock_guard<std::mutex>&
cache_lock) override {
+ ++direct_loads;
+ FSFileCacheStorage::load_file_blocks_directly_unlocked(mgr, hash,
tablet_id, cache_lock);
+ }
+
+ size_t direct_loads = 0;
+};
+
+TEST_F(BlockFileCacheTest, CompletedAsyncOpenSkipsDirectLoad) {
+ const auto path = caches_dir / "completed_async_open_skips_direct_load";
+ fs::remove_all(path);
+ fs::create_directories(path);
+ Defer remove_cache {[&] { fs::remove_all(path); }};
+ auto settings = get_file_cache_settings(8_mb, 0);
+ settings.max_file_block_size = 4096;
+ BlockFileCache cache(path.string(), settings);
+ auto storage = std::make_unique<CountingFSFileCacheStorage>();
+ auto* storage_ptr = storage.get();
+ cache._storage = std::move(storage);
+ ASSERT_TRUE(cache.initialize().ok());
+ storage_ptr->_cache_background_load_thread.join();
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ ReadStatistics stats;
+ CacheContext context;
+ context.stats = &stats;
+ const auto hash =
BlockFileCache::hash("completed_async_open_skips_direct_load");
+ FileBlocks blocks;
+ bool fully_covered = true;
+ ASSERT_TRUE(cache.get_downloaded_blocks_if_fully_covered(hash, 4096, 8192,
context, &blocks,
+ &fully_covered)
+ .ok());
+ EXPECT_FALSE(fully_covered);
+ auto result = cache.probe(hash, 4096, 8192, context);
+ EXPECT_EQ(result.file_blocks, (std::vector<FileBlockSPtr> {nullptr,
nullptr}));
+ auto holder = cache.get_or_set(hash, 4096, 8192, context);
+ EXPECT_EQ(storage_ptr->direct_loads, 0);
+}
+
+enum class LazyRestoreLookup { GET_OR_SET, DOWNLOADED, PROBE };
+
+class FileCacheLazyRestoreTest
+ : public BlockFileCacheTest,
+ public testing::WithParamInterface<std::tuple<LazyRestoreLookup,
bool>> {
+protected:
+ void SetUp() override {
+ fs::remove_all(path);
+ fs::create_directories(path);
+ {
+ // Destruction drains pending writes before the restarted cache
opens the database.
+ CacheBlockMetaStore meta_store((path / "meta").string());
+ meta_store.put(BlockMetaKey(tablet_id, hash, block_size),
+ BlockMeta(FileCacheType::INDEX, block_size));
+ meta_store.put(BlockMetaKey(tablet_id, hash, 2 * block_size),
+ BlockMeta(FileCacheType::NORMAL, block_size));
+ }
+ sync_point->set_call_back(
+ "BlockFileCache::TmpFile1",
+ [&](auto&&) {
+ std::unique_lock lock(mutex);
+ background_paused = true;
+ cv.notify_all();
+ cv.wait(lock, [&] { return resume_background; });
+ },
+ &guard);
+ sync_point->enable_processing();
+ auto settings = get_file_cache_settings(8_mb, 0);
+ settings.max_file_block_size = block_size;
+ cache = std::make_unique<BlockFileCache>(path.string(), settings);
+ ASSERT_TRUE(cache->initialize().ok());
+ {
+ std::unique_lock lock(mutex);
+ ASSERT_TRUE(
+ cv.wait_for(lock, std::chrono::seconds(10), [&] { return
background_paused; }));
+ }
+ ASSERT_FALSE(cache->get_async_open_success());
+ ASSERT_FALSE(cache->_files.contains(hash));
+ }
+
+ void TearDown() override {
+ {
+ std::lock_guard lock(mutex);
+ resume_background = true;
+ }
+ cv.notify_all();
+ cache.reset();
+ sync_point->disable_processing();
+ fs::remove_all(path);
+ }
+
+ static constexpr int64_t tablet_id = 47;
+ static constexpr size_t block_size = 4096;
+ const fs::path path = caches_dir / "lazy_restore_file_blocks";
+ const UInt128Wrapper hash =
BlockFileCache::hash("lazy_restore_file_blocks");
+ std::mutex mutex;
+ std::condition_variable cv;
+ bool background_paused = false;
+ bool resume_background = false;
+ SyncPoint* sync_point = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ std::unique_ptr<BlockFileCache> cache;
+};
+
+TEST_P(FileCacheLazyRestoreTest, RestoreAllFileBlocksBeforeBackgroundLoad) {
+ const auto [lookup, unaligned] = GetParam();
+ ReadStatistics stats;
+ CacheContext context;
+ context.stats = &stats;
+ context.tablet_id = tablet_id;
+ const size_t offset = block_size + (unaligned ? 100 : 0);
+ const size_t size = 3 * block_size - offset;
+ FileBlocks blocks;
+ switch (lookup) {
+ case LazyRestoreLookup::GET_OR_SET: {
+ auto holder = cache->get_or_set(hash, offset, size, context);
+ blocks = holder.file_blocks;
+ break;
+ }
+ case LazyRestoreLookup::DOWNLOADED: {
+ bool fully_covered = false;
+ ASSERT_TRUE(cache->get_downloaded_blocks_if_fully_covered(hash,
offset, size, context,
+ &blocks,
&fully_covered)
+ .ok());
+ EXPECT_TRUE(fully_covered);
+ break;
+ }
+ case LazyRestoreLookup::PROBE: {
+ auto result = cache->probe(hash, offset, size, context);
+ blocks.assign(result.file_blocks.begin(), result.file_blocks.end());
+ break;
+ }
+ }
+
+ ASSERT_EQ(blocks.size(), 2);
+ size_t expected_offset = block_size;
+ for (const auto& block : blocks) {
+ ASSERT_NE(block, nullptr);
+ EXPECT_EQ(block->state(), FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(block->offset(), expected_offset);
+ EXPECT_EQ(block->range().size(), block_size);
+ EXPECT_EQ(block->tablet_id(), tablet_id);
+ EXPECT_EQ(block->cache_type(),
+ expected_offset == block_size ? FileCacheType::INDEX :
FileCacheType::NORMAL);
+ expected_offset += block_size;
+ }
+ EXPECT_EQ(cache->_cur_cache_size, 2 * block_size);
+
+ {
+ std::lock_guard lock(mutex);
+ resume_background = true;
+ }
+ cv.notify_all();
+ auto* storage = static_cast<FSFileCacheStorage*>(cache->_storage.get());
+ storage->_cache_background_load_thread.join();
+ ASSERT_TRUE(cache->get_async_open_success());
+ std::lock_guard cache_lock(cache->_mutex);
+ EXPECT_EQ(cache->_cur_cache_size, 2 * block_size);
+ for (const auto& block : blocks) {
+ EXPECT_EQ(cache->_files.at(hash).at(block->offset()).file_block,
block);
+ }
+}
+
+INSTANTIATE_TEST_SUITE_P(Startup, FileCacheLazyRestoreTest,
+
testing::Values(std::make_tuple(LazyRestoreLookup::GET_OR_SET, false),
+
std::make_tuple(LazyRestoreLookup::GET_OR_SET, true),
+
std::make_tuple(LazyRestoreLookup::DOWNLOADED, false),
+
std::make_tuple(LazyRestoreLookup::DOWNLOADED, true),
+
std::make_tuple(LazyRestoreLookup::PROBE, false)));
+
+} // namespace
+
//TODO(zhengyu): check version2 start
//TODO(zhengyu): check version2 version3 mixed start
diff --git a/be/test/io/cache/cache_block_meta_store_test.cpp
b/be/test/io/cache/cache_block_meta_store_test.cpp
index 2e8eacfaa7d..68ac74c8a04 100644
--- a/be/test/io/cache/cache_block_meta_store_test.cpp
+++ b/be/test/io/cache/cache_block_meta_store_test.cpp
@@ -150,6 +150,41 @@ TEST_F(CacheBlockMetaStoreTest, RangeQuery) {
EXPECT_FALSE(iterator->valid());
}
+TEST_F(CacheBlockMetaStoreTest, RangeQueryByFile) {
+ const UInt128Wrapper hash((static_cast<uint128_t>(123) << 64) | 456);
+ const UInt128Wrapper next_hash((static_cast<uint128_t>(123) << 64) | 457);
+ const UInt128Wrapper next_high_hash((static_cast<uint128_t>(124) << 64) |
456);
+ const std::vector<BlockMetaKey> keys = {{47, hash, 4096},
+ {47, hash, 8192},
+ {47, next_hash, 4096},
+ {47, next_high_hash, 4096},
+ {48, hash, 4096}};
+ const BlockMeta meta(FileCacheType::NORMAL, 4096, 0);
+ for (const auto& key : keys) {
+ meta_store_->put(key, meta);
+ }
+ // Closing drains the asynchronous queue without a timing-dependent wait.
+ meta_store_.reset();
+ meta_store_ =
std::make_unique<CacheBlockMetaStore>(test_db_path_.string());
+
+ auto iterator = meta_store_->range_get(47, hash);
+ ASSERT_NE(iterator, nullptr);
+ for (size_t offset : {4096, 8192}) {
+ ASSERT_TRUE(iterator->valid());
+ EXPECT_EQ(iterator->key(), BlockMetaKey(47, hash, offset));
+ EXPECT_EQ(iterator->value(), meta);
+ iterator->next();
+ }
+ EXPECT_FALSE(iterator->valid());
+
+ auto missing = meta_store_->range_get(49, hash);
+ ASSERT_NE(missing, nullptr);
+ EXPECT_FALSE(missing->valid());
+ missing = meta_store_->range_get(47,
UInt128Wrapper(static_cast<uint128_t>(0)));
+ ASSERT_NE(missing, nullptr);
+ EXPECT_FALSE(missing->valid());
+}
+
TEST_F(CacheBlockMetaStoreTest, DeleteOperation) {
uint128_t hash1 = (static_cast<uint128_t>(123) << 64) | 456;
BlockMetaKey key1(1, UInt128Wrapper(hash1), 0);
@@ -857,4 +892,4 @@ TEST_F(CacheBlockMetaStoreTest, IteratorErrorHandling) {
EXPECT_EQ(count, 1);
}
-} // namespace doris::io
\ No newline at end of file
+} // namespace doris::io
diff --git a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
index 879f1607963..fd562db7031 100644
--- a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
+++ b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp
@@ -142,10 +142,12 @@ fs::path create_peer_test_file(const std::string& name,
std::string_view content
}
void populate_cache_block_with_content(BlockFileCache* cache, const fs::path&
file_path,
- size_t offset, size_t size,
std::string_view content) {
+ size_t offset, size_t size,
std::string_view content,
+ int64_t tablet_id = 0) {
ReadStatistics stats;
CacheContext context;
context.stats = &stats;
+ context.tablet_id = tablet_id;
auto hash = BlockFileCache::hash(file_path.filename().native());
auto holder = cache->get_or_set(hash, offset, size, context);
auto blocks = fromHolder(holder);
@@ -551,6 +553,9 @@ private:
bool _owned_internal_client_cache = false;
};
+class PeerFileCacheLazyRestoreTest : public CachedRemoteFileReaderPeerTest,
+ public testing::WithParamInterface<bool>
{};
+
} // namespace
TEST_F(CachedRemoteFileReaderPeerTest, read_at_uses_peer_cache_when_available)
{
@@ -1005,6 +1010,83 @@ TEST_F(CachedRemoteFileReaderPeerTest,
peer_file_cache_handler_allows_cached_blo
EXPECT_EQ(response.datas(0).data(), content.substr(0, kPeerTestBlockSize));
}
+TEST_P(PeerFileCacheLazyRestoreTest, RestoreBlocksDuringStartup) {
+ const bool has_rowset_meta = GetParam();
+ const std::string content = "abcdefghijklmnop";
+ const fs::path file_path = "peer_lazy_restore.dat";
+ const auto hash = BlockFileCache::hash(file_path.filename().native());
+ const fs::path cache_path = caches_dir / "peer_lazy_restore_cache";
+ clear_cached_remote_reader_factory();
+ Defer cleanup_cache {[&]() {
+ clear_cached_remote_reader_factory();
+ fs::remove_all(cache_path);
+ }};
+ auto* cache = create_peer_test_cache(cache_path, kPeerTestBlockSize);
+ const int64_t tablet_id = has_rowset_meta ? kPeerTestTabletId : 0;
+ populate_cache_block_with_content(cache, file_path, kPeerTestBlockSize,
kPeerTestBlockSize,
+ content, tablet_id);
+ populate_cache_block_with_content(cache, file_path, 2 *
kPeerTestBlockSize, kPeerTestBlockSize,
+ content, tablet_id);
+ // Closing the cache drains metadata writes before the restarted instance
opens it.
+ clear_cached_remote_reader_factory();
+
+ std::mutex mutex;
+ std::condition_variable cv;
+ bool background_paused = false;
+ bool resume_background = false;
+ auto* sync_point = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sync_point->set_call_back(
+ "BlockFileCache::TmpFile1",
+ [&](auto&&) {
+ std::unique_lock lock(mutex);
+ background_paused = true;
+ cv.notify_all();
+ cv.wait(lock, [&] { return resume_background; });
+ },
+ &guard);
+ Defer resume_and_close {[&]() {
+ {
+ std::lock_guard lock(mutex);
+ resume_background = true;
+ }
+ cv.notify_all();
+ clear_cached_remote_reader_factory();
+ sync_point->disable_processing();
+ }};
+ sync_point->enable_processing();
+ ASSERT_TRUE(FileCacheFactory::instance()
+ ->create_file_cache(cache_path.string(),
+
create_peer_test_settings(kPeerTestBlockSize))
+ .ok());
+ cache = FileCacheFactory::instance()->get_by_path(hash);
+ {
+ std::unique_lock lock(mutex);
+ ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(10), [&] { return
background_paused; }));
+ }
+ ASSERT_FALSE(cache->get_async_open_success());
+ ASSERT_FALSE(cache->_files.contains(hash));
+
+ auto request = create_peer_cache_block_request(file_path,
kPeerTestBlockSize,
+ kPeerTestBlockSize,
content.size());
+ if (has_rowset_meta) {
+ request.mutable_rowset_meta()->set_rowset_id(0);
+ request.mutable_rowset_meta()->set_tablet_id(kPeerTestTabletId);
+ }
+ PFetchPeerDataResponse response;
+ brpc::Controller cntl;
+ const auto st = doris::test_handle_peer_file_cache_block_request(&request,
&response, &cntl);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(response.datas_size(), 1);
+ EXPECT_EQ(response.datas(0).data(), content.substr(kPeerTestBlockSize,
kPeerTestBlockSize));
+ // A request for one block must also restore the other persisted offset of
this file.
+ ASSERT_EQ(cache->_files.at(hash).size(), 2);
+ EXPECT_EQ(cache->_files.at(hash).at(2 *
kPeerTestBlockSize).file_block->state(),
+ FileBlock::State::DOWNLOADED);
+}
+
+INSTANTIATE_TEST_SUITE_P(Startup, PeerFileCacheLazyRestoreTest,
testing::Bool());
+
TEST_F(CachedRemoteFileReaderPeerTest,
peer_fetch_queue_timeout_rejects_expired_request) {
const int32_t old_peer_fetch_queue_timeout_ms =
config::peer_fetch_queue_timeout_ms;
Defer restore_config {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
