Merge branch 'unstable' into fix/xpending-parse-end-id

Author: jihuayu

cmake/jemalloc.cmake

@@ -32,8 +32,8 @@ endif()
 include(cmake/utils.cmake)
 
 FetchContent_DeclareGitHubWithMirror(jemalloc
-  facebook/jemalloc 5.3.0
-  SHA1=1be8fdba021e9d6ed201e7d6a3c464b2223fc927
+  jemalloc/jemalloc 5.3.1
+  MD5=d421650ffe4b9e0ec463a9c26135f657
 )
 
 FetchContent_GetProperties(jemalloc)

kvrocks.conf

@@ -77,6 +77,26 @@ repl-namespace-enabled no
 #
 # proto-max-bulk-len 536870912
 
+# The encoding mode for newly created hash keys.
+# - legacy: use the original subkey format (no per-field expire support).
+#   Existing data written in this mode is always readable regardless of this setting.
+# - field-expiration: use the new subkey format that prepends an 8-byte expire
+#   timestamp to each field value. Required for HEXPIRE/HPERSIST/HPTTL commands.
+# NOTE: this only affects newly created keys. Existing keys retain their original mode.
+# Default: legacy
+# Available options: legacy, field-expiration
+hash-encoding-mode legacy
+
+# How HLEN computes the number of fields when some fields have a TTL.
+# - accurate: return the exact count. O(1) when no fields have TTL or none have
+#   expired yet; otherwise performs a scan to remove expired fields and update metadata.
+# - approximate: always return the stored size without scanning. The count may include
+#   fields that have already expired but have not yet been cleaned up.
+# This option has no effect when hash-encoding-mode is legacy.
+# Default: accurate
+# Available options: accurate, approximate
+hash-length-mode accurate
+
 # Persist the cluster nodes topology in local file($dir/nodes.conf). This configuration
 # takes effect only if the cluster mode was enabled.
 #

src/config/config.cc

@@ -93,6 +93,16 @@ const std::vector<ConfigEnum<BlockCacheType>> cache_types{[] {
 const std::vector<ConfigEnum<MigrationType>> migration_types{{"redis-command", MigrationType::kRedisCommand},
                                                              {"raw-key-value", MigrationType::kRawKeyValue}};
 
+const std::vector<ConfigEnum<HashSubkeyEncodingMode>> hash_subkey_encoding_modes{
+    {"legacy", HashSubkeyEncodingMode::kLegacy},
+    {"field-expiration", HashSubkeyEncodingMode::kFieldExpiration},
+};
+
+const std::vector<ConfigEnum<HashLengthMode>> hash_length_modes{
+    {"accurate", HashLengthMode::kAccurate},
+    {"approximate", HashLengthMode::kApproximate},
+};
+
 std::string TrimRocksDbPrefix(std::string s) {
   constexpr std::string_view prefix = "rocksdb.";
   if (!util::StartsWithICase(s, prefix)) return s;
@@ -237,6 +247,11 @@ Config::Config() {
       {"redis-cursor-compatible", false, new YesNoField(&redis_cursor_compatible, true)},
       {"redis-databases", true, new IntField(&redis_databases, 0, 0, INT_MAX)},
       {"resp3-enabled", false, new YesNoField(&resp3_enabled, true)},
+      {"hash-encoding-mode", false,
+       new EnumField<HashSubkeyEncodingMode>(&hash_encoding_mode, hash_subkey_encoding_modes,
+                                             HashSubkeyEncodingMode::kLegacy)},
+      {"hash-length-mode", false,
+       new EnumField<HashLengthMode>(&hash_length_mode, hash_length_modes, HashLengthMode::kAccurate)},
       {"repl-namespace-enabled", false, new YesNoField(&repl_namespace_enabled, false)},
       {"proto-max-bulk-len", false,
        new IntWithUnitField<uint64_t>(&proto_max_bulk_len, std::to_string(512 * MiB), 1 * MiB,

src/config/config.h

@@ -180,6 +180,8 @@ struct Config {
   int redis_databases = 0;
   bool resp3_enabled = false;
   int log_retention_days;
+  HashSubkeyEncodingMode hash_encoding_mode = HashSubkeyEncodingMode::kLegacy;
+  HashLengthMode hash_length_mode = HashLengthMode::kAccurate;
 
   // load_tokens is used to buffer the tokens when loading,
   // don't use it to authenticate or rewrite the configuration file.

src/search/indexer.cc

@@ -43,7 +43,7 @@ StatusOr<FieldValueRetriever> FieldValueRetriever::Create(IndexOnDataType type,
     std::string ns_key = db.AppendNamespacePrefix(key);
     HashMetadata metadata(false);
 
-    auto s = db.GetMetadata(ctx, ns_key, &metadata);
+    auto s = db.getMetadata(ctx, ns_key, &metadata);
     if (!s.ok()) return {s.IsNotFound() ? Status::NotFound : Status::NotOK, s.ToString()};
     return FieldValueRetriever(db, metadata, key);
   } else if (type == IndexOnDataType::JSON) {
@@ -98,7 +98,7 @@ StatusOr<kqir::Value> FieldValueRetriever::ParseFromJson(const jsoncons::json &v
   }
 }
 
-StatusOr<kqir::Value> FieldValueRetriever::ParseFromHash(const std::string &value,
+StatusOr<kqir::Value> FieldValueRetriever::ParseFromHash(std::string_view value,
                                                          const redis::IndexFieldMetadata *type) {
   if (auto numeric [[maybe_unused]] = dynamic_cast<const redis::NumericFieldMetadata *>(type)) {
     auto num = GET_OR_RET(ParseFloat(value));
@@ -137,7 +137,11 @@ StatusOr<kqir::Value> FieldValueRetriever::Retrieve(engine::Context &ctx, std::s
     if (s.IsNotFound()) return {Status::NotFound, s.ToString()};
     if (!s.ok()) return {Status::NotOK, s.ToString()};
 
-    return ParseFromHash(value, type);
+    Slice decoded_value(value);
+    s = metadata.DecodeSubkeyValue(&decoded_value);
+    if (!s.ok()) return {Status::NotOK, s.ToString()};
+
+    return ParseFromHash(decoded_value.ToStringView(), type);
   } else if (std::holds_alternative<JsonData>(db)) {
     auto &value = std::get<JsonData>(db);
 

src/search/indexer.h

@@ -67,7 +67,7 @@ struct FieldValueRetriever {
   StatusOr<kqir::Value> Retrieve(engine::Context &ctx, std::string_view field, const redis::IndexFieldMetadata *type);
 
   static StatusOr<kqir::Value> ParseFromJson(const jsoncons::json &value, const redis::IndexFieldMetadata *type);
-  static StatusOr<kqir::Value> ParseFromHash(const std::string &value, const redis::IndexFieldMetadata *type);
+  static StatusOr<kqir::Value> ParseFromHash(std::string_view value, const redis::IndexFieldMetadata *type);
 };
 
 struct IndexUpdater {

src/storage/redis_db.cc

@@ -558,42 +558,7 @@ rocksdb::Status Database::KeyExist(engine::Context &ctx, const std::string &key)
 rocksdb::Status SubKeyScanner::Scan(engine::Context &ctx, RedisType type, const Slice &user_key,
                                     const std::string &cursor, uint64_t limit, const std::string &subkey_prefix,
                                     std::vector<std::string> *keys, std::vector<std::string> *values) {
-  uint64_t cnt = 0;
-  std::string ns_key = AppendNamespacePrefix(user_key);
-  Metadata metadata(type, false);
-  rocksdb::Status s = GetMetadata(ctx, {type}, ns_key, &metadata);
-  if (!s.ok()) return s;
-
-  auto iter = util::UniqueIterator(ctx, ctx.DefaultScanOptions());
-  std::string match_prefix_key =
-      InternalKey(ns_key, subkey_prefix, metadata.version, storage_->IsSlotIdEncoded()).Encode();
-
-  std::string start_key;
-  if (!cursor.empty()) {
-    start_key = InternalKey(ns_key, cursor, metadata.version, storage_->IsSlotIdEncoded()).Encode();
-  } else {
-    start_key = match_prefix_key;
-  }
-  for (iter->Seek(start_key); iter->Valid(); iter->Next()) {
-    if (!cursor.empty() && iter->key() == start_key) {
-      // if cursor is not empty, then we need to skip start_key
-      // because we already return that key in the last scan
-      continue;
-    }
-    if (!iter->key().starts_with(match_prefix_key)) {
-      break;
-    }
-    InternalKey ikey(iter->key(), storage_->IsSlotIdEncoded());
-    keys->emplace_back(ikey.GetSubKey().ToString());
-    if (values != nullptr) {
-      values->emplace_back(iter->value().ToString());
-    }
-    cnt++;
-    if (limit > 0 && cnt >= limit) {
-      break;
-    }
-  }
-  return iter->status();
+  return scanSubkeys<Metadata>(ctx, type, user_key, cursor, limit, subkey_prefix, keys, values);
 }
 
 RedisType WriteBatchLogData::GetRedisType() const { return type_; }

src/storage/redis_db.h

@@ -22,11 +22,13 @@
 
 #include <optional>
 #include <string>
+#include <type_traits>
 #include <utility>
 #include <variant>
 #include <vector>
 
 #include "cluster/cluster_defs.h"
+#include "common/db_util.h"
 #include "redis_metadata.h"
 #include "storage.h"
 
@@ -182,6 +184,68 @@ class SubKeyScanner : public redis::Database {
   rocksdb::Status Scan(engine::Context &ctx, RedisType type, const Slice &user_key, const std::string &cursor,
                        uint64_t limit, const std::string &subkey_prefix, std::vector<std::string> *keys,
                        std::vector<std::string> *values = nullptr);
+
+ protected:
+  struct RawSubkeyValueAdapter {
+    template <typename MetadataT>
+    rocksdb::Status operator()(const MetadataT &, Slice *) const {
+      return rocksdb::Status::OK();
+    }
+  };
+
+  template <typename MetadataT>
+  static MetadataT createScanMetadata(RedisType type) {
+    if constexpr (std::is_same_v<MetadataT, Metadata>) {
+      return Metadata(type, false);
+    } else {
+      return MetadataT(false);
+    }
+  }
+
+  template <typename MetadataT, typename ValueAdapter = RawSubkeyValueAdapter>
+  rocksdb::Status scanSubkeys(engine::Context &ctx, RedisType type, const Slice &user_key, const std::string &cursor,
+                              uint64_t limit, const std::string &subkey_prefix, std::vector<std::string> *keys,
+                              std::vector<std::string> *values = nullptr, ValueAdapter value_adapter = {}) {
+    uint64_t cnt = 0;
+    std::string ns_key = AppendNamespacePrefix(user_key);
+    auto metadata = createScanMetadata<MetadataT>(type);
+    rocksdb::Status s = GetMetadata(ctx, {type}, ns_key, &metadata);
+    if (!s.ok()) return s;
+
+    auto iter = util::UniqueIterator(ctx, ctx.DefaultScanOptions());
+    std::string match_prefix_key =
+        InternalKey(ns_key, subkey_prefix, metadata.version, storage_->IsSlotIdEncoded()).Encode();
+
+    std::string start_key;
+    if (!cursor.empty()) {
+      start_key = InternalKey(ns_key, cursor, metadata.version, storage_->IsSlotIdEncoded()).Encode();
+    } else {
+      start_key = match_prefix_key;
+    }
+    for (iter->Seek(start_key); iter->Valid(); iter->Next()) {
+      if (!cursor.empty() && iter->key() == start_key) {
+        // if cursor is not empty, then we need to skip start_key
+        // because we already return that key in the last scan
+        continue;
+      }
+      if (!iter->key().starts_with(match_prefix_key)) {
+        break;
+      }
+      InternalKey ikey(iter->key(), storage_->IsSlotIdEncoded());
+      keys->emplace_back(ikey.GetSubKey().ToString());
+      if (values != nullptr) {
+        Slice value(iter->value());
+        s = value_adapter(metadata, &value);
+        if (!s.ok()) return s;
+        values->emplace_back(value.data(), value.size());
+      }
+      cnt++;
+      if (limit > 0 && cnt >= limit) {
+        break;
+      }
+    }
+    return iter->status();
+  }
 };
 
 class WriteBatchLogData {

src/storage/redis_metadata.cc

@@ -38,6 +38,7 @@ const int VersionCounterBits = 11;
 static std::atomic<uint64_t> version_counter_ = 0;
 
 constexpr const char *kErrMetadataTooShort = "metadata is too short";
+constexpr const char *kErrHashSubkeyValueTooShort = "hash subkey value is too short";
 
 InternalKey::InternalKey(Slice input, bool slot_id_encoded) : slot_id_encoded_(slot_id_encoded) {
   uint32_t key_size = 0;
@@ -339,6 +340,80 @@ bool Metadata::IsEmptyableType() const {
 
 bool Metadata::Expired() const { return ExpireAt(util::GetTimeStampMS()); }
 
+void HashMetadata::Encode(std::string *dst) const {
+  Metadata::Encode(dst);
+  if (IsLegacySubkeyEncoding()) {
+    return;
+  }
+
+  PutFixed8(dst, static_cast<uint8_t>(mode));
+  PutFixed64(dst, expsz);
+  PutFixed64(dst, lower);
+  PutFixed64(dst, upper);
+}
+
+rocksdb::Status HashMetadata::Decode(Slice *input) {
+  if (auto s = Metadata::Decode(input); !s.ok()) {
+    return s;
+  }
+
+  if (input->empty()) {
+    mode = HashSubkeyEncodingMode::kLegacy;
+    expsz = 0;
+    lower = 0;
+    upper = 0;
+    return rocksdb::Status::OK();
+  }
+
+  if (input->size() < 1 + 8 + 8 + 8) {
+    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+  }
+
+  uint8_t encoded_mode = 0;
+  GetFixed8(input, &encoded_mode);
+  if (encoded_mode > static_cast<uint8_t>(HashSubkeyEncodingMode::kFieldExpiration)) {
+    return rocksdb::Status::InvalidArgument("invalid hash subkey encoding mode");
+  }
+
+  mode = static_cast<HashSubkeyEncodingMode>(encoded_mode);
+  GetFixed64(input, &expsz);
+  GetFixed64(input, &lower);
+  GetFixed64(input, &upper);
+  return rocksdb::Status::OK();
+}
+
+std::string HashMetadata::EncodeSubkeyValue(Slice value, uint64_t expire) const {
+  if (IsLegacySubkeyEncoding()) {
+    return value.ToString();
+  }
+
+  std::string encoded;
+  encoded.reserve(kFieldExpirationPrefixSize + value.size());
+  PutFixed64(&encoded, expire);
+  encoded.append(value.data(), value.size());
+  return encoded;
+}
+
+rocksdb::Status HashMetadata::DecodeSubkeyValue(Slice *value, uint64_t *expire) const {
+  if (IsLegacySubkeyEncoding()) {
+    if (expire != nullptr) {
+      *expire = 0;
+    }
+    return rocksdb::Status::OK();
+  }
+
+  if (value->size() < kFieldExpirationPrefixSize) {
+    return rocksdb::Status::InvalidArgument(kErrHashSubkeyValueTooShort);
+  }
+
+  uint64_t encoded_expire = 0;
+  GetFixed64(value, &encoded_expire);
+  if (expire != nullptr) {
+    *expire = encoded_expire;
+  }
+  return rocksdb::Status::OK();
+}
+
 ListMetadata::ListMetadata(bool generate_version)
     : Metadata(kRedisList, generate_version), head(UINT64_MAX / 2), tail(head) {}
 

src/storage/redis_metadata.h

@@ -103,6 +103,16 @@ enum RedisCommand {
 constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
 constexpr const char *kErrMsgKeyExpired = "the key was expired";
 
+enum class HashSubkeyEncodingMode : uint8_t {
+  kLegacy = 0,
+  kFieldExpiration = 1,
+};
+
+enum class HashLengthMode : uint8_t {
+  kAccurate = 0,
+  kApproximate = 1,
+};
+
 using rocksdb::Slice;
 
 struct KeyNumStats {
@@ -211,7 +221,25 @@ class Metadata {
 
 class HashMetadata : public Metadata {
  public:
-  explicit HashMetadata(bool generate_version = true) : Metadata(kRedisHash, generate_version) {}
+  static constexpr size_t kFieldExpirationPrefixSize = sizeof(uint64_t);
+
+  HashSubkeyEncodingMode mode = HashSubkeyEncodingMode::kLegacy;
+  uint64_t expsz = 0;
+  uint64_t lower = 0;
+  uint64_t upper = 0;
+
+  explicit HashMetadata(bool generate_version = true, HashSubkeyEncodingMode mode = HashSubkeyEncodingMode::kLegacy)
+      : Metadata(kRedisHash, generate_version), mode(mode) {}
+
+  bool IsLegacySubkeyEncoding() const { return mode == HashSubkeyEncodingMode::kLegacy; }
+  bool IsFieldExpirationEncoding() const { return mode == HashSubkeyEncodingMode::kFieldExpiration; }
+
+  [[nodiscard]] std::string EncodeSubkeyValue(Slice value, uint64_t expire = 0) const;
+  [[nodiscard]] rocksdb::Status DecodeSubkeyValue(Slice *value, uint64_t *expire = nullptr) const;
+
+  void Encode(std::string *dst) const override;
+  using Metadata::Decode;
+  rocksdb::Status Decode(Slice *input) override;
 };
 
 class SetMetadata : public Metadata {