[Enhancement](ms) Add sharded LRU cache for tablet index metadata to reduce FDB IO

cloud/src/common/config.h

@@ -158,6 +158,12 @@ CONF_mBool(enable_mvcc_meta_check, "false");
 
 CONF_mInt64(mow_job_key_check_expiration_diff_seconds, "600"); // 10min
 
+// KV cache config
+CONF_mBool(enable_tablet_index_cache, "true");
+CONF_mInt64(ms_tablet_index_cache_capacity, "10000");
+CONF_mInt64(recycler_tablet_index_cache_capacity, "10000");
+CONF_mInt64(tablet_index_cache_ttl_seconds, "0"); // 0 means no TTL
+
 CONF_String(test_s3_ak, "");
 CONF_String(test_s3_sk, "");
 CONF_String(test_s3_endpoint, "");

cloud/src/common/kv_cache.h

@@ -0,0 +1,171 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <array>
+#include <chrono>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <tuple>
+#include <unordered_map>
+
+namespace doris::cloud {
+
+// Sharded LRU Cache to reduce lock contention
+// KeyTuple: std::tuple type, corresponding to BasicKeyInfo::base_type in keys.h
+// ValuePB: protobuf message type
+template <typename KeyTuple, typename ValuePB, size_t NumShards = 16>
+class KvCache {
+public:
+    explicit KvCache(size_t capacity, int64_t ttl_seconds = 0)
+            : shard_capacity_(capacity / NumShards + 1), ttl_seconds_(ttl_seconds) {
+        for (auto& shard : shards_) {
+            shard = std::make_unique<Shard>(shard_capacity_, ttl_seconds);
+        }
+    }
+
+    // Query cache, returns true and fills value if hit
+    bool get(const KeyTuple& key, ValuePB* value) { return get_shard(key)->get(key, value); }
+
+    // Write to cache
+    void put(const KeyTuple& key, const ValuePB& value) { get_shard(key)->put(key, value); }
+
+    // Invalidate single entry
+    void invalidate(const KeyTuple& key) { get_shard(key)->invalidate(key); }
+
+    void clear() {
+        for (auto& shard : shards_) {
+            shard->clear();
+        }
+    }
+
+    size_t size() const {
+        size_t total = 0;
+        for (const auto& shard : shards_) {
+            total += shard->size();
+        }
+        return total;
+    }
+
+private:
+    struct Entry {
+        KeyTuple key;
+        ValuePB value;
+        int64_t expire_time;
+    };
+
+    struct KeyHash {
+        size_t operator()(const KeyTuple& k) const {
+            return std::apply(
+                    [](const auto&... args) {
+                        size_t seed = 0;
+                        ((seed ^= std::hash<std::decay_t<decltype(args)>> {}(args) + 0x9e3779b9 +
+                                  (seed << 6) + (seed >> 2)),
+                         ...);
+                        return seed;
+                    },
+                    k);
+        }
+    };
+
+    class Shard {
+    public:
+        explicit Shard(size_t capacity, int64_t ttl_seconds)
+                : capacity_(capacity), ttl_seconds_(ttl_seconds) {}
+
+        bool get(const KeyTuple& key, ValuePB* value) {
+            std::lock_guard lock(mu_);
+            auto it = map_.find(key);
+            if (it == map_.end()) {
+                return false;
+            }
+            // Check TTL expiration
+            if (ttl_seconds_ > 0 && it->second->expire_time < now_seconds()) {
+                list_.erase(it->second);
+                map_.erase(it);
+                return false;
+            }
+            list_.splice(list_.begin(), list_, it->second);
+            *value = it->second->value;
+            return true;
+        }
+
+        void put(const KeyTuple& key, const ValuePB& value) {
+            std::lock_guard lock(mu_);
+            int64_t expire_time = ttl_seconds_ > 0 ? now_seconds() + ttl_seconds_ : 0;
+            auto it = map_.find(key);
+            if (it != map_.end()) {
+                it->second->value = value;
+                it->second->expire_time = expire_time;
+                list_.splice(list_.begin(), list_, it->second);
+                return;
+            }
+            if (map_.size() >= capacity_) {
+                map_.erase(list_.back().key);
+                list_.pop_back();
+            }
+            list_.push_front({key, value, expire_time});
+            map_[key] = list_.begin();
+        }
+
+        void invalidate(const KeyTuple& key) {
+            std::lock_guard lock(mu_);
+            auto it = map_.find(key);
+            if (it != map_.end()) {
+                list_.erase(it->second);
+                map_.erase(it);
+            }
+        }
+
+        void clear() {
+            std::lock_guard lock(mu_);
+            map_.clear();
+            list_.clear();
+        }
+
+        size_t size() const {
+            std::lock_guard lock(mu_);
+            return map_.size();
+        }
+
+    private:
+        static int64_t now_seconds() {
+            return std::chrono::duration_cast<std::chrono::seconds>(
+                           std::chrono::steady_clock::now().time_since_epoch())
+                    .count();
+        }
+
+        size_t capacity_;
+        int64_t ttl_seconds_;
+        mutable std::mutex mu_;
+        std::list<Entry> list_;
+        std::unordered_map<KeyTuple, typename std::list<Entry>::iterator, KeyHash> map_;
+    };
+
+    Shard* get_shard(const KeyTuple& key) {
+        size_t hash = KeyHash {}(key);
+        return shards_[hash % NumShards].get();
+    }
+
+    size_t shard_capacity_;
+    int64_t ttl_seconds_;
+    std::array<std::unique_ptr<Shard>, NumShards> shards_;
+};
+
+} // namespace doris::cloud

cloud/src/common/kv_cache_manager.h

@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "common/kv_cache.h"
+#include "gen_cpp/cloud.pb.h"
+
+namespace doris::cloud {
+
+struct CacheConfig {
+    size_t tablet_index_capacity = 10000;
+    int64_t tablet_index_ttl_seconds = 0;
+};
+
+class KvCacheManager {
+public:
+    using TabletIndexCache = KvCache<std::tuple<std::string, int64_t>, TabletIndexPB>;
+
+    explicit KvCacheManager(const CacheConfig& config)
+            : tablet_index_cache_(std::make_unique<TabletIndexCache>(
+                      config.tablet_index_capacity, config.tablet_index_ttl_seconds)) {}
+
+    TabletIndexCache* tablet_index_cache() { return tablet_index_cache_.get(); }
+
+private:
+    std::unique_ptr<TabletIndexCache> tablet_index_cache_;
+};
+
+} // namespace doris::cloud

cloud/src/meta-service/meta_service.cpp

@@ -52,6 +52,7 @@
 #include "common/bvars.h"
 #include "common/config.h"
 #include "common/encryption_util.h"
+#include "common/kv_cache_manager.h"
 #include "common/logging.h"
 #include "common/stats.h"
 #include "common/stopwatch.h"
@@ -78,6 +79,8 @@ using namespace std::chrono;
 
 namespace doris::cloud {
 
+static KvCacheManager* g_ms_cache_manager = nullptr;
+
 MetaServiceImpl::MetaServiceImpl(std::shared_ptr<TxnKv> txn_kv,
                                  std::shared_ptr<ResourceManager> resource_mgr,
                                  std::shared_ptr<RateLimiter> rate_limiter,
@@ -90,6 +93,12 @@ MetaServiceImpl::MetaServiceImpl(std::shared_ptr<TxnKv> txn_kv,
           snapshot_manager_(std::move(snapshot_manager)) {
     rate_limiter_->init(this);
     delete_bitmap_lock_white_list_->init();
+    CacheConfig config;
+    config.tablet_index_capacity = config::ms_tablet_index_cache_capacity;
+    config.tablet_index_ttl_seconds = config::tablet_index_cache_ttl_seconds;
+    if (config::enable_tablet_index_cache) {
+        g_ms_cache_manager = new KvCacheManager(config);
+    }
 }
 
 MetaServiceImpl::~MetaServiceImpl() = default;
@@ -191,6 +200,15 @@ bool is_dropped_tablet(Transaction* txn, const std::string& instance_id, int64_t
 
 void get_tablet_idx(MetaServiceCode& code, std::string& msg, Transaction* txn,
                     const std::string& instance_id, int64_t tablet_id, TabletIndexPB& tablet_idx) {
+    // Cache lookup
+    auto cache_key = std::make_tuple(instance_id, tablet_id);
+    if (g_ms_cache_manager &&
+        g_ms_cache_manager->tablet_index_cache()->get(cache_key, &tablet_idx)) {
+        LOG(INFO) << "finish get tablet index from cache, tablet_id= " << tablet_id;
+        return;
+    }
+
+    // FDB read
     std::string key, val;
     meta_tablet_idx_key({instance_id, tablet_id}, &key);
     TxnErrorCode err = txn->get(key, &val);
@@ -216,6 +234,11 @@ void get_tablet_idx(MetaServiceCode& code, std::string& msg, Transaction* txn,
                      << " idx_pb_tablet_id=" << tablet_idx.tablet_id() << " key=" << hex(key);
         return;
     }
+
+    // Cache put
+    if (g_ms_cache_manager) {
+        g_ms_cache_manager->tablet_index_cache()->put(cache_key, tablet_idx);
+    }
 }
 
 void MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,

cloud/src/recycler/recycler.cpp

@@ -45,6 +45,7 @@
 #include <variant>
 
 #include "common/defer.h"
+#include "common/kv_cache_manager.h"
 #include "common/stopwatch.h"
 #include "meta-service/meta_service.h"
 #include "meta-service/meta_service_helper.h"
@@ -408,6 +409,10 @@ int Recycler::start(brpc::Server* server) {
     g_bvar_recycler_task_max_concurrency.set_value(config::recycle_concurrency);
     S3Environment::getInstance();
 
+    if (config::enable_tablet_index_cache) {
+        init_recycler_cache();
+    }
+
     if (config::enable_checker) {
         checker_ = std::make_unique<Checker>(txn_kv_);
         int ret = checker_->start();
@@ -2806,7 +2811,22 @@ int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id,
         }
         for (auto& k : tablet_idx_keys) {
             txn->remove(k);
+            // Invalidate cache for removed tablet_idx_keys
+            if (g_recycler_cache_manager) {
+                // Extract tablet_id from key: meta_tablet_idx_key({instance_id, tablet_id})
+                // The key format is known, we can parse tablet_id from it
+                std::string_view k1 = k;
+                k1.remove_prefix(1);
+                // 0x01 "meta" ${instance_id} "tablet_index" ${tablet_id}
+                std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;
+                decode_key(&k1, &out);
+                DCHECK_EQ(out.size(), 4) << k1;
+                auto tablet_id = std::get<int64_t>(std::get<0>(out[3]));
+                g_recycler_cache_manager->tablet_index_cache()->invalidate(
+                        std::make_tuple(instance_id_, tablet_id));
+            }
         }
+
         for (auto& k : restore_job_keys) {
             txn->remove(k);
         }

cloud/src/recycler/util.cpp

@@ -21,13 +21,24 @@
 
 #include <cstdint>
 
+#include "common/config.h"
+#include "common/kv_cache_manager.h"
 #include "common/util.h"
 #include "meta-service/meta_service_schema.h"
 #include "meta-store/keys.h"
 #include "meta-store/txn_kv.h"
 #include "meta-store/txn_kv_error.h"
 
 namespace doris::cloud {
+
+KvCacheManager* g_recycler_cache_manager = nullptr;
+
+void init_recycler_cache() {
+    CacheConfig config;
+    config.tablet_index_capacity = config::recycler_tablet_index_cache_capacity;
+    config.tablet_index_ttl_seconds = config::tablet_index_cache_ttl_seconds;
+    g_recycler_cache_manager = new KvCacheManager(config);
+}
 namespace config {
 extern int32_t recycle_job_lease_expired_ms;
 } // namespace config
@@ -237,6 +248,14 @@ int lease_instance_recycle_job(TxnKv* txn_kv, std::string_view key, const std::s
 // ret: 0: success, 1: tablet not found, -1: failed
 int get_tablet_idx(TxnKv* txn_kv, const std::string& instance_id, int64_t tablet_id,
                    TabletIndexPB& tablet_idx) {
+    // Cache lookup
+    auto cache_key = std::make_tuple(instance_id, tablet_id);
+    if (g_recycler_cache_manager &&
+        g_recycler_cache_manager->tablet_index_cache()->get(cache_key, &tablet_idx)) {
+        LOG(INFO) << "finish get tablet index from cache, tablet_id= " << tablet_id;
+        return 0;
+    }
+
     std::unique_ptr<Transaction> txn;
     TxnErrorCode err = txn_kv->create_txn(&txn);
     if (err != TxnErrorCode::TXN_OK) {
@@ -266,6 +285,11 @@ int get_tablet_idx(TxnKv* txn_kv, const std::string& instance_id, int64_t tablet
                      << " idx_pb_tablet_id=" << tablet_idx.tablet_id() << " key=" << hex(key);
         return -1;
     }
+
+    // Cache put
+    if (g_recycler_cache_manager) {
+        g_recycler_cache_manager->tablet_index_cache()->put(cache_key, tablet_idx);
+    }
     return 0;
 }
 

cloud/src/recycler/util.h

@@ -97,4 +97,10 @@ int get_tablet_idx(TxnKv* txn_kv, const std::string& instance_id, int64_t tablet
 
 int get_tablet_meta(TxnKv* txn_kv, const std::string& instance_id, int64_t tablet_id,
                     TabletMetaCloudPB& tablet_meta);
+
+void init_recycler_cache();
+
+class KvCacheManager;
+extern KvCacheManager* g_recycler_cache_manager;
+
 } // namespace doris::cloud