This is an automated email from the ASF dual-hosted git repository.
LindaSummer pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new d3965fcc5 feat(tdigest): implement `tdigest.cdf` command (#3551)
d3965fcc5 is described below
commit d3965fcc5066229be4e0b7725623b741a1f18e64
Author: Edward Xu <[email protected]>
AuthorDate: Tue Aug 18 10:43:58 2026 +0800
feat(tdigest): implement `tdigest.cdf` command (#3551)
# Issue
Close #2807
# Proposed Changes
- Implement `tdigest.cdf` command
- Add unit tests and integration tests
---------
Co-authored-by: Anirudh Lakhanpal <[email protected]>
---
src/commands/cmd_tdigest.cc | 44 +++-
src/types/redis_tdigest.cc | 37 +++
src/types/redis_tdigest.h | 7 +
src/types/tdigest.h | 191 ++++++++++++++-
tests/cppunit/types/tdigest_test.cc | 306 ++++++++++++++++++++++++-
tests/gocase/unit/type/tdigest/tdigest_test.go | 164 +++++++++++++
6 files changed, 745 insertions(+), 4 deletions(-)
diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc
index c0baf8ca1..149239859 100644
--- a/src/commands/cmd_tdigest.cc
+++ b/src/commands/cmd_tdigest.cc
@@ -556,6 +556,47 @@ class CommandTDigestTrimmedMean : public Commander {
double high_cut_quantile_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() == 2) return {Status::RedisParseErr,
errWrongNumOfArguments};
+ key_name_ = args[1];
+ inputs_.reserve(args.size() - 2);
+ for (size_t i = 2; i < args.size(); i++) {
+ auto value = ParseFloat(args[i]);
+ if (!value) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ if (std::isnan(*value)) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ inputs_.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());
+ TDigestCDFResult result;
+ auto s = tdigest.CDF(ctx, key_name_, inputs_, &result);
+ if (!s.ok()) {
+ if (s.IsNotFound()) {
+ return {Status::RedisExecErr, errKeyNotFound};
+ }
+ return {Status::RedisExecErr, s.ToString()};
+ }
+
+ output->append(redis::MultiLen(result.cdf_values.size()));
+ for (auto const value : result.cdf_values) {
+ output->append(conn->Double(value));
+ }
+ return Status::OK();
+ }
+
+ private:
+ std::string key_name_;
+ std::vector<double> inputs_;
+};
+
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}};
@@ -573,5 +614,6 @@ REDIS_REGISTER_COMMANDS(TDigest,
MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
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));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4,
"write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -3,
"read-only", 1, 1, 1));
} // namespace redis
diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc
index 4d0325f3d..38c836dad 100644
--- a/src/types/redis_tdigest.cc
+++ b/src/types/redis_tdigest.cc
@@ -28,12 +28,14 @@
#include <rocksdb/status.h>
#include <algorithm>
+#include <cstdint>
#include <iterator>
#include <limits>
#include <memory>
#include <range/v3/algorithm/minmax.hpp>
#include <range/v3/range/conversion.hpp>
#include <range/v3/view/join.hpp>
+#include <range/v3/view/map.hpp>
#include <range/v3/view/transform.hpp>
#include <vector>
@@ -570,6 +572,41 @@ 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.total_observations == 0) {
+ result->cdf_values = std::vector<double>(inputs.size(),
std::numeric_limits<double>::quiet_NaN());
+ return rocksdb::Status::OK();
+ }
+
+ if (auto status = mergeNodes(ctx, ns_key, &metadata); !status.ok()) {
+ return status;
+ }
+ }
+
+ std::vector<Centroid> centroids;
+ if (auto status = dumpCentroids(ctx, ns_key, metadata, ¢roids);
!status.ok()) {
+ return status;
+ }
+
+ auto dump_centroids = DummyCentroids<false>(metadata, centroids);
+ if (auto status = TDigestCDF(centroids, dump_centroids.Min(),
dump_centroids.Max(), dump_centroids.TotalWeight(),
+ inputs, &result->cdf_values);
+ !status.IsOK()) {
+ return rocksdb::Status::InvalidArgument(status.Msg());
+ }
+ return rocksdb::Status::OK();
+}
+
rocksdb::Status TDigest::GetMetaData(engine::Context& context, const Slice&
digest_name, TDigestMetadata* metadata) {
auto ns_key = AppendNamespacePrefix(digest_name);
return Database::GetMetadata(context, {kRedisTDigest}, ns_key, metadata);
diff --git a/src/types/redis_tdigest.h b/src/types/redis_tdigest.h
index 8cf4562d2..445445c0f 100644
--- a/src/types/redis_tdigest.h
+++ b/src/types/redis_tdigest.h
@@ -49,6 +49,10 @@ struct TDigestMergeOptions {
bool override_flag = false;
};
+struct TDigestCDFResult {
+ std::vector<double> cdf_values;
+};
+
struct TDigestQuantitleResult {
std::optional<std::vector<double>> quantiles;
};
@@ -93,6 +97,9 @@ class TDigest : public SubKeyScanner {
double high_cut_quantile,
TDigestTrimmedMeanResult* result);
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 };
diff --git a/src/types/tdigest.h b/src/types/tdigest.h
index 531f693ed..66b2991bb 100644
--- a/src/types/tdigest.h
+++ b/src/types/tdigest.h
@@ -22,10 +22,11 @@
#include <fmt/format.h>
+#include <algorithm>
+#include <cmath>
+#include <iterator>
#include <limits>
#include <map>
-#include <numeric>
-#include <variant>
#include <vector>
#include "common/status.h"
@@ -172,6 +173,192 @@ struct DoubleComparator {
bool operator()(const double& a, const double& b) const { return
DoubleCompare(a, b) == -1; }
};
+// Match RedisBloom t-digest-c CDF behavior: if min/max is outside the
first/last centroid mean, the exact
+// boundary sample is treated as a singleton with weight 1. Its center rank is
0.5 at min and
+// total_weight - 0.5 at max; interpolation toward an inner centroid starts
after the singleton, at rank 1 or
+// total_weight - 1.
+// refer to implementation:
+//
https://github.com/RedisBloom/t-digest-c/blob/50edef336eb27ed5b19e7f9be05494683ca58515/src/tdigest.c#L223
+inline Status TDigestCDF(const std::vector<Centroid>& centroids, double
centroids_min, double centroids_max,
+ double total_weight, const std::vector<double>&
inputs, std::vector<double>* result) {
+ if (centroids.empty() || total_weight <= 0) {
+ return Status{Status::InvalidArgument, "invalid or empty tdigest"};
+ }
+
+ std::map<double, std::vector<size_t>> sorted_unique_input_idx_map;
+ for (size_t i = 0; i < inputs.size(); ++i) {
+ sorted_unique_input_idx_map[inputs[i]].push_back(i);
+ }
+
+ std::vector<double> sorted_unique_inputs;
+ sorted_unique_inputs.reserve(sorted_unique_input_idx_map.size());
+ std::transform(sorted_unique_input_idx_map.cbegin(),
sorted_unique_input_idx_map.cend(),
+ std::back_inserter(sorted_unique_inputs), [](const auto&
pair) { return pair.first; });
+
+ constexpr double kSingletonBoundaryWeight = 1.0;
+ constexpr double kHalfSingletonBoundaryWeight = kSingletonBoundaryWeight / 2;
+
+ std::vector<double> sorted_result_weights;
+ sorted_result_weights.reserve(sorted_unique_inputs.size());
+ if (centroids.size() == 1) {
+ // only one centroid, min should equal max, and all inputs should be
either less than, equal to,
+ // or greater than the centroid mean
+ const double width = centroids_max - centroids_min;
+ for (const auto input : sorted_unique_inputs) {
+ if (input < centroids_min) {
+ sorted_result_weights.push_back(0.0);
+ continue;
+ }
+
+ if (input > centroids_max) {
+ sorted_result_weights.push_back(total_weight);
+ continue;
+ }
+
+ if (input - centroids_min <= width) {
+ // min and max are too close to do any viable interpolation, treat the
centroid as a singleton
+ sorted_result_weights.push_back(total_weight / 2);
+ } else {
+ // interpolate if somehow we have weight > 0 and max != min, which
should not happen in a valid tdigest
+ sorted_result_weights.push_back((input - centroids_min) / width *
total_weight);
+ }
+ }
+ } else {
+ auto first_valid_input_it = std::find_if(sorted_unique_inputs.cbegin(),
sorted_unique_inputs.cend(),
+ [centroids_min](double input) {
return input >= centroids_min; });
+ auto last_valid_input_it = std::find_if(sorted_unique_inputs.crbegin(),
sorted_unique_inputs.crend(),
+ [centroids_max](double input) {
return input <= centroids_max; });
+ auto input_idx = (first_valid_input_it == sorted_unique_inputs.cend())
+ ? sorted_unique_inputs.size()
+ : std::distance(sorted_unique_inputs.cbegin(),
first_valid_input_it);
+ auto last_valid_input_idx = (last_valid_input_it ==
sorted_unique_inputs.crend())
+ ? 0
+ :
std::distance(sorted_unique_inputs.cbegin(), last_valid_input_it.base());
+
+ // fill in 0 for inputs less than the min boundary
+ for (auto i = 0; i < input_idx; ++i) {
+ sorted_result_weights.push_back(0.);
+ }
+
+ size_t centroid_idx = 0;
+
+ // greater than the min boundary, but less than the first centroid mean
+ while (centroid_idx == 0 && input_idx < last_valid_input_idx &&
+ sorted_unique_inputs[input_idx] < centroids[centroid_idx].mean) {
+ auto cdf_input = sorted_unique_inputs[input_idx];
+ auto current_centroid = centroids[centroid_idx];
+ const auto width = current_centroid.mean - centroids_min;
+ double interpolated_weight = std::numeric_limits<double>::quiet_NaN();
+ if (width > 0) {
+ if (cdf_input == centroids_min) {
+ interpolated_weight = kHalfSingletonBoundaryWeight;
+ } else {
+ // there must be a singleton at the min boundary, so the
interpolation starts after it, at rank 1
+ interpolated_weight =
+ Lerp(kHalfSingletonBoundaryWeight, current_centroid.weight / 2,
(cdf_input - centroids_min) / width);
+ }
+ } else {
+ // this should be redundant of the check cdf_input < centroids_min,
but for clarity
+ interpolated_weight = 0.;
+ }
+ sorted_result_weights.push_back(interpolated_weight);
+ ++input_idx;
+ }
+
+ double weight_so_far = 0.;
+ while (centroid_idx < centroids.size() - 1 && input_idx <
last_valid_input_idx) {
+ auto cdf_input = sorted_unique_inputs[input_idx];
+ auto current_centroid = centroids[centroid_idx];
+ auto next_centroid = centroids[centroid_idx + 1];
+
+ if (cdf_input == current_centroid.mean) {
+ double dw = 0.;
+ auto same_mean_idx = centroid_idx;
+ while (same_mean_idx < centroids.size() &&
centroids[same_mean_idx].mean == current_centroid.mean) {
+ dw += centroids[same_mean_idx].weight;
+ ++same_mean_idx;
+ }
+ sorted_result_weights.push_back(weight_so_far + dw / 2);
+ ++input_idx;
+ continue;
+ }
+
+ if (current_centroid.mean < cdf_input && cdf_input < next_centroid.mean)
{
+ if (next_centroid.mean - current_centroid.mean > 0) {
+ double left_exclude_weight = 0;
+ double right_exclude_weight = 0;
+ if (current_centroid.weight == kSingletonBoundaryWeight) {
+ if (next_centroid.weight == kSingletonBoundaryWeight) {
+ // both adjacent centroids are singletons, include the left
exact sample and exclude the right one.
+ sorted_result_weights.push_back(weight_so_far +
kSingletonBoundaryWeight);
+ // weight_so_far += current_centroid.weight;
+ ++input_idx;
+ continue;
+ } else {
+ left_exclude_weight = kHalfSingletonBoundaryWeight;
+ }
+ } else if (next_centroid.weight == kSingletonBoundaryWeight) {
+ right_exclude_weight = kHalfSingletonBoundaryWeight;
+ }
+
+ double dw = (current_centroid.weight + next_centroid.weight) / 2;
+ double dw_no_singleton = dw - left_exclude_weight -
right_exclude_weight;
+ double base_weight = weight_so_far + current_centroid.weight / 2 +
left_exclude_weight;
+ auto interpolated_weight =
+ Lerp(base_weight, base_weight + dw_no_singleton,
+ (cdf_input - current_centroid.mean) / (next_centroid.mean -
current_centroid.mean));
+ sorted_result_weights.push_back(interpolated_weight);
+ ++input_idx;
+ }
+ continue;
+ }
+
+ ++centroid_idx;
+ weight_so_far += current_centroid.weight;
+ }
+
+ while (centroid_idx == centroids.size() - 1 && input_idx <
last_valid_input_idx &&
+ sorted_unique_inputs[input_idx] < centroids[centroid_idx].mean) {
+ auto cdf_input = sorted_unique_inputs[input_idx];
+ auto current_centroid = centroids[centroid_idx];
+ const auto width = current_centroid.mean - centroids_min;
+ double interpolated_weight = std::numeric_limits<double>::quiet_NaN();
+ if (width > 0) {
+ if (cdf_input == centroids_min) {
+ interpolated_weight = kHalfSingletonBoundaryWeight;
+ } else {
+ // there must be a singleton at the min boundary, so the
interpolation starts after it, at rank
+ // kHalfSingletonBoundaryWeight
+ interpolated_weight =
+ Lerp(kHalfSingletonBoundaryWeight, current_centroid.weight / 2,
(cdf_input - centroids_min) / width);
+ }
+ } else {
+ // this should be redundant of the check cdf_input < centroids_min,
but for clarity
+ interpolated_weight = 0;
+ }
+ sorted_result_weights.push_back(interpolated_weight);
+ ++input_idx;
+ }
+
+ // fill in 1 for inputs greater than the max boundary
+ while (input_idx < sorted_unique_inputs.size()) {
+ // handle remaining inputs
+ sorted_result_weights.push_back(total_weight);
+ ++input_idx;
+ }
+ }
+
+ result->clear();
+ result->resize(inputs.size(), std::numeric_limits<double>::quiet_NaN());
+ for (size_t i = 0; i < sorted_unique_inputs.size(); ++i) {
+ for (auto idx : sorted_unique_input_idx_map[sorted_unique_inputs[i]]) {
+ (*result)[idx] = std::clamp(sorted_result_weights[i] / total_weight,
0.0, 1.0);
+ }
+ }
+
+ return Status::OK();
+}
+
template <bool Reverse, typename TD>
inline Status TDigestByRank(TD&& td, const std::vector<int>& inputs,
std::vector<double>* result) {
result->clear();
diff --git a/tests/cppunit/types/tdigest_test.cc
b/tests/cppunit/types/tdigest_test.cc
index f82519d2f..410fd3404 100644
--- a/tests/cppunit/types/tdigest_test.cc
+++ b/tests/cppunit/types/tdigest_test.cc
@@ -30,13 +30,14 @@
#include <range/v3/algorithm/shuffle.hpp>
#include <range/v3/range.hpp>
#include <range/v3/view/chunk.hpp>
+#include <range/v3/view/concat.hpp>
#include <range/v3/view/iota.hpp>
#include <range/v3/view/join.hpp>
+#include <range/v3/view/repeat.hpp>
#include <range/v3/view/transform.hpp>
#include <string>
#include <vector>
-#include "logging.h"
#include "storage/redis_metadata.h"
#include "test_base.h"
#include "time_util.h"
@@ -948,3 +949,306 @@ TEST_F(RedisTDigestTest,
MergeWithUserSpecifiedCompression) {
// Verify total observations: dest(1) + src(1) = 2
EXPECT_EQ(metadata.total_observations, 2);
}
+
+TEST_F(RedisTDigestTest, CDFTest) {
+ std::string cdf_tdigest_name = "test_cdf_digest" +
std::to_string(util::GetTimeStampMS());
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, cdf_tdigest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples = {1, 2, 2, 3, 3, 3, 4, 4, 4, 4, 5, 5, 5, 5, 5};
+ status = tdigest_->Add(*ctx_, cdf_tdigest_name, samples);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {0, 1, 2, 3, 4, 5, 6};
+ redis::TDigestCDFResult result;
+
+ status = tdigest_->CDF(*ctx_, cdf_tdigest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ std::vector<double> expected = {0.00, 0.03, 0.13, 0.29, 0.53, 0.83, 1.00};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR(result.cdf_values[i], expected[i], 0.015) <<
fmt::format("Mismatch at index {}", i);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFReturnsNaNOnEmptyTDigest) {
+ std::string test_digest_name = "test_digest_cdf_nan" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> values = {0.0, 1.0, 2.0, 3.0};
+ redis::TDigestCDFResult result;
+
+ status = tdigest_->CDF(*ctx_, test_digest_name, values, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+ ASSERT_EQ(result.cdf_values.size(), values.size());
+ for (const auto cdf : result.cdf_values) {
+ EXPECT_TRUE(std::isnan(cdf));
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFDuplicateValues) {
+ std::string test_digest_name = "test_cdf_duplicates" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {10, 10, 10, 20, 20});
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {5, 10, 20, 25};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ std::vector<double> expected = {0, 0.3, 0.8, 1};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR(result.cdf_values[i], expected[i], 0.001) <<
fmt::format("Mismatch at index {}", i);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFSignedZeroQueries) {
+ std::string test_digest_name = "test_cdf_signed_zero" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {-1, 0, 1});
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {-0.0, 0.0};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+ EXPECT_NEAR(result.cdf_values[0], 0.5, 0.001);
+ EXPECT_NEAR(result.cdf_values[1], 0.5, 0.001);
+}
+
+TEST_F(RedisTDigestTest, CDFUniformDistribution) {
+ std::string test_digest_name = "test_cdf_uniform" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {200}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples = ranges::views::iota(1, 101) |
+ ranges::views::transform([](int i) { return
(double)i; }) |
+ ranges::to<std::vector<double>>();
+ status = tdigest_->Add(*ctx_, test_digest_name, samples);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {1, 25, 50, 75, 100};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ std::vector<double> expected = {0.01, 0.25, 0.50, 0.75, 1.00};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR(result.cdf_values[i], expected[i], 0.02) <<
fmt::format("Mismatch at index {}, val={}", i, cdf_vals[i]);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFInterpolatesCompressedCentroids) {
+ std::string test_digest_name = "test_cdf_compressed" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {.compression = 10},
&exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples = ranges::views::iota(0, 100) |
+ ranges::views::transform([](int i) { return
static_cast<double>(i); }) |
+ ranges::to<std::vector<double>>();
+ status = tdigest_->Add(*ctx_, test_digest_name, samples);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {20, 40, 50, 60, 80};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ std::vector<double> expected = {0.205, 0.405, 0.505, 0.605, 0.805};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR(result.cdf_values[i], expected[i], 0.001)
+ << fmt::format("Mismatch at index {}, val={}", i, cdf_vals[i]);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFDoesNotInterpolateBetweenSingletonCentroids) {
+ std::string test_digest_name = "test_cdf_singletons" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {.compression =
100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {0, 10, 20});
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {11};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+ EXPECT_NEAR(result.cdf_values[0], 2.0 / 3, 0.001);
+}
+
+TEST_F(RedisTDigestTest, CDFMultipleAdds) {
+ std::string test_digest_name = "test_cdf_multiadd" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples1 = {1, 2, 3, 4, 5};
+ std::vector<double> samples2 = {6, 7, 8, 9, 10};
+ status = tdigest_->Add(*ctx_, test_digest_name, samples1);
+ ASSERT_TRUE(status.ok());
+ status = tdigest_->Add(*ctx_, test_digest_name, samples2);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {1, 5, 7, 10};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> expected = {0.10, 0.50, 0.70, 1.00};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR((result.cdf_values)[i], expected[i], 0.06)
+ << fmt::format("Mismatch at index {}, val={}", i, cdf_vals[i]);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFSkewedDistribution) {
+ std::string test_digest_name = "test_cdf_skewed" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {200}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples =
+ ranges::views::concat(
+ ranges::views::repeat(0.0) | ranges::views::take(100),
+ ranges::views::iota(1, 11) | ranges::views::transform([](int i) {
return static_cast<double>(i); })) |
+ ranges::to<std::vector<double>>();
+
+ status = tdigest_->Add(*ctx_, test_digest_name, samples);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {0, 1, 5, 10};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> expected = {0.4545, 0.91, 0.95, 1.00};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR((result.cdf_values)[i], expected[i], 0.03)
+ << fmt::format("Mismatch at index {}, val={}", i, cdf_vals[i]);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFRepeatedCentroids) {
+ std::string test_digest_name = "test_cdf_repeated_centroids" +
std::to_string(util::GetTimeStampMS());
+
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {200}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> samples = {-40, -36, -27, -13, -12, 7, 7, 25, 47, 50};
+
+ status = tdigest_->Add(*ctx_, test_digest_name, samples);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {0, 6.9, 7, 7.1, 10};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> expected = {0.5, 0.5, 0.6, 0.7, 0.7};
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+
+ for (size_t i = 0; i < cdf_vals.size(); i++) {
+ EXPECT_NEAR((result.cdf_values)[i], expected[i], 0.03)
+ << fmt::format("Mismatch at index {}, val={}", i, cdf_vals[i]);
+ }
+}
+
+TEST_F(RedisTDigestTest, CDFSingleWeightedCentroid) {
+ std::string test_digest_name = "test_cdf_single_weighted_centroid" +
std::to_string(util::GetTimeStampMS());
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {5, 5, 5, 5, 5, 5, 5, 5, 5,
5});
+ ASSERT_TRUE(status.ok());
+
+ std::vector<double> cdf_vals = {4, 5, 6};
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+ ASSERT_EQ(result.cdf_values.size(), cdf_vals.size());
+ EXPECT_NEAR(result.cdf_values[0], 0.0, 0.001);
+ EXPECT_NEAR(result.cdf_values[1], 0.5, 0.001);
+ EXPECT_NEAR(result.cdf_values[2], 1.0, 0.001);
+}
+
+TEST_F(RedisTDigestTest, CDFAllInputsBelowMinimum) {
+ std::string test_digest_name = "test_cdf_all_below_minimum" +
std::to_string(util::GetTimeStampMS());
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {1, 2});
+ ASSERT_TRUE(status.ok());
+
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, {-2, -1, 0}, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ EXPECT_EQ(result.cdf_values, std::vector<double>({0, 0, 0}));
+}
+
+TEST_F(RedisTDigestTest, CDFAllInputsAboveMaximum) {
+ std::string test_digest_name = "test_cdf_all_above_maximum" +
std::to_string(util::GetTimeStampMS());
+ bool exists = false;
+ auto status = tdigest_->Create(*ctx_, test_digest_name, {100}, &exists);
+ ASSERT_FALSE(exists);
+ ASSERT_TRUE(status.ok());
+
+ status = tdigest_->Add(*ctx_, test_digest_name, {1, 2});
+ ASSERT_TRUE(status.ok());
+
+ redis::TDigestCDFResult result;
+ status = tdigest_->CDF(*ctx_, test_digest_name, {3, 4, 5}, &result);
+ ASSERT_TRUE(status.ok()) << status.ToString();
+
+ EXPECT_EQ(result.cdf_values, std::vector<double>({1, 1, 1}));
+}
diff --git a/tests/gocase/unit/type/tdigest/tdigest_test.go
b/tests/gocase/unit/type/tdigest/tdigest_test.go
index e4daa3e18..1cfa6bf0a 100644
--- a/tests/gocase/unit/type/tdigest/tdigest_test.go
+++ b/tests/gocase/unit/type/tdigest/tdigest_test.go
@@ -45,6 +45,7 @@ const (
errMsgLowCutQuantileRange = "low_cut_percentile and
high_cut_percentile should be in [0,1]"
errMsgHighCutQuantileRange = "low_cut_percentile and
high_cut_percentile should be in [0,1]"
errMsgLowCutQuantileLess = "low_cut_percentile should be
lower than high_cut_percentile"
+ errValueIsNotFloat = "value is not a valid float"
)
type tdigestInfo struct {
@@ -1309,4 +1310,167 @@ func tdigestTestsByRankAndByRevRank(t *testing.T,
configs util.KvrocksServerConf
}
}
})
+
+ t.Run("tdigest.cdf with different arguments", func(t *testing.T) {
+ keyPrefix := "tdigest_cdf_"
+ isRESP3 := configs["resp3-enabled"] == "yes"
+
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CDF").Err(),
errMsgWrongNumberArg)
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CDF",
keyPrefix+"key1").Err(), errMsgWrongNumberArg)
+
+ // non-existent key
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CDF",
keyPrefix+"nonexistent", "1.0").Err(), errMsgKeyNotExist)
+
+ // invalid float value
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CDF",
keyPrefix+"key2", "invalid").Err(), errValueIsNotFloat)
+
+ // create a tdigest and add some data
+ tdigestKey := keyPrefix + "source"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE",
tdigestKey).Err())
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", tdigestKey,
"1.0", "2.0", "3.0", "4.0", "5.0").Err())
+
+ // single-value CDF query
+ rsp := rdb.Do(ctx, "TDIGEST.CDF", tdigestKey, "3.0")
+ require.NoError(t, rsp.Err())
+ vals, err := rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, 1)
+ require.NotEqual(t, "nan", vals[0])
+
+ // multi-value CDF query
+ rsp = rdb.Do(ctx, "TDIGEST.CDF", tdigestKey, "0.0", "2.5",
"5.0", "10.0")
+ require.NoError(t, rsp.Err())
+ vals, err = rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, 4)
+
+ // empty tdigest should return NaN
+ emptyKey := keyPrefix + "empty"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE",
emptyKey).Err())
+ rsp = rdb.Do(ctx, "TDIGEST.CDF", emptyKey, "1.0")
+ require.NoError(t, rsp.Err())
+ vals, err = rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, 1)
+ if isRESP3 {
+ cdf, ok := vals[0].(float64)
+ require.True(t, ok, "expected float64 but got %T",
vals[0])
+ require.True(t, math.IsNaN(cdf), "expected NaN but got
%v", cdf)
+ } else {
+ require.Equal(t, "nan", vals[0])
+ }
+
+ // Test with an empty digest and multi-valued CDF.
+ rsp = rdb.Do(ctx, "TDIGEST.CDF", emptyKey, "0.5", "1.0", "1.5",
"2.2")
+ require.NoError(t, rsp.Err())
+ vals, err = rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, 4)
+ if isRESP3 {
+ for i, v := range vals {
+ cdf, ok := v.(float64)
+ require.True(t, ok, "expected float64 but got
%T at index %d", v, i)
+ require.True(t, math.IsNaN(cdf), "expected NaN
but got %v at index %d", cdf, i)
+ }
+ } else {
+ require.Equal(t, []interface{}{"nan", "nan", "nan",
"nan"}, vals)
+ }
+
+ {
+ // test with samples, these data are generated from
redis tdigest.cdf command
+ var samplesTestKey = keyPrefix + "samples"
+ samples := []string{"42.210704046603865",
+ "35.2057544168016",
+ "12.444722535953744",
+ "-75.17683888127605",
+ "-27.367122500244108",
+ "77.34406489041416",
+ "-48.75891720991832",
+ "-42.93819033339253",
+ "44.855673646883474",
+ "50.891754534273815",
+ "-7.886891965257249",
+ "-89.26236837061113",
+ "48.50939634261886",
+ "59.88989586866117",
+ "32.46873341607849",
+ "-11.729446399857835",
+ "-100.32834587901394",
+ "-10.854053691728382",
+ "31.32150940192227",
+ "-19.568850657626257",
+ "-99.55268951300809",
+ "46.23219381980718",
+ "4.557500453188453",
+ "-52.306747310528394",
+ "50.02350257515229",
+ "27.897077091194205",
+ "-36.101222754212685",
+ "-34.17897261387189",
+ "-91.53560443207508",
+ "46.348114674645984",
+ }
+
+ cdfArgs := []string{
+ "60.01887553433912",
+ "-37.776384417214345",
+ "-82.915002521946",
+ "-87.44554390843862",
+ "31.95001239955863",
+ "-82.89948255423627",
+ "-73.80710164584889",
+ "-78.59695062757828",
+ "-30.343669845908025",
+ "-49.798720052628795",
+ }
+
+ expectedCdfs := []float64{
+ 0.9666666666666667,
+ 0.26666666666666666,
+ 0.13333333333333333,
+ 0.13333333333333333,
+ 0.6333333333333333,
+ 0.13333333333333333,
+ 0.16666666666666666,
+ 0.13333333333333333,
+ 0.3333333333333333,
+ 0.2,
+ }
+
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE",
samplesTestKey, "compression", "100").Err())
+
+ var addReqArgs = make([]interface{}, 0, len(samples)+2)
+ addReqArgs = append(addReqArgs, "TDIGEST.ADD",
samplesTestKey)
+ for _, sample := range samples {
+ addReqArgs = append(addReqArgs, sample)
+ }
+ require.NoError(t, rdb.Do(ctx, addReqArgs...).Err())
+
+ var cdfReqArgs = make([]interface{}, 0, len(cdfArgs)+2)
+ cdfReqArgs = append(cdfReqArgs, "TDIGEST.CDF",
samplesTestKey)
+ for _, arg := range cdfArgs {
+ cdfReqArgs = append(cdfReqArgs, arg)
+ }
+
+ rsp = rdb.Do(ctx, cdfReqArgs...)
+ require.NoError(t, rsp.Err())
+ vals, err = rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, len(cdfArgs))
+ for i, v := range vals {
+ var cdf float64
+ if isRESP3 {
+ var ok bool
+ cdf, ok = v.(float64)
+ require.True(t, ok, "expected float64
but got %T at index %d", v, i)
+ } else {
+ str, ok := v.(string)
+ require.True(t, ok, "expected string
but got %T at index %d", v, i)
+ cdf, err = strconv.ParseFloat(str, 64)
+ require.NoError(t, err)
+ }
+ require.InDelta(t, expectedCdfs[i], cdf, 0.01,
"CDF mismatch at index %d", i)
+ }
+ }
+ })
}
