src/commands/cmd_tdigest.cc - kvrocks - Git at Google

 /*
  * 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 <range/v3/range/conversion.hpp>
 #include <range/v3/view/transform.hpp>

 #include "command_parser.h"
 #include "commander.h"
 #include "commands/error_constants.h"
 #include "parse_util.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
 #include "status.h"
 #include "string_util.h"
 #include "types/redis_tdigest.h"

 namespace redis {
 namespace {
 constexpr auto kCompressionArg = "compression";
 constexpr auto kOverrideArg = "override";

 constexpr auto kInfoCompression = "Compression";
 constexpr auto kInfoCapacity = "Capacity";
 constexpr auto kInfoMergedNodes = "Merged nodes";
 constexpr auto kInfoUnmergedNodes = "Unmerged nodes";
 constexpr auto kInfoMergedWeight = "Merged weight";
 constexpr auto kInfoUnmergedWeight = "Unmerged weight";
 constexpr auto kInfoObservations = "Observations";
 constexpr auto kInfoTotalCompressions = "Total compressions";
 constexpr auto kNan = "nan";

 constexpr const char *errParseLowCutQuantile = "error parsing low_cut_percentile";
 constexpr const char *errParseHighCutQuantile = "error parsing high_cut_percentile";
 constexpr const char *errCutQuantileRange = "low_cut_percentile and high_cut_percentile should be in [0,1]";
 constexpr const char *errLowCutQuantileLess = "low_cut_percentile should be lower than high_cut_percentile";
 }  // namespace

 class CommandTDigestCreate : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     CommandParser parser(args, 1);
     key_name_ = GET_OR_RET(parser.TakeStr());
     options_.compression = 100;
     if (parser.EatEqICase(kCompressionArg)) {
       if (!parser.Good()) {
         return {Status::RedisParseErr, errWrongNumOfArguments};
       }
       auto status = parser.TakeInt<int32_t>();
       if (!status) {
         return {Status::RedisParseErr, errParseCompression};
       }
       auto compression = *status;
       if (compression <= 0) {
         return {Status::RedisParseErr, errCompressionMustBePositive};
       }
       if (compression < 1 || compression > static_cast<int32_t>(kTDigestMaxCompression)) {
         return {Status::RedisParseErr, errCompressionOutOfRange};
       }
       options_.compression = static_cast<uint32_t>(compression);
     }
     if (parser.Good()) {
       return {Status::RedisParseErr, errWrongNumOfArguments};
     }

     return Status::OK();
   }

   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     bool exists = false;
     auto s = tdigest.Create(ctx, key_name_, options_, &exists);
     if (!s.ok()) {
       if (exists) {
         return {Status::RedisExecErr, errKeyAlreadyExists};
       }
       return {Status::RedisExecErr, s.ToString()};
     }
     *output = redis::RESP_OK;
     return Status::OK();
   }

  private:
   std::string key_name_;
   TDigestCreateOptions options_;
 };

 class CommandTDigestInfo : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];
     return Status::OK();
   }

   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     TDigestMetadata metadata;
     auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     output->append(conn->HeaderOfMap(8));
     output->append(redis::BulkString(kInfoCompression));
     output->append(redis::Integer(metadata.compression));
     output->append(redis::BulkString(kInfoCapacity));
     output->append(redis::Integer(metadata.capacity));
     output->append(redis::BulkString(kInfoMergedNodes));
     output->append(redis::Integer(metadata.merged_nodes));
     output->append(redis::BulkString(kInfoUnmergedNodes));
     output->append(redis::Integer(metadata.unmerged_nodes));
     output->append(redis::BulkString(kInfoMergedWeight));
     output->append(redis::Integer(metadata.merged_weight));
     output->append(redis::BulkString(kInfoUnmergedWeight));
     output->append(redis::Integer(metadata.total_weight - metadata.merged_weight));
     output->append(redis::BulkString(kInfoObservations));
     output->append(redis::Integer(metadata.total_observations));
     output->append(redis::BulkString(kInfoTotalCompressions));
     output->append(redis::Integer(metadata.merge_times));
     // "Memory usage" is not meaningful for kvrocks, so we don't provide it here.
     return Status::OK();
   }

  private:
   std::string key_name_;
 };

 class CommandTDigestAdd : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];

     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());

     auto s = tdigest.Add(ctx, key_name_, values_);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     *output = redis::RESP_OK;
     return Status::OK();
   }

  private:
   std::string key_name_;
   std::vector<double> values_;
 };

 template <bool Reverse>
 class TDigestRankCommand : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];

     std::set<std::string> unique_inputs_set(args.begin() + 2, args.end());
     origin_inputs_.assign(args.begin() + 2, args.end());

     unique_inputs_.reserve(unique_inputs_set.size());
     size_t i = 0;
     for (const auto &input : unique_inputs_set) {
       auto value = ParseFloat(input);
       if (!value) {
         return {Status::RedisParseErr, errValueIsNotFloat};
       }
       unique_inputs_.push_back(*value);
       unique_inputs_order_[input] = i;
       ++i;
     }
     return Status::OK();
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     std::vector<int> result;
     result.reserve(origin_inputs_.size());

     if (const auto s =
             [&]() {
               if constexpr (Reverse) {
                 return tdigest.RevRank(ctx, key_name_, unique_inputs_, &result);
               } else {
                 return tdigest.Rank(ctx, key_name_, unique_inputs_, &result);
               }
             }();
         !s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     std::vector<std::string> rev_ranks;
     rev_ranks.reserve(origin_inputs_.size());
     for (const auto &v : origin_inputs_) {
       rev_ranks.push_back(redis::Integer(result[unique_inputs_order_[v]]));
     }
     *output = redis::Array(rev_ranks);
     return Status::OK();
   }

  private:
   std::string key_name_;
   std::vector<double> unique_inputs_;
   std::map<std::string, size_t> unique_inputs_order_;
   std::vector<std::string> origin_inputs_;
 };

 class CommandTDigestRevRank : public TDigestRankCommand<true> {};

 class CommandTDigestRank : public TDigestRankCommand<false> {};

 template <bool Reverse>
 class TDigestByRankCommand : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];

     std::set<std::string> unique_inputs_set(args.begin() + 2, args.end());
     origin_inputs_.assign(args.begin() + 2, args.end());

     unique_inputs_.reserve(unique_inputs_set.size());
     size_t i = 0;
     for (const auto &input : unique_inputs_set) {
       auto value = ParseInt<int>(input);
       if (!value) {
         return {Status::RedisParseErr, errValueNotInteger};
       }
       if (*value < 0) {
         return {Status::InvalidArgument, errInvalidRankValue};
       }
       unique_inputs_.push_back(*value);
       unique_inputs_order_[input] = i;
       ++i;
     }
     return Status::OK();
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     std::vector<double> result;
     result.reserve(origin_inputs_.size());

     if (const auto s =
             [&]() {
               if constexpr (Reverse) {
                 return tdigest.ByRevRank(ctx, key_name_, unique_inputs_, &result);
               } else {
                 return tdigest.ByRank(ctx, key_name_, unique_inputs_, &result);
               }
             }();
         !s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     std::vector<std::string> ranks;
     ranks.reserve(origin_inputs_.size());
     auto is_resp3 = conn->GetProtocolVersion() == RESP::v3;
     for (const auto &v : origin_inputs_) {
       auto rank_value = result[unique_inputs_order_[v]];
       if (is_resp3) {
         ranks.push_back(redis::Double(redis::RESP::v3, rank_value));
       } else {
         ranks.push_back(redis::BulkString(fmt::format("{}", rank_value)));
       }
     }
     *output = redis::Array(ranks);
     return Status::OK();
   }

  private:
   std::string key_name_;
   std::vector<int> unique_inputs_;
   std::map<std::string, size_t> unique_inputs_order_;
   std::vector<std::string> origin_inputs_;
 };

 class CommandTDigestByRevRank : public TDigestByRankCommand<true> {};

 class CommandTDigestByRank : public TDigestByRankCommand<false> {};

 class CommandTDigestMinMax : public Commander {
  public:
   explicit CommandTDigestMinMax(bool is_min) : is_min_(is_min) {}

   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];
     return Status::OK();
   }

   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     TDigestMetadata metadata;
     auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     if (metadata.total_observations == 0) {
       *output = redis::BulkString("nan");
       return Status::OK();
     }

     double value = is_min_ ? metadata.minimum : metadata.maximum;
     *output = redis::BulkString(fmt::format("{}", value));
     return Status::OK();
   }

  private:
   std::string key_name_;
   bool is_min_;
 };
 class CommandTDigestReset : public Commander {
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];
     return Status::OK();
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     TDigestMetadata metadata;
     auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }
     if (metadata.total_observations == 0) {
       *output = redis::RESP_OK;
       return Status::OK();
     }
     s = tdigest.Reset(ctx, key_name_);
     if (!s.ok()) {
       return {Status::RedisExecErr, s.ToString()};
     }
     *output = redis::RESP_OK;
     return Status::OK();
   }

  private:
   std::string key_name_;
 };
 // Then replace the existing template implementation and type aliases with:
 class CommandTDigestMin : public CommandTDigestMinMax {
  public:
   CommandTDigestMin() : CommandTDigestMinMax(true) {}
 };

 class CommandTDigestMax : public CommandTDigestMinMax {
  public:
   CommandTDigestMax() : CommandTDigestMinMax(false) {}
 };
 class CommandTDigestQuantile : public Commander {
   Status Parse(const std::vector<std::string> &args) override {
     key_name_ = args[1];
     quantiles_.reserve(args.size() - 2);
     for (size_t i = 2; i < args.size(); i++) {
       auto value = ParseFloat(args[i]);
       if (!value) {
         return {Status::RedisParseErr, errValueIsNotFloat};
       }
       quantiles_.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());
     TDigestQuantitleResult result;
     auto s = tdigest.Quantile(ctx, key_name_, quantiles_, &result);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }
     auto quantile_strings = result.quantiles
                                 ? (ranges::views::transform(*result.quantiles, util::Float2String) | ranges::to_vector)
                                 : std::vector<std::string>(quantiles_.size(), kNan);
     *output = conn->MultiBulkString(quantile_strings);
     return Status::OK();
   }

  private:
   std::string key_name_;
   std::vector<double> quantiles_;
 };

 class CommandTDigestMerge : public Commander {
   Status Parse(const std::vector<std::string> &args) override {
     CommandParser parser(args, 1);
     dest_key_ = GET_OR_RET(parser.TakeStr());
     auto numkeys = parser.TakeInt();
     if (!numkeys) {
       return {Status::RedisParseErr, errParsingNumkeys};
     }

     if (*numkeys <= 0) {
       return {Status::RedisParseErr, errNumkeysMustBePositive};
     }

     if (static_cast<int64_t>(args.size()) < (3 + *numkeys)) {
       return {Status::RedisParseErr, errWrongNumOfArguments};
     }

     source_keys_.reserve(*numkeys);

     for (auto i = 3; i < (3 + *numkeys); i++) {
       auto src_digest = GET_OR_RET(parser.TakeStr());
       source_keys_.emplace_back(std::move(src_digest));
     }

     while (parser.Good()) {
       // more arguments than expected compression and override
       if (options_.compression > 0 && options_.override_flag) {
         return {Status::RedisParseErr, errWrongNumOfArguments};
       }

       if (parser.EatEqICase(kCompressionArg)) {
         // compression already set or without a compression value
         if (options_.compression > 0 || !parser.Good()) {
           return {Status::RedisParseErr, errWrongNumOfArguments};
         }

         if (auto compression = parser.TakeInt<uint32_t>(); !compression) {
           return {Status::RedisParseErr, errParseCompression};
         } else if (*compression <= 0 || *compression > kTDigestMaxCompression) {
           return {Status::RedisParseErr, errCompressionOutOfRange};
         } else {
           options_.compression = *compression;
         }
       } else if (parser.EatEqICase(kOverrideArg)) {
         if (options_.override_flag) {  // override already set
           return {Status::RedisParseErr, errWrongNumOfArguments};
         }
         options_.override_flag = true;
       } else {
         return {Status::RedisParseErr, errWrongKeyword};
       }
     }

     return Status::OK();
   }

   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     auto s = tdigest.Merge(ctx, dest_key_, source_keys_, options_);
     if (!s.ok()) {
       return {Status::RedisExecErr, s.IsNotFound() ? errKeyNotFound : s.ToString()};
     }
     *output = redis::RESP_OK;
     return Status::OK();
   }

  private:
   std::string dest_key_;
   std::vector<std::string> source_keys_;
   TDigestMergeOptions options_;
 };

 class CommandTDigestTrimmedMean : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override {
     if (args.size() != 4) {
       return {Status::RedisParseErr, errWrongNumOfArguments};
     }

     key_name_ = args[1];

     auto low_cut_quantile = ParseFloat(args[2]);
     if (!low_cut_quantile || std::isnan(*low_cut_quantile)) {
       return {Status::RedisParseErr, errParseLowCutQuantile};
     }
     low_cut_quantile_ = *low_cut_quantile;

     auto high_cut_quantile = ParseFloat(args[3]);
     if (!high_cut_quantile || std::isnan(*high_cut_quantile)) {
       return {Status::RedisParseErr, errParseHighCutQuantile};
     }
     high_cut_quantile_ = *high_cut_quantile;

     if (!std::isfinite(low_cut_quantile_) || low_cut_quantile_ < 0.0 || low_cut_quantile_ > 1.0) {
       return {Status::RedisParseErr, errCutQuantileRange};
     }
     if (!std::isfinite(high_cut_quantile_) || high_cut_quantile_ < 0.0 || high_cut_quantile_ > 1.0) {
       return {Status::RedisParseErr, errCutQuantileRange};
     }
     if (low_cut_quantile_ >= high_cut_quantile_) {
       return {Status::RedisParseErr, errLowCutQuantileLess};
     }

     return Status::OK();
   }

   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     TDigest tdigest(srv->storage, conn->GetNamespace());
     TDigestTrimmedMeanResult result;

     auto s = tdigest.TrimmedMean(ctx, key_name_, low_cut_quantile_, high_cut_quantile_, &result);
     if (!s.ok()) {
       if (s.IsNotFound()) {
         return {Status::RedisExecErr, errKeyNotFound};
       }
       return {Status::RedisExecErr, s.ToString()};
     }

     if (!result.mean.has_value()) {
       *output = redis::BulkString(kNan);
     } else {
       *output = redis::BulkString(util::Float2String(*result.mean));
     }

     return Status::OK();
   }

  private:
   std::string key_name_;
   double low_cut_quantile_;
   double high_cut_quantile_;
 };

 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}};
 }

 REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.create", -2, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestInfo>("tdigest.info", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestAdd>("tdigest.add", -3, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestMax>("tdigest.max", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestRevRank>("tdigest.revrank", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestRank>("tdigest.rank", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestByRevRank>("tdigest.byrevrank", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestByRank>("tdigest.byrank", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestTrimmedMean>("tdigest.trimmed_mean", 4, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
 }  // namespace redis
	/*
	* 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 <range/v3/range/conversion.hpp>
	#include <range/v3/view/transform.hpp>

	#include "command_parser.h"
	#include "commander.h"
	#include "commands/error_constants.h"
	#include "parse_util.h"
	#include "server/redis_reply.h"
	#include "server/server.h"
	#include "status.h"
	#include "string_util.h"
	#include "types/redis_tdigest.h"

	namespace redis {
	namespace {
	constexpr auto kCompressionArg = "compression";
	constexpr auto kOverrideArg = "override";

	constexpr auto kInfoCompression = "Compression";
	constexpr auto kInfoCapacity = "Capacity";
	constexpr auto kInfoMergedNodes = "Merged nodes";
	constexpr auto kInfoUnmergedNodes = "Unmerged nodes";
	constexpr auto kInfoMergedWeight = "Merged weight";
	constexpr auto kInfoUnmergedWeight = "Unmerged weight";
	constexpr auto kInfoObservations = "Observations";
	constexpr auto kInfoTotalCompressions = "Total compressions";
	constexpr auto kNan = "nan";

	constexpr const char *errParseLowCutQuantile = "error parsing low_cut_percentile";
	constexpr const char *errParseHighCutQuantile = "error parsing high_cut_percentile";
	constexpr const char *errCutQuantileRange = "low_cut_percentile and high_cut_percentile should be in [0,1]";
	constexpr const char *errLowCutQuantileLess = "low_cut_percentile should be lower than high_cut_percentile";
	} // namespace

	class CommandTDigestCreate : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	CommandParser parser(args, 1);
	key_name_ = GET_OR_RET(parser.TakeStr());
	options_.compression = 100;
	if (parser.EatEqICase(kCompressionArg)) {
	if (!parser.Good()) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}
	auto status = parser.TakeInt<int32_t>();
	if (!status) {
	return {Status::RedisParseErr, errParseCompression};
	}
	auto compression = *status;
	if (compression <= 0) {
	return {Status::RedisParseErr, errCompressionMustBePositive};
	}
	if (compression < 1 \|\| compression > static_cast<int32_t>(kTDigestMaxCompression)) {
	return {Status::RedisParseErr, errCompressionOutOfRange};
	}
	options_.compression = static_cast<uint32_t>(compression);
	}
	if (parser.Good()) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	return Status::OK();
	}

	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	bool exists = false;
	auto s = tdigest.Create(ctx, key_name_, options_, &exists);
	if (!s.ok()) {
	if (exists) {
	return {Status::RedisExecErr, errKeyAlreadyExists};
	}
	return {Status::RedisExecErr, s.ToString()};
	}
	*output = redis::RESP_OK;
	return Status::OK();
	}

	private:
	std::string key_name_;
	TDigestCreateOptions options_;
	};

	class CommandTDigestInfo : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];
	return Status::OK();
	}

	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	TDigestMetadata metadata;
	auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	output->append(conn->HeaderOfMap(8));
	output->append(redis::BulkString(kInfoCompression));
	output->append(redis::Integer(metadata.compression));
	output->append(redis::BulkString(kInfoCapacity));
	output->append(redis::Integer(metadata.capacity));
	output->append(redis::BulkString(kInfoMergedNodes));
	output->append(redis::Integer(metadata.merged_nodes));
	output->append(redis::BulkString(kInfoUnmergedNodes));
	output->append(redis::Integer(metadata.unmerged_nodes));
	output->append(redis::BulkString(kInfoMergedWeight));
	output->append(redis::Integer(metadata.merged_weight));
	output->append(redis::BulkString(kInfoUnmergedWeight));
	output->append(redis::Integer(metadata.total_weight - metadata.merged_weight));
	output->append(redis::BulkString(kInfoObservations));
	output->append(redis::Integer(metadata.total_observations));
	output->append(redis::BulkString(kInfoTotalCompressions));
	output->append(redis::Integer(metadata.merge_times));
	// "Memory usage" is not meaningful for kvrocks, so we don't provide it here.
	return Status::OK();
	}

	private:
	std::string key_name_;
	};

	class CommandTDigestAdd : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];

	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());

	auto s = tdigest.Add(ctx, key_name_, values_);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	*output = redis::RESP_OK;
	return Status::OK();
	}

	private:
	std::string key_name_;
	std::vector<double> values_;
	};

	template <bool Reverse>
	class TDigestRankCommand : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];

	std::set<std::string> unique_inputs_set(args.begin() + 2, args.end());
	origin_inputs_.assign(args.begin() + 2, args.end());

	unique_inputs_.reserve(unique_inputs_set.size());
	size_t i = 0;
	for (const auto &input : unique_inputs_set) {
	auto value = ParseFloat(input);
	if (!value) {
	return {Status::RedisParseErr, errValueIsNotFloat};
	}
	unique_inputs_.push_back(*value);
	unique_inputs_order_[input] = i;
	++i;
	}
	return Status::OK();
	}
	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	std::vector<int> result;
	result.reserve(origin_inputs_.size());

	if (const auto s =
	[&]() {
	if constexpr (Reverse) {
	return tdigest.RevRank(ctx, key_name_, unique_inputs_, &result);
	} else {
	return tdigest.Rank(ctx, key_name_, unique_inputs_, &result);
	}
	}();
	!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	std::vector<std::string> rev_ranks;
	rev_ranks.reserve(origin_inputs_.size());
	for (const auto &v : origin_inputs_) {
	rev_ranks.push_back(redis::Integer(result[unique_inputs_order_[v]]));
	}
	*output = redis::Array(rev_ranks);
	return Status::OK();
	}

	private:
	std::string key_name_;
	std::vector<double> unique_inputs_;
	std::map<std::string, size_t> unique_inputs_order_;
	std::vector<std::string> origin_inputs_;
	};

	class CommandTDigestRevRank : public TDigestRankCommand<true> {};

	class CommandTDigestRank : public TDigestRankCommand<false> {};

	template <bool Reverse>
	class TDigestByRankCommand : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];

	std::set<std::string> unique_inputs_set(args.begin() + 2, args.end());
	origin_inputs_.assign(args.begin() + 2, args.end());

	unique_inputs_.reserve(unique_inputs_set.size());
	size_t i = 0;
	for (const auto &input : unique_inputs_set) {
	auto value = ParseInt<int>(input);
	if (!value) {
	return {Status::RedisParseErr, errValueNotInteger};
	}
	if (*value < 0) {
	return {Status::InvalidArgument, errInvalidRankValue};
	}
	unique_inputs_.push_back(*value);
	unique_inputs_order_[input] = i;
	++i;
	}
	return Status::OK();
	}
	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	std::vector<double> result;
	result.reserve(origin_inputs_.size());

	if (const auto s =
	[&]() {
	if constexpr (Reverse) {
	return tdigest.ByRevRank(ctx, key_name_, unique_inputs_, &result);
	} else {
	return tdigest.ByRank(ctx, key_name_, unique_inputs_, &result);
	}
	}();
	!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	std::vector<std::string> ranks;
	ranks.reserve(origin_inputs_.size());
	auto is_resp3 = conn->GetProtocolVersion() == RESP::v3;
	for (const auto &v : origin_inputs_) {
	auto rank_value = result[unique_inputs_order_[v]];
	if (is_resp3) {
	ranks.push_back(redis::Double(redis::RESP::v3, rank_value));
	} else {
	ranks.push_back(redis::BulkString(fmt::format("{}", rank_value)));
	}
	}
	*output = redis::Array(ranks);
	return Status::OK();
	}

	private:
	std::string key_name_;
	std::vector<int> unique_inputs_;
	std::map<std::string, size_t> unique_inputs_order_;
	std::vector<std::string> origin_inputs_;
	};

	class CommandTDigestByRevRank : public TDigestByRankCommand<true> {};

	class CommandTDigestByRank : public TDigestByRankCommand<false> {};

	class CommandTDigestMinMax : public Commander {
	public:
	explicit CommandTDigestMinMax(bool is_min) : is_min_(is_min) {}

	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];
	return Status::OK();
	}

	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	TDigestMetadata metadata;
	auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	if (metadata.total_observations == 0) {
	*output = redis::BulkString("nan");
	return Status::OK();
	}

	double value = is_min_ ? metadata.minimum : metadata.maximum;
	*output = redis::BulkString(fmt::format("{}", value));
	return Status::OK();
	}

	private:
	std::string key_name_;
	bool is_min_;
	};
	class CommandTDigestReset : public Commander {
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];
	return Status::OK();
	}
	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	TDigestMetadata metadata;
	auto s = tdigest.GetMetaData(ctx, key_name_, &metadata);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}
	if (metadata.total_observations == 0) {
	*output = redis::RESP_OK;
	return Status::OK();
	}
	s = tdigest.Reset(ctx, key_name_);
	if (!s.ok()) {
	return {Status::RedisExecErr, s.ToString()};
	}
	*output = redis::RESP_OK;
	return Status::OK();
	}

	private:
	std::string key_name_;
	};
	// Then replace the existing template implementation and type aliases with:
	class CommandTDigestMin : public CommandTDigestMinMax {
	public:
	CommandTDigestMin() : CommandTDigestMinMax(true) {}
	};

	class CommandTDigestMax : public CommandTDigestMinMax {
	public:
	CommandTDigestMax() : CommandTDigestMinMax(false) {}
	};
	class CommandTDigestQuantile : public Commander {
	Status Parse(const std::vector<std::string> &args) override {
	key_name_ = args[1];
	quantiles_.reserve(args.size() - 2);
	for (size_t i = 2; i < args.size(); i++) {
	auto value = ParseFloat(args[i]);
	if (!value) {
	return {Status::RedisParseErr, errValueIsNotFloat};
	}
	quantiles_.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());
	TDigestQuantitleResult result;
	auto s = tdigest.Quantile(ctx, key_name_, quantiles_, &result);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}
	auto quantile_strings = result.quantiles
	? (ranges::views::transform(*result.quantiles, util::Float2String) \| ranges::to_vector)
	: std::vector<std::string>(quantiles_.size(), kNan);
	*output = conn->MultiBulkString(quantile_strings);
	return Status::OK();
	}

	private:
	std::string key_name_;
	std::vector<double> quantiles_;
	};

	class CommandTDigestMerge : public Commander {
	Status Parse(const std::vector<std::string> &args) override {
	CommandParser parser(args, 1);
	dest_key_ = GET_OR_RET(parser.TakeStr());
	auto numkeys = parser.TakeInt();
	if (!numkeys) {
	return {Status::RedisParseErr, errParsingNumkeys};
	}

	if (*numkeys <= 0) {
	return {Status::RedisParseErr, errNumkeysMustBePositive};
	}

	if (static_cast<int64_t>(args.size()) < (3 + *numkeys)) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	source_keys_.reserve(*numkeys);

	for (auto i = 3; i < (3 + *numkeys); i++) {
	auto src_digest = GET_OR_RET(parser.TakeStr());
	source_keys_.emplace_back(std::move(src_digest));
	}

	while (parser.Good()) {
	// more arguments than expected compression and override
	if (options_.compression > 0 && options_.override_flag) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	if (parser.EatEqICase(kCompressionArg)) {
	// compression already set or without a compression value
	if (options_.compression > 0 \|\| !parser.Good()) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	if (auto compression = parser.TakeInt<uint32_t>(); !compression) {
	return {Status::RedisParseErr, errParseCompression};
	} else if (compression <= 0 \|\| compression > kTDigestMaxCompression) {
	return {Status::RedisParseErr, errCompressionOutOfRange};
	} else {
	options_.compression = *compression;
	}
	} else if (parser.EatEqICase(kOverrideArg)) {
	if (options_.override_flag) { // override already set
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}
	options_.override_flag = true;
	} else {
	return {Status::RedisParseErr, errWrongKeyword};
	}
	}

	return Status::OK();
	}

	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	auto s = tdigest.Merge(ctx, dest_key_, source_keys_, options_);
	if (!s.ok()) {
	return {Status::RedisExecErr, s.IsNotFound() ? errKeyNotFound : s.ToString()};
	}
	*output = redis::RESP_OK;
	return Status::OK();
	}

	private:
	std::string dest_key_;
	std::vector<std::string> source_keys_;
	TDigestMergeOptions options_;
	};

	class CommandTDigestTrimmedMean : public Commander {
	public:
	Status Parse(const std::vector<std::string> &args) override {
	if (args.size() != 4) {
	return {Status::RedisParseErr, errWrongNumOfArguments};
	}

	key_name_ = args[1];

	auto low_cut_quantile = ParseFloat(args[2]);
	if (!low_cut_quantile \|\| std::isnan(*low_cut_quantile)) {
	return {Status::RedisParseErr, errParseLowCutQuantile};
	}
	low_cut_quantile_ = *low_cut_quantile;

	auto high_cut_quantile = ParseFloat(args[3]);
	if (!high_cut_quantile \|\| std::isnan(*high_cut_quantile)) {
	return {Status::RedisParseErr, errParseHighCutQuantile};
	}
	high_cut_quantile_ = *high_cut_quantile;

	if (!std::isfinite(low_cut_quantile_) \|\| low_cut_quantile_ < 0.0 \|\| low_cut_quantile_ > 1.0) {
	return {Status::RedisParseErr, errCutQuantileRange};
	}
	if (!std::isfinite(high_cut_quantile_) \|\| high_cut_quantile_ < 0.0 \|\| high_cut_quantile_ > 1.0) {
	return {Status::RedisParseErr, errCutQuantileRange};
	}
	if (low_cut_quantile_ >= high_cut_quantile_) {
	return {Status::RedisParseErr, errLowCutQuantileLess};
	}

	return Status::OK();
	}

	Status Execute(engine::Context &ctx, Server srv, Connection conn, std::string *output) override {
	TDigest tdigest(srv->storage, conn->GetNamespace());
	TDigestTrimmedMeanResult result;

	auto s = tdigest.TrimmedMean(ctx, key_name_, low_cut_quantile_, high_cut_quantile_, &result);
	if (!s.ok()) {
	if (s.IsNotFound()) {
	return {Status::RedisExecErr, errKeyNotFound};
	}
	return {Status::RedisExecErr, s.ToString()};
	}

	if (!result.mean.has_value()) {
	*output = redis::BulkString(kNan);
	} else {
	output = redis::BulkString(util::Float2String(result.mean));
	}

	return Status::OK();
	}

	private:
	std::string key_name_;
	double low_cut_quantile_;
	double high_cut_quantile_;
	};

	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}};
	}

	REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.create", -2, "write", 1, 1, 1),
	MakeCmdAttr<CommandTDigestInfo>("tdigest.info", 2, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestAdd>("tdigest.add", -3, "write", 1, 1, 1),
	MakeCmdAttr<CommandTDigestMax>("tdigest.max", 2, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestRevRank>("tdigest.revrank", -3, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestRank>("tdigest.rank", -3, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestByRevRank>("tdigest.byrevrank", -3, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestByRank>("tdigest.byrank", -3, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestTrimmedMean>("tdigest.trimmed_mean", 4, "read-only", 1, 1, 1),
	MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
	MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
	} // namespace redis