feat(rdb): add DUMP support for HyperLogLog type

src/storage/rdb/rdb.cc

@@ -35,6 +35,7 @@
 #include "types/redis_bitmap.h"
 #include "types/redis_bitmap_string.h"
 #include "types/redis_hash.h"
+#include "types/redis_hyperloglog.h"
 #include "types/redis_list.h"
 #include "types/redis_set.h"
 #include "types/redis_sortedint.h"
@@ -723,7 +724,7 @@ Status RDB::Dump(const std::string &key, const RedisType type) {
 
 Status RDB::SaveObjectType(const RedisType type) {
   int robj_type = -1;
-  if (type == kRedisString || type == kRedisBitmap) {
+  if (type == kRedisString || type == kRedisBitmap || type == kRedisHyperLogLog) {
     robj_type = RDBTypeString;
   } else if (type == kRedisHash) {
     robj_type = RDBTypeHash;
@@ -815,6 +816,12 @@ Status RDB::SaveObject(const std::string &key, const RedisType type) {
       return {Status::RedisExecErr, si_status.ToString()};
     }
     return SaveSortedintObject(ids);
+  } else if (type == kRedisHyperLogLog) {
+    redis::HyperLogLog hll_db(storage_, ns_);
+    std::string value;
+    auto s = hll_db.Get(ctx, key, &value);
+    if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+    return SaveStringObject(value);
   } else {
     WARN("Invalid or Not supported object type: {}", (int)type);
     return {Status::NotOK, "Invalid or Not supported object type"};

src/types/redis_hyperloglog.cc

@@ -330,4 +330,26 @@ rocksdb::Status HyperLogLog::getRegisters(engine::Context &ctx, const Slice &ns_
   return rocksdb::Status::OK();
 }
 
+rocksdb::Status HyperLogLog::Get(engine::Context &ctx, const Slice &user_key, std::string *value) {
+  std::string ns_key = AppendNamespacePrefix(user_key);
+  std::vector<std::string> register_segments;
+  auto s = getRegisters(ctx, ns_key, &register_segments);
+  if (!s.ok()) return s;
+
+  value->clear();
+  value->append("HYLL", 4);
+  value->push_back(0);
+  value->append(3, 0);
+  value->append(8, 0);
+
+  for (const auto &segment : register_segments) {
+    if (segment.empty()) {
+      value->append(kHyperLogLogSegmentBytes, 0);
+    } else {
+      value->append(segment);
+    }
+  }
+  return rocksdb::Status::OK();
+}
+
 }  // namespace redis

src/types/redis_hyperloglog.h

@@ -28,6 +28,7 @@ namespace redis {
 class HyperLogLog : public Database {
  public:
   explicit HyperLogLog(engine::Storage *storage, const std::string &ns) : Database(storage, ns) {}
+  rocksdb::Status Get(engine::Context &ctx, const Slice &user_key, std::string *value);
   rocksdb::Status Add(engine::Context &ctx, const Slice &user_key, const std::vector<uint64_t> &element_hashes,
                       uint64_t *ret);
   rocksdb::Status Count(engine::Context &ctx, const Slice &user_key, uint64_t *ret);

tests/gocase/unit/dump/dump_test.go

@@ -220,3 +220,30 @@ func TestDump_SortedInt(t *testing.T) {
 	expectedMembers := []string{"5", "12", "23", "89", "100"}
 	require.ElementsMatch(t, expectedMembers, members)
 }
+
+func TestDump_HyperLogLog(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	key := "test_hyperloglog_key"
+	require.NoError(t, rdb.Del(ctx, key).Err())
+
+	require.EqualValues(t, 1, rdb.Do(ctx, "PFADD", key, "a", "b", "c").Val())
+	require.EqualValues(t, 3, rdb.Do(ctx, "PFCOUNT", key).Val())
+
+	serialized, err := rdb.Dump(ctx, key).Result()
+	require.NoError(t, err)
+
+	restoredKey := fmt.Sprintf("restore_%s", key)
+	require.NoError(t, rdb.RestoreReplace(ctx, restoredKey, 0, serialized).Err())
+
+	require.EqualValues(t, "string", rdb.Type(ctx, restoredKey).Val())
+
+	content := rdb.Get(ctx, restoredKey).Val()
+	require.Equal(t, 12304, len(content), "HLL data should be 12304 bytes (Header + Registers)")
+
+	require.Equal(t, "HYLL", content[:4], "Should have Redis HLL Header")
+}