feat(hyperloglog): add support of the Hyperloglog data structure

src/commands/cmd_hll.cc

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <types/redis_hyperloglog.h>
+
+#include <algorithm>
+
+#include "commander.h"
+#include "commands/command_parser.h"
+#include "commands/error_constants.h"
+#include "error_constants.h"
+#include "parse_util.h"
+#include "server/redis_reply.h"
+#include "server/server.h"
+#include "storage/redis_metadata.h"
+
+namespace redis {
+
+/// PFADD key [element [element ...]]
+/// Complexity: O(1) for each element added.
+class CommandPfAdd final : public Commander {
+ public:
+  Status Execute(Server *srv, Connection *conn, std::string *output) override {
+    redis::HyperLogLog hll(srv->storage, conn->GetNamespace());
+    std::vector<uint64_t> hashes(args_.size() - 1);
+    for (size_t i = 1; i < args_.size(); i++) {
+      hashes[i - 1] = redis::HyperLogLog::HllHash(args_[i]);
+    }
+    uint64_t ret{};
+    auto s = hll.Add(args_[0], hashes, &ret);
+    if (!s.ok() && !s.IsNotFound()) {
+      return {Status::RedisExecErr, s.ToString()};
+    }
+    *output = redis::Integer(ret);
+    return Status::OK();
+  }
+};
+
+/// PFCOUNT key [key ...]
+/// Complexity: O(1) with a very small average constant time when called with a single key.
+///              O(N) with N being the number of keys, and much bigger constant times,
+///              when called with multiple keys.
+///
+/// TODO(mwish): Currently we don't supports merge, so only one key is supported.
+class CommandPfCount final : public Commander {
+  Status Execute(Server *srv, Connection *conn, std::string *output) override {
+    redis::HyperLogLog hll(srv->storage, conn->GetNamespace());
+    uint64_t ret{};
+    auto s = hll.Count(args_[0], &ret);
+    if (!s.ok() && !s.IsNotFound()) {
+      return {Status::RedisExecErr, s.ToString()};
+    }
+    if (s.IsNotFound()) {
+      ret = 0;
+    }
+    *output = redis::Integer(ret);
+    return Status::OK();
+  }
+};
+
+REDIS_REGISTER_COMMANDS(MakeCmdAttr<CommandPfAdd>("pfadd", -2, "write", 1, 1, 1),
+                        MakeCmdAttr<CommandPfCount>("pfcount", 2, "read-only", 1, 1, 1), );
+
+}  // namespace redis

src/storage/redis_metadata.cc

@@ -329,7 +329,7 @@ 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;
+  return IsSingleKVType() || Type() == kRedisStream || Type() == kRedisBloomFilter || Type() == kRedisHyperLogLog;
 }
 
 bool Metadata::Expired() const { return ExpireAt(util::GetTimeStampMS()); }
@@ -472,3 +472,26 @@ rocksdb::Status JsonMetadata::Decode(Slice *input) {
 
   return rocksdb::Status::OK();
 }
+
+void HyperLogLogMetadata::Encode(std::string *dst) const {
+  Metadata::Encode(dst);
+  PutFixed8(dst, static_cast<uint8_t>(this->encode_type));
+}
+
+rocksdb::Status HyperLogLogMetadata::Decode(Slice *input) {
+  if (auto s = Metadata::Decode(input); !s.ok()) {
+    return s;
+  }
+
+  uint8_t encoded_type = 0;
+  if (!GetFixed8(input, &encoded_type)) {
+    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+  }
+  // Check validity of encode type
+  if (encoded_type > 0) {
+    return rocksdb::Status::InvalidArgument(fmt::format("Invalid encode type {}", encoded_type));
+  }
+  this->encode_type = static_cast<EncodeType>(encoded_type);
+
+  return rocksdb::Status::OK();
+}

src/storage/redis_metadata.h

@@ -49,6 +49,7 @@ enum RedisType : uint8_t {
   kRedisStream = 8,
   kRedisBloomFilter = 9,
   kRedisJson = 10,
+  kRedisHyperLogLog = 11,
 };
 
 struct RedisTypes {
@@ -90,8 +91,9 @@ enum RedisCommand {
   kRedisCmdLMove,
 };
 
-const std::vector<std::string> RedisTypeNames = {"none",   "string",    "hash",   "list",      "set",      "zset",
-                                                 "bitmap", "sortedint", "stream", "MBbloom--", "ReJSON-RL"};
+const std::vector<std::string> RedisTypeNames = {"none",   "string",    "hash",      "list",
+                                                 "set",    "zset",      "bitmap",    "sortedint",
+                                                 "stream", "MBbloom--", "ReJSON-RL", "hyperloglog"};
 
 constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
 constexpr const char *kErrMsgKeyExpired = "the key was expired";
@@ -313,3 +315,23 @@ class JsonMetadata : public Metadata {
   void Encode(std::string *dst) const override;
   rocksdb::Status Decode(Slice *input) override;
 };
+
+class HyperLogLogMetadata : public Metadata {
+ public:
+  enum class EncodeType : uint8_t {
+    // Redis-style dense encoding implement as bitmap like sub keys to
+    // store registers by segment in data column family.
+    // The registers are stored in 6-bit format and each segment contains
+    // 768 registers.
+    DENSE = 0,
+    // TODO(mwish): sparse encoding
+    // SPARSE = 1,
+  };
+
+  explicit HyperLogLogMetadata(bool generate_version = true) : Metadata(kRedisHyperLogLog, generate_version) {}
+
+  void Encode(std::string *dst) const override;
+  rocksdb::Status Decode(Slice *input) override;
+
+  EncodeType encode_type = EncodeType::DENSE;
+};

src/storage/storage.h

@@ -42,6 +42,10 @@
 #include "observer_or_unique.h"
 #include "status.h"
 
+#if defined(__sparc__) || defined(__arm__)
+#define USE_ALIGNED_ACCESS
+#endif
+
 enum class StorageEngineType : uint16_t {
   RocksDB,
   Speedb,