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

src/storage/redis_metadata.cc

@@ -30,6 +30,7 @@
 
 #include "cluster/redis_slot.h"
 #include "encoding.h"
+#include "rocksdb/status.h"
 #include "time_util.h"
 
 // 52 bit for microseconds and 11 bit for counter
@@ -331,7 +332,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 +499,40 @@ rocksdb::Status HyperLogLogMetadata::Decode(Slice *input) {
 
   return rocksdb::Status::OK();
 }
+
+void TDigestMetadata::Encode(std::string *dst) const {
+  Metadata::Encode(dst);
+  PutFixed64(dst, compression);
+  PutFixed64(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(uint64_t) * 8) {
+    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+  }
+
+  GetFixed64(input, &compression);
+  GetFixed64(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

@@ -25,6 +25,7 @@
 #include <atomic>
 #include <bitset>
 #include <initializer_list>
+#include <limits>
 #include <string>
 #include <vector>
 
@@ -50,6 +51,7 @@ enum RedisType : uint8_t {
   kRedisBloomFilter = 9,
   kRedisJson = 10,
   kRedisHyperLogLog = 11,
+  kRedisTDigest = 12,
 };
 
 struct RedisTypes {
@@ -91,9 +93,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", "tdigest"};
 
 constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
 constexpr const char *kErrMsgKeyExpired = "the key was expired";
@@ -336,3 +338,27 @@ class HyperLogLogMetadata : public Metadata {
 
   EncodeType encode_type = EncodeType::DENSE;
 };
+
+class TDigestMetadata : public Metadata {
+ public:
+  uint64_t compression;
+  uint64_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>::infinity();
+  double maximum = -1 * std::numeric_limits<double>::infinity();
+  uint64_t total_observations = 0;
+  uint64_t merge_times = 0;
+
+  explicit TDigestMetadata(uint64_t compression, uint64_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); }
+};

src/storage/storage.cc

@@ -362,6 +362,7 @@ Status Storage::Open(DBOpenMode mode) {
   column_families.emplace_back(std::string(kPropagateColumnFamilyName), propagate_opts);
   column_families.emplace_back(std::string(kStreamColumnFamilyName), subkey_opts);
   column_families.emplace_back(std::string(kSearchColumnFamilyName), search_opts);
+  column_families.emplace_back(std::string(kTDigestColumnFamilyName), subkey_opts);
 
   std::vector<std::string> old_column_families;
   auto s = rocksdb::DB::ListColumnFamilies(options, config_->db_dir, &old_column_families);

src/storage/storage.h

@@ -70,9 +70,10 @@ enum class ColumnFamilyID : uint32_t {
   Propagate,
   Stream,
   Search,
+  TDigest,
 };
 
-constexpr uint32_t kMaxColumnFamilyID = static_cast<uint32_t>(ColumnFamilyID::Search);
+constexpr uint32_t kMaxColumnFamilyID = static_cast<uint32_t>(ColumnFamilyID::TDigest);
 
 namespace engine {
 
@@ -147,6 +148,7 @@ constexpr const std::string_view kPubSubColumnFamilyName = "pubsub";
 constexpr const std::string_view kPropagateColumnFamilyName = "propagate";
 constexpr const std::string_view kStreamColumnFamilyName = "stream";
 constexpr const std::string_view kSearchColumnFamilyName = "search";
+constexpr const std::string_view kTDigestColumnFamilyName = "tdigest";
 
 class ColumnFamilyConfigs {
  public:
@@ -185,6 +187,10 @@ class ColumnFamilyConfigs {
     return {ColumnFamilyID::Search, kSearchColumnFamilyName, /*is_minor=*/true};
   }
 
+  static ColumnFamilyConfig TDigestColumnFamily() {
+    return {ColumnFamilyID::Search, kTDigestColumnFamilyName, /*is_minor=*/true};
+  }
+
   /// ListAllColumnFamilies returns all column families in kvrocks.
   static const std::vector<ColumnFamilyConfig> &ListAllColumnFamilies() { return AllCfs; }
 
@@ -196,11 +202,11 @@ class ColumnFamilyConfigs {
   // Caution: don't change the order of column family, or the handle will be mismatched
   inline const static std::vector<ColumnFamilyConfig> AllCfs = {
       PrimarySubkeyColumnFamily(), MetadataColumnFamily(), SecondarySubkeyColumnFamily(), PubSubColumnFamily(),
-      PropagateColumnFamily(),     StreamColumnFamily(),   SearchColumnFamily(),
+      PropagateColumnFamily(),     StreamColumnFamily(),   SearchColumnFamily(),          TDigestColumnFamily(),
   };
   inline const static std::vector<ColumnFamilyConfig> AllCfsWithoutDefault = {
-      MetadataColumnFamily(),  SecondarySubkeyColumnFamily(), PubSubColumnFamily(),
-      PropagateColumnFamily(), StreamColumnFamily(),          SearchColumnFamily(),
+      MetadataColumnFamily(), SecondarySubkeyColumnFamily(), PubSubColumnFamily(),  PropagateColumnFamily(),
+      StreamColumnFamily(),   SearchColumnFamily(),          TDigestColumnFamily(),
   };
 };