feat(tdigest): Implement TDIGEST.CDF command

src/commands/cmd_tdigest.cc

@@ -358,7 +358,51 @@ class CommandTDigestMerge : public Commander {
   std::vector<std::string> source_keys_;
   TDigestMergeOptions options_;
 };
+class CommandTDigestCDF : public Commander {
+  Status Parse(const std::vector<std::string> &args) override {
+    key_name_ = args[1];
+    if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
+    values_.reserve(args.size() - 2);
+    for (size_t i = 2; i < args.size(); i++) {
+      auto value = ParseFloat(args[i]);
+      if (!value) {
+        return {Status::RedisParseErr, errValueIsNotFloat};
+      }
+      values_.push_back(*value);
+    }
+    return Status::OK();
+  }
+  Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    TDigest tdigest(srv->storage, conn->GetNamespace());
+    std::vector<std::string> cdf_result;
+    TDigestCDFResult result;
+    TDigestMetadata metadata;
+    auto meta_status = tdigest.GetMetaData(ctx, key_name_, &metadata);
+    if (!meta_status.ok()) {
+      if (meta_status.IsNotFound()) {
+        return {Status::RedisExecErr, errKeyNotFound};
+      }
+      return {Status::RedisExecErr, meta_status.ToString()};
+    }
+    if (metadata.total_observations == 0) {
+      *output = redis::MultiBulkString(RESP::v2, cdf_result);
+      return Status::OK();
+    }
+    auto s = tdigest.CDF(ctx, key_name_, values_, &result);
+    if (!s.ok()) {
+      return {Status::RedisExecErr, s.ToString()};
+    }
+    for (const auto &val : result.cdf_values) {
+      cdf_result.push_back(std::to_string(val));
+    }
+    *output = redis::MultiBulkString(RESP::v2, cdf_result);
+    return Status::OK();
+  }
 
+ private:
+  std::string key_name_;
+  std::vector<double> values_;
+};
 std::vector<CommandKeyRange> GetMergeKeyRange(const std::vector<std::string> &args) {
   auto numkeys = ParseInt<int>(args[2], 10).ValueOr(0);
   return {{1, 1, 1}, {3, 2 + numkeys, 1}};
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
                         MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
-                        MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+                        MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+                        MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
 }  // namespace redis

src/types/redis_tdigest.cc

@@ -414,6 +414,66 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest,
 
   return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
 }
+rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector<double>& inputs,
+                             TDigestCDFResult* result) {
+  auto ns_key = AppendNamespacePrefix(digest_name);
+  TDigestMetadata metadata;
+  {
+    LockGuard guard(storage_->GetLockManager(), ns_key);
+
+    if (auto status = getMetaDataByNsKey(ctx, ns_key, &metadata); !status.ok()) {
+      return status;
+    }
+
+    if (metadata.unmerged_nodes > 0) {
+      auto batch = storage_->GetWriteBatchBase();
+      WriteBatchLogData log_data(kRedisTDigest);
+      if (auto status = batch->PutLogData(log_data.Encode()); !status.ok()) {
+        return status;
+      }
+
+      if (auto status = mergeCurrentBuffer(ctx, ns_key, batch, &metadata); !status.ok()) {
+        return status;
+      }
+
+      std::string metadata_bytes;
+      metadata.Encode(&metadata_bytes);
+      if (auto status = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); !status.ok()) {
+        return status;
+      }
+
+      if (auto status = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); !status.ok()) {
+        return status;
+      }
+      ctx.RefreshLatestSnapshot();
+    }
+  }
+  std::vector<Centroid> centroids;
+  if (auto status = dumpCentroids(ctx, ns_key, metadata, &centroids); !status.ok()) {
+    return status;
+  }
+  auto dump_centroids = DummyCentroids(metadata, centroids);
+  double total_weight = dump_centroids.TotalWeight();
+  std::vector<double> results;
+  for (double val : inputs) {
+    auto iter_begin = dump_centroids.Begin();
+    auto iter_end = dump_centroids.End();
+    double eq_count = 0;
+    double smaller_count = 0;
+    for (; iter_begin->Valid(); iter_begin->Next()) {
+      auto current_centroid = iter_begin->GetCentroid();
+      if (val > current_centroid->mean) {
+        smaller_count++;
+      } else if (val == current_centroid->mean) {
+        eq_count++;
+      }
+    }
+    double cdf_val = (smaller_count / total_weight) + ((eq_count / 2) / total_weight);
+    results.push_back(cdf_val);
+  }
+  result->cdf_values = results;
+  return rocksdb::Status::OK();
+}
 
 rocksdb::Status TDigest::GetMetaData(engine::Context& context, const Slice& digest_name, TDigestMetadata* metadata) {
   auto ns_key = AppendNamespacePrefix(digest_name);

src/types/redis_tdigest.h

@@ -48,7 +48,9 @@ struct TDigestMergeOptions {
   uint32_t compression = 0;
   bool override_flag = false;
 };
-
+struct TDigestCDFResult {
+  std::vector<double> cdf_values;
+};
 struct TDigestQuantitleResult {
   std::optional<std::vector<double>> quantiles;
 };
@@ -79,6 +81,8 @@ class TDigest : public SubKeyScanner {
                         const TDigestMergeOptions& options);
 
   rocksdb::Status GetMetaData(engine::Context& context, const Slice& digest_name, TDigestMetadata* metadata);
+  rocksdb::Status CDF(engine::Context& ctx, const Slice& digest_name, const std::vector<double>& inputs,
+                      TDigestCDFResult* result);
 
  private:
   enum class SegmentType : uint8_t { kBuffer = 0, kCentroids = 1, kGuardFlag = 0xFF };