[Refactor] Split data cache engine into disk cache engine and memory cache engine. (backport #62760) (#62799)

Signed-off-by: trueeyu <[email protected]>
Co-authored-by: trueeyu <[email protected]>

Author: mergify[bot]

be/src/agent/task_worker_pool.cpp

@@ -848,7 +848,8 @@ void* ReportDataCacheMetricsTaskWorkerPool::_worker_thread_callback(void* arg_th
         request.__set_report_version(g_report_version.load(std::memory_order_relaxed));
 
         TDataCacheMetrics t_metrics{};
-        const LocalCacheEngine* cache = DataCache::GetInstance()->local_cache();
+        // TODO: mem_metrics + disk_metrics
+        const LocalCacheEngine* cache = DataCache::GetInstance()->local_disk_cache();
         if (cache != nullptr && cache->is_initialized()) {
             const auto metrics = cache->cache_metrics();
             DataCacheUtils::set_metrics_from_thrift(t_metrics, metrics);

be/src/bench/object_cache_bench.cpp

@@ -129,8 +129,6 @@ void ObjectCacheBench::init_cache(CacheType cache_type) {
         _page_cache = std::make_shared<StoragePageCache>();
         _page_cache->init(_lru_cache.get());
     } else {
-        opt.engine = "starcache";
-
         _star_cache = std::make_shared<StarCacheEngine>();
         Status st = _star_cache->init(opt);
         if (!st.ok()) {

be/src/cache/block_cache/block_cache.cpp

@@ -36,7 +36,7 @@ BlockCache::~BlockCache() {
     (void)shutdown();
 }
 
-Status BlockCache::init(const CacheOptions& options, std::shared_ptr<LocalCacheEngine> local_cache,
+Status BlockCache::init(const BlockCacheOptions& options, std::shared_ptr<LocalCacheEngine> local_cache,
                         std::shared_ptr<RemoteCacheEngine> remote_cache) {
     _block_size = std::min(options.block_size, MAX_BLOCK_SIZE);
     _local_cache = std::move(local_cache);

be/src/cache/block_cache/block_cache.h

@@ -33,7 +33,7 @@ class BlockCache {
     ~BlockCache();
 
     // Init the block cache instance
-    Status init(const CacheOptions& options, std::shared_ptr<LocalCacheEngine> local_cache,
+    Status init(const BlockCacheOptions& options, std::shared_ptr<LocalCacheEngine> local_cache,
                 std::shared_ptr<RemoteCacheEngine> remote_cache);
 
     // Write data buffer to cache, the `offset` must be aligned by block size

be/src/cache/cache_options.h

@@ -42,7 +42,15 @@ struct DirSpace {
     size_t size;
 };
 
-struct CacheOptions {
+struct RemoteCacheOptions {
+    double skip_read_factor = 0;
+};
+
+struct MemCacheOptions {
+    size_t mem_space_size = 0;
+};
+
+struct DiskCacheOptions {
     // basic
     size_t mem_space_size = 0;
     std::vector<DirSpace> dir_spaces;
@@ -54,7 +62,6 @@ struct CacheOptions {
     bool enable_direct_io = false;
     bool enable_tiered_cache = true;
     bool enable_datacache_persistence = false;
-    std::string engine;
     size_t max_concurrent_inserts = 0;
     size_t max_flying_memory_mb = 0;
     double scheduler_threads_per_cpu = 0;
@@ -63,6 +70,10 @@ struct CacheOptions {
     std::string eviction_policy;
 };
 
+struct BlockCacheOptions {
+    size_t block_size = 0;
+};
+
 struct WriteCacheOptions {
     int8_t priority = 0;
     // If ttl_seconds=0 (default), no ttl restriction will be set. If an old one exists, remove it.

be/src/cache/datacache.cpp

@@ -44,37 +44,32 @@ Status DataCache::init(const std::vector<StorePath>& store_paths) {
     _page_cache = std::make_shared<StoragePageCache>();
 
 #if defined(WITH_STARCACHE)
-    if (config::datacache_engine == "" || config::datacache_engine == "starcache") {
-        config::datacache_engine = "starcache";
-    } else {
-        config::datacache_engine = "lrucache";
-    }
-#else
-    config::datacache_engine = "lrucache";
+    _local_disk_cache_engine = "starcache";
 #endif
+    _local_mem_cache_engine = "lrucache";
 
     if (!config::datacache_enable) {
         config::disable_storage_page_cache = true;
         config::block_cache_enable = false;
         return Status::OK();
     }
 
-    ASSIGN_OR_RETURN(auto cache_options, _init_cache_options());
+    ASSIGN_OR_RETURN(auto mem_cache_options, _init_mem_cache_options());
 
-    if (config::datacache_engine == "starcache") {
 #if defined(WITH_STARCACHE)
-        RETURN_IF_ERROR(_init_starcache_engine(&cache_options));
-        RETURN_IF_ERROR(_init_peer_cache(cache_options));
+    ASSIGN_OR_RETURN(auto disk_cache_options, _init_disk_cache_options());
+    RETURN_IF_ERROR(_init_starcache_engine(&disk_cache_options));
 
-        if (config::block_cache_enable) {
-            RETURN_IF_ERROR(_block_cache->init(cache_options, _local_cache, _remote_cache));
-        }
-#else
-        return Status::InternalError("starcache engine is not supported");
-#endif
-    } else {
-        RETURN_IF_ERROR(_init_lrucache_engine(cache_options));
+    auto remote_cache_options = _init_remote_cache_options();
+    RETURN_IF_ERROR(_init_peer_cache(remote_cache_options));
+
+    if (config::block_cache_enable) {
+        auto block_cache_options = _init_block_cache_options();
+        RETURN_IF_ERROR(_block_cache->init(block_cache_options, _local_disk_cache, _remote_cache));
     }
+#endif
+
+    RETURN_IF_ERROR(_init_lrucache_engine(mem_cache_options));
 
     RETURN_IF_ERROR(_init_page_cache());
 
@@ -100,14 +95,15 @@ void DataCache::destroy() {
     LOG(INFO) << "pagecache shutdown successfully";
 
     _block_cache.reset();
-    _local_cache.reset();
+    _local_mem_cache.reset();
+    _local_disk_cache.reset();
     _remote_cache.reset();
     LOG(INFO) << "datacache shutdown successfully";
 }
 
 bool DataCache::adjust_mem_capacity(int64_t delta, size_t min_capacity) {
-    if (_local_cache != nullptr) {
-        Status st = _local_cache->adjust_mem_quota(delta, min_capacity);
+    if (_local_mem_cache != nullptr) {
+        Status st = _local_mem_cache->adjust_mem_quota(delta, min_capacity);
         if (st.ok()) {
             return true;
         } else {
@@ -119,52 +115,67 @@ bool DataCache::adjust_mem_capacity(int64_t delta, size_t min_capacity) {
 }
 
 size_t DataCache::get_mem_capacity() const {
-    if (_local_cache != nullptr) {
-        return _local_cache->mem_quota();
+    if (_local_mem_cache != nullptr) {
+        return _local_mem_cache->mem_quota();
     } else {
         return 0;
     }
 }
 
-Status DataCache::_init_lrucache_engine(const CacheOptions& cache_options) {
-    _local_cache = std::make_shared<LRUCacheEngine>();
-    RETURN_IF_ERROR(_local_cache->init(cache_options));
+Status DataCache::_init_lrucache_engine(const MemCacheOptions& cache_options) {
+    _local_mem_cache = std::make_shared<LRUCacheEngine>();
+    RETURN_IF_ERROR(reinterpret_cast<LRUCacheEngine*>(_local_mem_cache.get())->init(cache_options));
     LOG(INFO) << "lrucache engine init successfully";
     return Status::OK();
 }
 
 Status DataCache::_init_page_cache() {
-    _page_cache->init(_local_cache.get());
+    _page_cache->init(_local_mem_cache.get());
     _page_cache->init_metrics();
     LOG(INFO) << "storage page cache init successfully";
     return Status::OK();
 }
 
 #if defined(WITH_STARCACHE)
-Status DataCache::_init_starcache_engine(CacheOptions* cache_options) {
+Status DataCache::_init_starcache_engine(DiskCacheOptions* cache_options) {
     // init starcache & disk monitor
     // TODO: DiskSpaceMonitor needs to be decoupled from StarCacheEngine.
-    _local_cache = std::make_shared<StarCacheEngine>();
-    _disk_space_monitor = std::make_shared<DiskSpaceMonitor>(_local_cache.get());
+    _local_disk_cache = std::make_shared<StarCacheEngine>();
+    _disk_space_monitor = std::make_shared<DiskSpaceMonitor>(_local_disk_cache.get());
     RETURN_IF_ERROR(_disk_space_monitor->init(&cache_options->dir_spaces));
-    RETURN_IF_ERROR(_local_cache->init(*cache_options));
+    RETURN_IF_ERROR(reinterpret_cast<StarCacheEngine*>(_local_disk_cache.get())->init(*cache_options));
     _disk_space_monitor->start();
     return Status::OK();
 }
 
-Status DataCache::_init_peer_cache(const CacheOptions& cache_options) {
+Status DataCache::_init_peer_cache(const RemoteCacheOptions& cache_options) {
     _remote_cache = std::make_shared<PeerCacheEngine>();
     return _remote_cache->init(cache_options);
 }
 #endif
 
-StatusOr<CacheOptions> DataCache::_init_cache_options() {
-    CacheOptions cache_options;
+RemoteCacheOptions DataCache::_init_remote_cache_options() {
+    RemoteCacheOptions cache_options{.skip_read_factor = config::datacache_skip_read_factor};
+    return cache_options;
+}
+
+StatusOr<MemCacheOptions> DataCache::_init_mem_cache_options() {
+    MemCacheOptions cache_options;
     RETURN_IF_ERROR(DataCacheUtils::parse_conf_datacache_mem_size(
             config::datacache_mem_size, _global_env->process_mem_limit(), &cache_options.mem_space_size));
-    cache_options.engine = config::datacache_engine;
+    return cache_options;
+}
+
+BlockCacheOptions DataCache::_init_block_cache_options() {
+    BlockCacheOptions cache_options;
+    cache_options.block_size = config::datacache_block_size;
+    return cache_options;
+}
+
+StatusOr<DiskCacheOptions> DataCache::_init_disk_cache_options() {
+    DiskCacheOptions cache_options;
 
-    if (config::datacache_engine == "starcache") {
+    if (_local_disk_cache_engine == "starcache") {
 #ifdef USE_STAROS
         std::vector<string> corresponding_starlet_dirs;
         if (config::datacache_unified_instance_enable && !config::starlet_cache_dir.empty()) {
@@ -276,8 +287,8 @@ void DataCache::try_release_resource_before_core_dump() {
         return release_all || modules.contains(name);
     };
 
-    if (_local_cache != nullptr && need_release("data_cache")) {
-        (void)_local_cache->update_mem_quota(0, false);
+    if (_local_mem_cache != nullptr && need_release("data_cache")) {
+        (void)_local_mem_cache->update_mem_quota(0, false);
     }
 }
 

be/src/cache/datacache.h

@@ -23,7 +23,7 @@ namespace starrocks {
 class Status;
 class StorePath;
 class RemoteCacheEngine;
-class CacheOptions;
+class DiskCacheOptions;
 class GlobalEnv;
 class DiskSpaceMonitor;
 class MemSpaceMonitor;
@@ -39,10 +39,16 @@ class DataCache {
 
     void try_release_resource_before_core_dump();
 
-    void set_local_cache(std::shared_ptr<LocalCacheEngine> local_cache) { _local_cache = std::move(local_cache); }
+    void set_local_mem_cache(std::shared_ptr<LocalCacheEngine> local_mem_cache) {
+        _local_mem_cache = std::move(local_mem_cache);
+    }
+    void set_local_disk_cache(std::shared_ptr<LocalCacheEngine> local_disk_cache) {
+        _local_disk_cache = std::move(local_disk_cache);
+    }
     void set_page_cache(std::shared_ptr<StoragePageCache> page_cache) { _page_cache = std::move(page_cache); }
 
-    LocalCacheEngine* local_cache() { return _local_cache.get(); }
+    LocalCacheEngine* local_mem_cache() { return _local_mem_cache.get(); }
+    LocalCacheEngine* local_disk_cache() { return _local_disk_cache.get(); }
     BlockCache* block_cache() const { return _block_cache.get(); }
     void set_block_cache(std::shared_ptr<BlockCache> block_cache) { _block_cache = std::move(block_cache); }
     StoragePageCache* page_cache() const { return _page_cache.get(); }
@@ -56,19 +62,26 @@ class DataCache {
     size_t get_mem_capacity() const;
 
 private:
-    StatusOr<CacheOptions> _init_cache_options();
+    StatusOr<MemCacheOptions> _init_mem_cache_options();
+    StatusOr<DiskCacheOptions> _init_disk_cache_options();
+    RemoteCacheOptions _init_remote_cache_options();
+    BlockCacheOptions _init_block_cache_options();
+
 #if defined(WITH_STARCACHE)
-    Status _init_starcache_engine(CacheOptions* cache_options);
-    Status _init_peer_cache(const CacheOptions& cache_options);
+    Status _init_starcache_engine(DiskCacheOptions* cache_options);
+    Status _init_peer_cache(const RemoteCacheOptions& cache_options);
 #endif
-    Status _init_lrucache_engine(const CacheOptions& cache_options);
+    Status _init_lrucache_engine(const MemCacheOptions& cache_options);
     Status _init_page_cache();
 
     GlobalEnv* _global_env;
     std::vector<StorePath> _store_paths;
 
     // cache engine
-    std::shared_ptr<LocalCacheEngine> _local_cache;
+    std::string _local_mem_cache_engine;
+    std::string _local_disk_cache_engine;
+    std::shared_ptr<LocalCacheEngine> _local_mem_cache;
+    std::shared_ptr<LocalCacheEngine> _local_disk_cache;
     std::shared_ptr<RemoteCacheEngine> _remote_cache;
 
     std::shared_ptr<BlockCache> _block_cache;

be/src/cache/local_cache_engine.h

@@ -27,7 +27,6 @@ class LocalCacheEngine {
 public:
     virtual ~LocalCacheEngine() = default;
 
-    virtual Status init(const CacheOptions& options) = 0;
     virtual bool is_initialized() const = 0;
 
     // Write data to cache

be/src/cache/lrucache_engine.cpp

@@ -17,7 +17,7 @@
 #include <butil/fast_rand.h>
 
 namespace starrocks {
-Status LRUCacheEngine::init(const CacheOptions& options) {
+Status LRUCacheEngine::init(const MemCacheOptions& options) {
     _cache = std::make_unique<ShardedLRUCache>(options.mem_space_size);
     _initialized.store(true, std::memory_order_relaxed);
     return Status::OK();

be/src/cache/lrucache_engine.h

@@ -25,7 +25,7 @@ class LRUCacheEngine final : public LocalCacheEngine {
     LRUCacheEngine() = default;
     virtual ~LRUCacheEngine() override = default;
 
-    Status init(const CacheOptions& options) override;
+    Status init(const MemCacheOptions& options);
     bool is_initialized() const override { return _initialized.load(std::memory_order_relaxed); }
 
     Status write(const std::string& key, const IOBuffer& buffer, WriteCacheOptions* options) override;