From edacc5a3c9c08c4e79f0325665b425f0f23bdd7b Mon Sep 17 00:00:00 2001
From: chejinge <945997690@qq.com>
Date: Tue, 11 Feb 2025 14:08:25 +0800
Subject: [PATCH] feat:new big key count (#3009)

Co-authored-by: chejinge <chejinge@360.cn>
---
 src/pika_admin.cc                     |  5 +++++
 src/storage/include/storage/storage.h |  1 +
 src/storage/src/redis.h               | 18 ++++++++++++++++++
 src/storage/src/redis_hashes.cc       | 22 ++++++++++++++++++++++
 src/storage/src/redis_lists.cc        | 19 +++++++++++++++++++
 src/storage/src/redis_sets.cc         | 13 +++++++++++++
 src/storage/src/redis_zsets.cc        | 25 +++++++++++++++++++++++++
 src/storage/src/storage.cc            | 27 +++++++++++++++++++++++++++
 8 files changed, 130 insertions(+)

diff --git a/src/pika_admin.cc b/src/pika_admin.cc
index 9453e0b870..a2a31fda3d 100644
--- a/src/pika_admin.cc
+++ b/src/pika_admin.cc
@@ -1406,6 +1406,7 @@ void InfoCmd::InfoData(std::string& info) {
   uint64_t total_table_reader_usage = 0;
   uint64_t memtable_usage = 0;
   uint64_t table_reader_usage = 0;
+  uint64_t total_big_key_count = 0;
   std::shared_lock db_rwl(g_pika_server->dbs_rw_);
   for (const auto& db_item : g_pika_server->dbs_) {
     if (!db_item.second) {
@@ -1417,6 +1418,9 @@ void InfoCmd::InfoData(std::string& info) {
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_CUR_SIZE_ALL_MEM_TABLES, &memtable_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_ESTIMATE_TABLE_READER_MEM, &table_reader_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors);
+
+    uint64_t big_key_count = db_item.second->storage()->GetBigKeyStatistics(db_item.first, "bigkey_property");
+    total_big_key_count += big_key_count;
     db_item.second->DBUnlockShared();
     total_memtable_usage += memtable_usage;
     total_table_reader_usage += table_reader_usage;
@@ -1436,6 +1440,7 @@ void InfoCmd::InfoData(std::string& info) {
   tmp_stream << "db_tablereader_usage:" << total_table_reader_usage << "\r\n";
   tmp_stream << "db_fatal:" << (total_background_errors != 0 ? "1" : "0") << "\r\n";
   tmp_stream << "db_fatal_msg:" << (total_background_errors != 0 ? db_fatal_msg_stream.str() : "nullptr") << "\r\n";
+  tmp_stream << "big_key_count:" << total_big_key_count << "\r\n";
 
   info.append(tmp_stream.str());
 }
diff --git a/src/storage/include/storage/storage.h b/src/storage/include/storage/storage.h
index 8c2e53a7b3..0094a03079 100644
--- a/src/storage/include/storage/storage.h
+++ b/src/storage/include/storage/storage.h
@@ -1100,6 +1100,7 @@ class Storage {
   Status EnableAutoCompaction(const OptionType& option_type, 
                     const std::string& db_type, const std::unordered_map<std::string, std::string>& options);
   void GetRocksDBInfo(std::string& info);
+  uint64_t GetBigKeyStatistics(const std::string& db_type, const std::string& property);
 
  private:
   std::unique_ptr<RedisStrings> strings_db_;
diff --git a/src/storage/src/redis.h b/src/storage/src/redis.h
index 21eaa2aa94..261382520b 100644
--- a/src/storage/src/redis.h
+++ b/src/storage/src/redis.h
@@ -9,6 +9,7 @@
 #include <memory>
 #include <string>
 #include <vector>
+#include <iostream>
 
 #include "rocksdb/db.h"
 #include "rocksdb/slice.h"
@@ -115,6 +116,21 @@ class Redis {
   Status SetSmallCompactionDurationThreshold(uint64_t small_compaction_duration_threshold);
   std::vector<rocksdb::ColumnFamilyHandle*> GetHandles(){ return handles_;};
   void GetRocksDBInfo(std::string &info, const char *prefix);
+  void CheckBigKeyAndLog(const std::string& key, uint64_t size) {
+    static const uint64_t kBigKeyThreshold = 10000;
+    if (size > kBigKeyThreshold) {
+      std::lock_guard<std::mutex> lock(big_key_access_mutex_);
+      big_key_access_count_[key]++;
+      std::cerr << "[BIGKEY DETECTED] Key: " << key
+                << ", Size: " << size
+                << ", Access Count: " << big_key_access_count_[key] << std::endl;
+    }
+  }
+
+  std::unordered_map<std::string, int> GetBigKeyStatistics() {
+    std::lock_guard<std::mutex> lock(big_key_access_mutex_);
+    return big_key_access_count_;
+  }
 
  protected:
   Storage* const storage_;
@@ -141,6 +157,8 @@ class Redis {
   Status UpdateSpecificKeyStatistics(const std::string& key, uint64_t count);
   Status UpdateSpecificKeyDuration(const std::string& key, uint64_t duration);
   Status AddCompactKeyTaskIfNeeded(const std::string& key, uint64_t count, uint64_t duration);
+  std::unordered_map<std::string, int> big_key_access_count_;
+  std::mutex big_key_access_mutex_;
 };
 
 }  //  namespace storage
diff --git a/src/storage/src/redis_hashes.cc b/src/storage/src/redis_hashes.cc
index b885a487dd..12c3dd35b6 100644
--- a/src/storage/src/redis_hashes.cc
+++ b/src/storage/src/redis_hashes.cc
@@ -214,6 +214,7 @@ Status RedisHashes::HDel(const Slice& key, const std::vector<std::string>& field
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       *ret = 0;
       return Status::OK();
@@ -266,6 +267,7 @@ Status RedisHashes::HGet(const Slice& key, const Slice& field, std::string* valu
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -290,6 +292,7 @@ Status RedisHashes::HGetall(const Slice& key, std::vector<FieldValue>* fvs) {
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -321,6 +324,7 @@ Status RedisHashes::HGetallWithTTL(const Slice& key, std::vector<FieldValue>* fv
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.count() == 0) {
       return Status::NotFound();
     } else if (parsed_hashes_meta_value.IsStale()) {
@@ -368,6 +372,7 @@ Status RedisHashes::HIncrby(const Slice& key, const Slice& field, int64_t value,
   char meta_value_buf[4] = {0};
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       version = parsed_hashes_meta_value.UpdateVersion();
       parsed_hashes_meta_value.set_count(1);
@@ -443,6 +448,7 @@ Status RedisHashes::HIncrbyfloat(const Slice& key, const Slice& field, const Sli
   char meta_value_buf[4] = {0};
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       version = parsed_hashes_meta_value.UpdateVersion();
       parsed_hashes_meta_value.set_count(1);
@@ -509,6 +515,7 @@ Status RedisHashes::HKeys(const Slice& key, std::vector<std::string>* fields) {
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -535,6 +542,7 @@ Status RedisHashes::HLen(const Slice& key, int32_t* ret) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       *ret = 0;
       return Status::NotFound("Stale");
@@ -563,6 +571,7 @@ Status RedisHashes::HMGet(const Slice& key, const std::vector<std::string>& fiel
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if ((is_stale = parsed_hashes_meta_value.IsStale()) || parsed_hashes_meta_value.count() == 0) {
       for (size_t idx = 0; idx < fields.size(); ++idx) {
         vss->push_back({std::string(), Status::NotFound()});
@@ -613,6 +622,7 @@ Status RedisHashes::HMSet(const Slice& key, const std::vector<FieldValue>& fvs)
   char meta_value_buf[4] = {0};
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       version = parsed_hashes_meta_value.InitialMetaValue();
       if (!parsed_hashes_meta_value.check_set_count(static_cast<int32_t>(filtered_fvs.size()))) {
@@ -673,6 +683,7 @@ Status RedisHashes::HSet(const Slice& key, const Slice& field, const Slice& valu
   char meta_value_buf[4] = {0};
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       version = parsed_hashes_meta_value.InitialMetaValue();
       parsed_hashes_meta_value.set_count(1);
@@ -731,6 +742,7 @@ Status RedisHashes::HSetnx(const Slice& key, const Slice& field, const Slice& va
   char meta_value_buf[4] = {0};
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       version = parsed_hashes_meta_value.InitialMetaValue();
       parsed_hashes_meta_value.set_count(1);
@@ -782,6 +794,7 @@ Status RedisHashes::HVals(const Slice& key, std::vector<std::string>* values) {
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -832,6 +845,7 @@ Status RedisHashes::HScan(const Slice& key, int64_t cursor, const std::string& p
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       *next_cursor = 0;
       return Status::NotFound();
@@ -896,6 +910,7 @@ Status RedisHashes::HScanx(const Slice& key, const std::string& start_field, con
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       *next_field = "";
       return Status::NotFound();
@@ -954,6 +969,7 @@ Status RedisHashes::PKHScanRange(const Slice& key, const Slice& field_start, con
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       return Status::NotFound();
     } else {
@@ -1013,6 +1029,7 @@ Status RedisHashes::PKHRScanRange(const Slice& key, const Slice& field_start, co
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale() || parsed_hashes_meta_value.count() == 0) {
       return Status::NotFound();
     } else {
@@ -1163,6 +1180,7 @@ Status RedisHashes::Expire(const Slice& key, int32_t ttl) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -1186,6 +1204,7 @@ Status RedisHashes::Del(const Slice& key) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -1281,6 +1300,7 @@ Status RedisHashes::Expireat(const Slice& key, int32_t timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -1303,6 +1323,7 @@ Status RedisHashes::Persist(const Slice& key) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_hashes_meta_value.count() == 0) {
@@ -1325,6 +1346,7 @@ Status RedisHashes::TTL(const Slice& key, int64_t* timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedHashesMetaValue parsed_hashes_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_hashes_meta_value.count());
     if (parsed_hashes_meta_value.IsStale()) {
       *timestamp = -2;
       return Status::NotFound("Stale");
diff --git a/src/storage/src/redis_lists.cc b/src/storage/src/redis_lists.cc
index 09a045a4d4..2c0d614861 100644
--- a/src/storage/src/redis_lists.cc
+++ b/src/storage/src/redis_lists.cc
@@ -206,6 +206,7 @@ Status RedisLists::LIndex(const Slice& key, int64_t index, std::string* element)
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     int32_t version = parsed_lists_meta_value.version();
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
@@ -238,6 +239,7 @@ Status RedisLists::LInsert(const Slice& key, const BeforeOrAfter& before_or_afte
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -331,6 +333,7 @@ Status RedisLists::LLen(const Slice& key, uint64_t* len) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -354,6 +357,7 @@ Status RedisLists::LPop(const Slice& key, int64_t count, std::vector<std::string
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -399,6 +403,7 @@ Status RedisLists::LPush(const Slice& key, const std::vector<std::string>& value
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale() || parsed_lists_meta_value.count() == 0) {
       version = parsed_lists_meta_value.InitialMetaValue();
     } else {
@@ -441,6 +446,7 @@ Status RedisLists::LPushx(const Slice& key, const std::vector<std::string>& valu
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -473,6 +479,7 @@ Status RedisLists::LRange(const Slice& key, int64_t start, int64_t stop, std::ve
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -521,6 +528,7 @@ Status RedisLists::LRangeWithTTL(const Slice& key, int64_t start, int64_t stop,
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.count() == 0) {
       return Status::NotFound();
     } else if (parsed_lists_meta_value.IsStale()) {
@@ -583,6 +591,7 @@ Status RedisLists::LRem(const Slice& key, int64_t count, const Slice& value, uin
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -699,6 +708,7 @@ Status RedisLists::LSet(const Slice& key, int64_t index, const Slice& value) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -730,6 +740,7 @@ Status RedisLists::LTrim(const Slice& key, int64_t start, int64_t stop) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     int32_t version = parsed_lists_meta_value.version();
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
@@ -791,6 +802,7 @@ Status RedisLists::RPop(const Slice& key, int64_t count, std::vector<std::string
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -948,6 +960,7 @@ Status RedisLists::RPush(const Slice& key, const std::vector<std::string>& value
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale() || parsed_lists_meta_value.count() == 0) {
       version = parsed_lists_meta_value.InitialMetaValue();
     } else {
@@ -990,6 +1003,7 @@ Status RedisLists::RPushx(const Slice& key, const std::vector<std::string>& valu
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -1123,6 +1137,7 @@ Status RedisLists::Expire(const Slice& key, int32_t ttl) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -1146,6 +1161,7 @@ Status RedisLists::Del(const Slice& key) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -1240,6 +1256,7 @@ Status RedisLists::Expireat(const Slice& key, int32_t timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -1262,6 +1279,7 @@ Status RedisLists::Persist(const Slice& key) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_lists_meta_value.count() == 0) {
@@ -1284,6 +1302,7 @@ Status RedisLists::TTL(const Slice& key, int64_t* timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedListsMetaValue parsed_lists_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_lists_meta_value.count());
     if (parsed_lists_meta_value.IsStale()) {
       *timestamp = -2;
       return Status::NotFound("Stale");
diff --git a/src/storage/src/redis_sets.cc b/src/storage/src/redis_sets.cc
index 5707e032d4..1c20e61d9d 100644
--- a/src/storage/src/redis_sets.cc
+++ b/src/storage/src/redis_sets.cc
@@ -213,6 +213,7 @@ rocksdb::Status RedisSets::SAdd(const Slice& key, const std::vector<std::string>
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale() || parsed_sets_meta_value.count() == 0) {
       version = parsed_sets_meta_value.InitialMetaValue();
       if (!parsed_sets_meta_value.check_set_count(static_cast<int32_t>(filtered_members.size()))) {
@@ -274,6 +275,7 @@ rocksdb::Status RedisSets::SCard(const Slice& key, int32_t* ret) {
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else {
@@ -651,6 +653,7 @@ rocksdb::Status RedisSets::SIsmember(const Slice& key, const Slice& member, int3
   rocksdb::Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -679,6 +682,7 @@ rocksdb::Status RedisSets::SMembers(const Slice& key, std::vector<std::string>*
   rocksdb::Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -712,6 +716,7 @@ Status RedisSets::SMembersWithTTL(const Slice& key,
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.count() == 0) {
       return Status::NotFound();
     } else if (parsed_sets_meta_value.IsStale()) {
@@ -846,6 +851,7 @@ rocksdb::Status RedisSets::SPop(const Slice& key, std::vector<std::string>* memb
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -941,6 +947,7 @@ rocksdb::Status RedisSets::SRandmember(const Slice& key, int32_t count, std::vec
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -1003,6 +1010,7 @@ rocksdb::Status RedisSets::SRem(const Slice& key, const std::vector<std::string>
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -1186,6 +1194,7 @@ rocksdb::Status RedisSets::SScan(const Slice& key, int64_t cursor, const std::st
   rocksdb::Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale() || parsed_sets_meta_value.count() == 0) {
       *next_cursor = 0;
       return rocksdb::Status::NotFound();
@@ -1348,6 +1357,7 @@ rocksdb::Status RedisSets::Expire(const Slice& key, int32_t ttl) {
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -1371,6 +1381,7 @@ rocksdb::Status RedisSets::Del(const Slice& key) {
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -1465,6 +1476,7 @@ rocksdb::Status RedisSets::Expireat(const Slice& key, int32_t timestamp) {
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
@@ -1487,6 +1499,7 @@ rocksdb::Status RedisSets::Persist(const Slice& key) {
   rocksdb::Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedSetsMetaValue parsed_sets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_sets_meta_value.count());
     if (parsed_sets_meta_value.IsStale()) {
       return rocksdb::Status::NotFound("Stale");
     } else if (parsed_sets_meta_value.count() == 0) {
diff --git a/src/storage/src/redis_zsets.cc b/src/storage/src/redis_zsets.cc
index 4f1436b542..b11d67ff57 100644
--- a/src/storage/src/redis_zsets.cc
+++ b/src/storage/src/redis_zsets.cc
@@ -220,6 +220,7 @@ Status RedisZSets::ZPopMax(const Slice& key, const int64_t count, std::vector<Sc
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -266,6 +267,7 @@ Status RedisZSets::ZPopMin(const Slice& key, const int64_t count, std::vector<Sc
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -324,6 +326,7 @@ Status RedisZSets::ZAdd(const Slice& key, const std::vector<ScoreMember>& score_
   if (s.ok()) {
     bool vaild = true;
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale() || parsed_zsets_meta_value.count() == 0) {
       vaild = false;
       version = parsed_zsets_meta_value.InitialMetaValue();
@@ -405,6 +408,7 @@ Status RedisZSets::ZCard(const Slice& key, int32_t* card) {
   Status s = db_->Get(default_read_options_, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       *card = 0;
       return Status::NotFound("Stale");
@@ -430,6 +434,7 @@ Status RedisZSets::ZCount(const Slice& key, double min, double max, bool left_cl
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -486,6 +491,7 @@ Status RedisZSets::ZIncrby(const Slice& key, const Slice& member, double increme
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale() || parsed_zsets_meta_value.count() == 0) {
       version = parsed_zsets_meta_value.InitialMetaValue();
     } else {
@@ -549,6 +555,7 @@ Status RedisZSets::ZRange(const Slice& key, int32_t start, int32_t stop, std::ve
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -595,6 +602,7 @@ Status RedisZSets::ZRangeWithTTL(const Slice& key, int32_t start, int32_t stop,
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.count() == 0) {
       return Status::NotFound();
     } else if (parsed_zsets_meta_value.IsStale()) {
@@ -654,6 +662,7 @@ Status RedisZSets::ZRangebyscore(const Slice& key, double min, double max, bool
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -719,6 +728,7 @@ Status RedisZSets::ZRank(const Slice& key, const Slice& member, int32_t* rank) {
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -769,6 +779,7 @@ Status RedisZSets::ZRem(const Slice& key, const std::vector<std::string>& member
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -818,6 +829,7 @@ Status RedisZSets::ZRemrangebyrank(const Slice& key, int32_t start, int32_t stop
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -874,6 +886,7 @@ Status RedisZSets::ZRemrangebyscore(const Slice& key, double min, double max, bo
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -944,6 +957,7 @@ Status RedisZSets::ZRevrange(const Slice& key, int32_t start, int32_t stop, std:
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -990,6 +1004,7 @@ Status RedisZSets::ZRevrangebyscore(const Slice& key, double min, double max, bo
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1055,6 +1070,7 @@ Status RedisZSets::ZRevrank(const Slice& key, const Slice& member, int32_t* rank
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1097,6 +1113,7 @@ Status RedisZSets::ZScore(const Slice& key, const Slice& member, double* score)
   Status s = db_->Get(read_options, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     int32_t version = parsed_zsets_meta_value.version();
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
@@ -1365,6 +1382,7 @@ Status RedisZSets::ZRangebylex(const Slice& key, const Slice& min, const Slice&
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale() || parsed_zsets_meta_value.count() == 0) {
       return Status::NotFound();
     } else {
@@ -1426,6 +1444,7 @@ Status RedisZSets::ZRemrangebylex(const Slice& key, const Slice& min, const Slic
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale() || parsed_zsets_meta_value.count() == 0) {
       return Status::NotFound();
     } else {
@@ -1485,6 +1504,7 @@ Status RedisZSets::Expire(const Slice& key, int32_t ttl) {
   Status s = db_->Get(default_read_options_, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1507,6 +1527,7 @@ Status RedisZSets::Del(const Slice& key) {
   Status s = db_->Get(default_read_options_, key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1601,6 +1622,7 @@ Status RedisZSets::Expireat(const Slice& key, int32_t timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1637,6 +1659,7 @@ Status RedisZSets::ZScan(const Slice& key, int64_t cursor, const std::string& pa
   Status s = db_->Get(read_options, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale() || parsed_zsets_meta_value.count() == 0) {
       *next_cursor = 0;
       return Status::NotFound();
@@ -1802,6 +1825,7 @@ Status RedisZSets::Persist(const Slice& key) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       return Status::NotFound("Stale");
     } else if (parsed_zsets_meta_value.count() == 0) {
@@ -1824,6 +1848,7 @@ Status RedisZSets::TTL(const Slice& key, int64_t* timestamp) {
   Status s = db_->Get(default_read_options_, handles_[0], key, &meta_value);
   if (s.ok()) {
     ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
+    CheckBigKeyAndLog(key.ToString(), parsed_zsets_meta_value.count());
     if (parsed_zsets_meta_value.IsStale()) {
       *timestamp = -2;
       return Status::NotFound("Stale");
diff --git a/src/storage/src/storage.cc b/src/storage/src/storage.cc
index 3126717859..8ae31aba3c 100644
--- a/src/storage/src/storage.cc
+++ b/src/storage/src/storage.cc
@@ -1855,6 +1855,33 @@ uint64_t Storage::GetProperty(const std::string& db_type, const std::string& pro
   return result;
 }
 
+uint64_t Storage::GetBigKeyStatistics(const std::string& db_type, const std::string& property) {
+  uint64_t out = 0;
+  uint64_t result = 0;
+  if (db_type == ALL_DB || db_type == HASHES_DB) {
+    hashes_db_->GetBigKeyStatistics();
+    result += out;
+  }
+
+  if (db_type == ALL_DB || db_type == LISTS_DB) {
+    lists_db_->GetBigKeyStatistics();
+    result += out;
+  }
+  if (db_type == ALL_DB || db_type == ZSETS_DB) {
+    zsets_db_->GetBigKeyStatistics();
+    result += out;
+  }
+  if (db_type == ALL_DB || db_type == SETS_DB) {
+    sets_db_->GetBigKeyStatistics();
+    result += out;
+  }
+  if (db_type == ALL_DB || db_type == STREAMS_DB) {
+    streams_db_->GetBigKeyStatistics();
+    result += out;
+  }
+  return result;
+}
+
 Status Storage::GetKeyNum(std::vector<KeyInfo>* key_infos) {
   KeyInfo key_info;
   // NOTE: keep the db order with string, hash, list, zset, set