feat(tdigest): add tdigest algorithm and storage encoding implementations

src/storage/redis_metadata.cc

@@ -331,7 +331,8 @@ bool Metadata::ExpireAt(uint64_t expired_ts) const {
 bool Metadata::IsSingleKVType() const { return Type() == kRedisString || Type() == kRedisJson; }
 
 bool Metadata::IsEmptyableType() const {
-  return IsSingleKVType() || Type() == kRedisStream || Type() == kRedisBloomFilter || Type() == kRedisHyperLogLog;
+  return IsSingleKVType() || Type() == kRedisStream || Type() == kRedisBloomFilter || Type() == kRedisHyperLogLog ||
+         Type() == kRedisTDigest;
 }
 
 bool Metadata::Expired() const { return ExpireAt(util::GetTimeStampMS()); }
@@ -497,3 +498,40 @@ rocksdb::Status HyperLogLogMetadata::Decode(Slice *input) {
 
   return rocksdb::Status::OK();
 }
+
+void TDigestMetadata::Encode(std::string *dst) const {
+  Metadata::Encode(dst);
+  PutFixed32(dst, compression);
+  PutFixed32(dst, capacity);
+  PutFixed64(dst, unmerged_nodes);
+  PutFixed64(dst, merged_nodes);
+  PutFixed64(dst, total_weight);
+  PutFixed64(dst, merged_weight);
+  PutDouble(dst, minimum);
+  PutDouble(dst, maximum);
+  PutFixed64(dst, total_observations);
+  PutFixed64(dst, merge_times);
+}
+
+rocksdb::Status TDigestMetadata::Decode(Slice *input) {
+  if (auto s = Metadata::Decode(input); !s.ok()) {
+    return s;
+  }
+
+  if (input->size() < (sizeof(uint32_t) * 2 + sizeof(uint64_t) * 6 + sizeof(double) * 2)) {
+    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+  }
+
+  GetFixed32(input, &compression);
+  GetFixed32(input, &capacity);
+  GetFixed64(input, &unmerged_nodes);
+  GetFixed64(input, &merged_nodes);
+  GetFixed64(input, &total_weight);
+  GetFixed64(input, &merged_weight);
+  GetDouble(input, &minimum);
+  GetDouble(input, &maximum);
+  GetFixed64(input, &total_observations);
+  GetFixed64(input, &merge_times);
+
+  return rocksdb::Status::OK();
+}

src/storage/redis_metadata.h

@@ -26,6 +26,7 @@
 #include <atomic>
 #include <bitset>
 #include <initializer_list>
+#include <limits>
 #include <string>
 #include <vector>
 
@@ -51,6 +52,7 @@ enum RedisType : uint8_t {
   kRedisBloomFilter = 9,
   kRedisJson = 10,
   kRedisHyperLogLog = 11,
+  kRedisTDigest = 12,
 };
 
 struct RedisTypes {
@@ -92,9 +94,9 @@ enum RedisCommand {
   kRedisCmdLMove,
 };
 
-const std::vector<std::string> RedisTypeNames = {"none",   "string",    "hash",      "list",
-                                                 "set",    "zset",      "bitmap",    "sortedint",
-                                                 "stream", "MBbloom--", "ReJSON-RL", "hyperloglog"};
+const std::vector<std::string> RedisTypeNames = {"none",      "string",      "hash",      "list",   "set",
+                                                 "zset",      "bitmap",      "sortedint", "stream", "MBbloom--",
+                                                 "ReJSON-RL", "hyperloglog", "TDIS-TYPE"};
 
 constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
 constexpr const char *kErrMsgKeyExpired = "the key was expired";
@@ -337,3 +339,27 @@ class HyperLogLogMetadata : public Metadata {
 
   EncodeType encode_type = EncodeType::DENSE;
 };
+
+class TDigestMetadata : public Metadata {
+ public:
+  uint32_t compression;
+  uint32_t capacity;
+  uint64_t unmerged_nodes = 0;
+  uint64_t merged_nodes = 0;
+  uint64_t total_weight = 0;
+  uint64_t merged_weight = 0;
+  double minimum = std::numeric_limits<double>::max();
+  double maximum = std::numeric_limits<double>::lowest();
+  uint64_t total_observations = 0;
+  uint64_t merge_times = 0;
+
+  explicit TDigestMetadata(uint32_t compression, uint32_t capacity, bool generate_version = true)
+      : Metadata(kRedisTDigest, generate_version), compression(compression), capacity(capacity) {}
+  explicit TDigestMetadata(bool generate_version = true) : TDigestMetadata(0, 0, generate_version) {}
+  void Encode(std::string *dst) const override;
+  rocksdb::Status Decode(Slice *input) override;
+
+  uint64_t TotalNodes() const { return merged_nodes + unmerged_nodes; }
+
+  double Delta() const { return 1. / static_cast<double>(compression); }
+};