From 8f3d5b39ebad530cf6bf50e0d3c81c55433bdc45 Mon Sep 17 00:00:00 2001 From: edward_xu Date: Mon, 13 Jul 2026 22:40:23 +0800 Subject: [PATCH 1/6] implement `tdigest.cdf` command Co-authored-by: Anirudh Lakhanpal --- src/commands/cmd_tdigest.cc | 65 +++++++- src/types/redis_tdigest.cc | 89 ++++++++++ src/types/redis_tdigest.h | 6 + tests/cppunit/types/tdigest_test.cc | 157 +++++++++++++++++- .../gocase/unit/type/tdigest/tdigest_test.go | 142 ++++++++++++++++ 5 files changed, 457 insertions(+), 2 deletions(-) diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc index c0baf8ca1a2..a101dc810f2 100644 --- a/src/commands/cmd_tdigest.cc +++ b/src/commands/cmd_tdigest.cc @@ -18,7 +18,10 @@ * */ +#include #include +#include +#include #include #include "command_parser.h" @@ -556,6 +559,65 @@ class CommandTDigestTrimmedMean : public Commander { double high_cut_quantile_; }; +class CommandTDigestCDF : public Commander { + Status Parse(const std::vector &args) override { + if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments}; + key_name_ = args[1]; + std::map> unique_inputs; + 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}; + } + if (unique_inputs.find(*value) == unique_inputs.cend()) { + unique_inputs[*value] = std::vector{i - 2}; + } else { + unique_inputs[*value].push_back(i - 2); + } + } + sorted_unique_inputs_with_idx_ = + ranges::views::transform(unique_inputs, + [](const auto &pair) { return std::make_pair(pair.first, pair.second); }) | + ranges::to_vector; + num_inputs_ = args.size() - 2; + return Status::OK(); + } + + Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override { + TDigest tdigest(srv->storage, conn->GetNamespace()); + TDigestCDFResult result; + std::vector uniq_cdfs = sorted_unique_inputs_with_idx_ | ranges::views::keys | ranges::to_vector; + auto s = tdigest.CDFUniqSorted(ctx, key_name_, uniq_cdfs, &result); + if (!s.ok()) { + if (s.IsNotFound()) { + return {Status::RedisExecErr, errKeyNotFound}; + } + return {Status::RedisExecErr, s.ToString()}; + } + + std::vector> cdf_uniq_results_with_idx = + ranges::views::transform(result.cdf_values, util::Float2String) | ranges::views::enumerate | ranges::to_vector; + std::vector cdf_result(num_inputs_, kNan); + + for (const auto &[idx, result_str] : cdf_uniq_results_with_idx) { + for (const auto &origin_idx : sorted_unique_inputs_with_idx_[idx].second) { + cdf_result[origin_idx] = result_str; + } + } + + *output = conn->MultiBulkString(cdf_result); + return Status::OK(); + } + + private: + std::string key_name_; + size_t num_inputs_; + std::vector>> sorted_unique_inputs_with_idx_; +}; + std::vector GetMergeKeyRange(const std::vector &args) { auto numkeys = ParseInt(args[2], 10).ValueOr(0); return {{1, 1, 1}, {3, 2 + numkeys, 1}}; @@ -573,5 +635,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr("tdigest.crea MakeCmdAttr("tdigest.quantile", -3, "read-only", 1, 1, 1), MakeCmdAttr("tdigest.trimmed_mean", 4, "read-only", 1, 1, 1), MakeCmdAttr("tdigest.reset", 2, "write", 1, 1, 1), - MakeCmdAttr("tdigest.merge", -4, "write", GetMergeKeyRange)); + MakeCmdAttr("tdigest.merge", -4, "write", GetMergeKeyRange), + MakeCmdAttr("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 4d0325f3da3..4af131e21e9 100644 --- a/src/types/redis_tdigest.cc +++ b/src/types/redis_tdigest.cc @@ -35,6 +35,7 @@ #include #include #include +#include #include #include "commands/error_constants.h" @@ -570,6 +571,94 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest, return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); } +rocksdb::Status TDigest::CDFUniqSorted(engine::Context& ctx, const Slice& digest_name, + const std::vector& inputs, TDigestCDFResult* result) { + if (!std::is_sorted(inputs.cbegin(), inputs.cend())) { + return rocksdb::Status::InvalidArgument( + "Internal error: inputs must be sorted in ascending order for CDF computation."); + } + + if (std::set(inputs.cbegin(), inputs.cend()).size() != inputs.size()) { + return rocksdb::Status::InvalidArgument("Internal error: inputs must be unique for CDF computation."); + } + + 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(inputs.size(), std::numeric_limits::quiet_NaN()); + return rocksdb::Status::OK(); + } + + 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 centroids; + if (auto status = dumpCentroids(ctx, ns_key, metadata, ¢roids); !status.ok()) { + return status; + } + + auto dump_centroids = DummyCentroids(metadata, centroids); + auto total_weight = dump_centroids.TotalWeight(); + auto iter = dump_centroids.Begin(); + double accum_weight = 0.; + std::vector results; + results.reserve(inputs.size()); + for (const auto val : inputs) { + double weight = accum_weight; + for (; iter->Valid(); iter->Next()) { + auto current_centroid_result = iter->GetCentroid(); + if (!current_centroid_result) { + return rocksdb::Status::InvalidArgument(current_centroid_result.Msg()); + } + auto& current_centroid = *current_centroid_result; + if (val < current_centroid.mean) { + break; + } + accum_weight += current_centroid.weight; + if (val > current_centroid.mean) { + weight += current_centroid.weight; + continue; + } + if (val == current_centroid.mean) { + weight += current_centroid.weight / 2; + continue; + } + } + double cdf_val = (weight / 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); return Database::GetMetadata(context, {kRedisTDigest}, ns_key, metadata); diff --git a/src/types/redis_tdigest.h b/src/types/redis_tdigest.h index 8cf4562d293..450f06ff784 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 cdf_values; +}; + struct TDigestQuantitleResult { std::optional> quantiles; }; @@ -92,6 +96,8 @@ class TDigest : public SubKeyScanner { rocksdb::Status TrimmedMean(engine::Context& ctx, const Slice& digest_name, double low_cut_quantile, double high_cut_quantile, TDigestTrimmedMeanResult* result); rocksdb::Status GetMetaData(engine::Context& context, const Slice& digest_name, TDigestMetadata* metadata); + rocksdb::Status CDFUniqSorted(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, + TDigestCDFResult* result); private: enum class SegmentType : uint8_t { kBuffer = 0, kCentroids = 1, kGuardFlag = 0xFF }; diff --git a/tests/cppunit/types/tdigest_test.cc b/tests/cppunit/types/tdigest_test.cc index f82519d2f73..26e04baa1b5 100644 --- a/tests/cppunit/types/tdigest_test.cc +++ b/tests/cppunit/types/tdigest_test.cc @@ -30,13 +30,14 @@ #include #include #include +#include #include #include +#include #include #include #include -#include "logging.h" #include "storage/redis_metadata.h" #include "test_base.h" #include "time_util.h" @@ -948,3 +949,157 @@ TEST_F(RedisTDigestTest, MergeWithUserSpecifiedCompression) { // Verify total observations: dest(1) + src(1) = 2 EXPECT_EQ(metadata.total_observations, 2); } + +TEST_F(RedisTDigestTest, CDF_Test) { + 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 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 cdf_vals = {0, 1, 2, 3, 4, 5, 6}; + redis::TDigestCDFResult result; + + status = tdigest_->CDFUniqSorted(*ctx_, cdf_tdigest_name, cdf_vals, &result); + ASSERT_TRUE(status.ok()) << status.ToString(); + + std::vector 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, CDF_returns_nan_on_empty_tdigest) { + 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 values = {0.0, 1.0, 2.0, 3.0}; + redis::TDigestCDFResult result; + + status = tdigest_->CDFUniqSorted(*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, CDF_duplicate_values) { + 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 cdf_vals = {5, 10, 20, 25}; + redis::TDigestCDFResult result; + status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + ASSERT_TRUE(status.ok()) << status.ToString(); + + std::vector 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, CDF_uniform_distribution) { + 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 samples = ranges::views::iota(1, 101) | + ranges::views::transform([](int i) { return (double)i; }) | + ranges::to>(); + status = tdigest_->Add(*ctx_, test_digest_name, samples); + ASSERT_TRUE(status.ok()); + + std::vector cdf_vals = {1, 25, 50, 75, 100}; + redis::TDigestCDFResult result; + status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + ASSERT_TRUE(status.ok()) << status.ToString(); + + std::vector 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, CDF_multiple_adds) { + 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 samples1 = {1, 2, 3, 4, 5}; + std::vector 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 cdf_vals = {1, 5, 7, 10}; + redis::TDigestCDFResult result; + status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + ASSERT_TRUE(status.ok()); + + std::vector 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, CDF_skewed_distribution) { + 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 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(i); })) | + ranges::to>(); + + status = tdigest_->Add(*ctx_, test_digest_name, samples); + ASSERT_TRUE(status.ok()); + + std::vector cdf_vals = {0, 1, 5, 10}; + redis::TDigestCDFResult result; + status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + ASSERT_TRUE(status.ok()); + + std::vector 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]); + } +} diff --git a/tests/gocase/unit/type/tdigest/tdigest_test.go b/tests/gocase/unit/type/tdigest/tdigest_test.go index e4daa3e188a..7144f741f2e 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,145 @@ func tdigestTestsByRankAndByRevRank(t *testing.T, configs util.KvrocksServerConf } } }) + + t.Run("tdigest.cdf with different arguments", func(t *testing.T) { + keyPrefix := "tdigest_cdf_" + + 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) + 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) + 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 { + 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) + } + } + }) } From 0fdd2bf273254a7d32abfce1b6648f6bd11d2dda Mon Sep 17 00:00:00 2001 From: edward_xu Date: Tue, 14 Jul 2026 00:03:48 +0800 Subject: [PATCH 2/6] refactor code for better sematics of --- src/commands/cmd_tdigest.cc | 35 +++++------------------------ src/types/redis_tdigest.cc | 35 +++++++++++++++++++++-------- src/types/redis_tdigest.h | 8 +++++-- tests/cppunit/types/tdigest_test.cc | 33 ++++++++++++++++++++++----- 4 files changed, 65 insertions(+), 46 deletions(-) diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc index a101dc810f2..a81c1475785 100644 --- a/src/commands/cmd_tdigest.cc +++ b/src/commands/cmd_tdigest.cc @@ -18,10 +18,7 @@ * */ -#include #include -#include -#include #include #include "command_parser.h" @@ -563,7 +560,7 @@ class CommandTDigestCDF : public Commander { Status Parse(const std::vector &args) override { if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments}; key_name_ = args[1]; - std::map> unique_inputs; + inputs_.reserve(args.size() - 2); for (size_t i = 2; i < args.size(); i++) { auto value = ParseFloat(args[i]); if (!value) { @@ -572,25 +569,15 @@ class CommandTDigestCDF : public Commander { if (std::isnan(*value)) { return {Status::RedisParseErr, errValueIsNotFloat}; } - if (unique_inputs.find(*value) == unique_inputs.cend()) { - unique_inputs[*value] = std::vector{i - 2}; - } else { - unique_inputs[*value].push_back(i - 2); - } + inputs_.push_back(*value); } - sorted_unique_inputs_with_idx_ = - ranges::views::transform(unique_inputs, - [](const auto &pair) { return std::make_pair(pair.first, pair.second); }) | - ranges::to_vector; - num_inputs_ = args.size() - 2; return Status::OK(); } Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override { TDigest tdigest(srv->storage, conn->GetNamespace()); TDigestCDFResult result; - std::vector uniq_cdfs = sorted_unique_inputs_with_idx_ | ranges::views::keys | ranges::to_vector; - auto s = tdigest.CDFUniqSorted(ctx, key_name_, uniq_cdfs, &result); + auto s = tdigest.CDF(ctx, key_name_, inputs_, &result); if (!s.ok()) { if (s.IsNotFound()) { return {Status::RedisExecErr, errKeyNotFound}; @@ -598,24 +585,14 @@ class CommandTDigestCDF : public Commander { return {Status::RedisExecErr, s.ToString()}; } - std::vector> cdf_uniq_results_with_idx = - ranges::views::transform(result.cdf_values, util::Float2String) | ranges::views::enumerate | ranges::to_vector; - std::vector cdf_result(num_inputs_, kNan); - - for (const auto &[idx, result_str] : cdf_uniq_results_with_idx) { - for (const auto &origin_idx : sorted_unique_inputs_with_idx_[idx].second) { - cdf_result[origin_idx] = result_str; - } - } - - *output = conn->MultiBulkString(cdf_result); + *output = + conn->MultiBulkString(result.cdf_values | ranges::views::transform(util::Float2String) | ranges::to_vector); return Status::OK(); } private: std::string key_name_; - size_t num_inputs_; - std::vector>> sorted_unique_inputs_with_idx_; + std::vector inputs_; }; std::vector GetMergeKeyRange(const std::vector &args) { diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc index 4af131e21e9..34e28717266 100644 --- a/src/types/redis_tdigest.cc +++ b/src/types/redis_tdigest.cc @@ -28,12 +28,14 @@ #include #include +#include #include #include #include #include #include #include +#include #include #include #include @@ -571,17 +573,32 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest, return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); } -rocksdb::Status TDigest::CDFUniqSorted(engine::Context& ctx, const Slice& digest_name, - const std::vector& inputs, TDigestCDFResult* result) { - if (!std::is_sorted(inputs.cbegin(), inputs.cend())) { - return rocksdb::Status::InvalidArgument( - "Internal error: inputs must be sorted in ascending order for CDF computation."); +rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, + TDigestCDFResult* result) { + std::map> sorted_unique_inputs_with_idx; + for (size_t i = 0; i < inputs.size(); ++i) { + sorted_unique_inputs_with_idx[inputs[i]].push_back(i); } - if (std::set(inputs.cbegin(), inputs.cend()).size() != inputs.size()) { - return rocksdb::Status::InvalidArgument("Internal error: inputs must be unique for CDF computation."); + std::vector cdf_values; + if (auto status = cdfUniqSorted(ctx, digest_name, sorted_unique_inputs_with_idx | ranges::views::keys | ranges::to_vector, &cdf_values); !status.ok()) { + return status; + } + result->cdf_values.resize(inputs.size(), std::numeric_limits::quiet_NaN()); + + auto idx = 0; + for (auto iter = sorted_unique_inputs_with_idx.cbegin(); iter != sorted_unique_inputs_with_idx.cend(); ++iter) { + for (auto original_idx : iter->second) { + result->cdf_values[original_idx] = cdf_values[idx]; + } + ++idx; } + return rocksdb::Status::OK(); +} + +rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest_name, + const std::vector& inputs, std::vector* values) { auto ns_key = AppendNamespacePrefix(digest_name); TDigestMetadata metadata; { @@ -592,7 +609,7 @@ rocksdb::Status TDigest::CDFUniqSorted(engine::Context& ctx, const Slice& digest } if (metadata.total_observations == 0) { - result->cdf_values = std::vector(inputs.size(), std::numeric_limits::quiet_NaN()); + *values = std::vector(inputs.size(), std::numeric_limits::quiet_NaN()); return rocksdb::Status::OK(); } @@ -655,7 +672,7 @@ rocksdb::Status TDigest::CDFUniqSorted(engine::Context& ctx, const Slice& digest double cdf_val = (weight / total_weight); results.push_back(cdf_val); } - result->cdf_values = results; + *values = std::move(results); return rocksdb::Status::OK(); } diff --git a/src/types/redis_tdigest.h b/src/types/redis_tdigest.h index 450f06ff784..1c8758f08f0 100644 --- a/src/types/redis_tdigest.h +++ b/src/types/redis_tdigest.h @@ -96,8 +96,9 @@ class TDigest : public SubKeyScanner { rocksdb::Status TrimmedMean(engine::Context& ctx, const Slice& digest_name, double low_cut_quantile, double high_cut_quantile, TDigestTrimmedMeanResult* result); rocksdb::Status GetMetaData(engine::Context& context, const Slice& digest_name, TDigestMetadata* metadata); - rocksdb::Status CDFUniqSorted(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, - TDigestCDFResult* result); + + rocksdb::Status CDF(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, + TDigestCDFResult* result); private: enum class SegmentType : uint8_t { kBuffer = 0, kCentroids = 1, kGuardFlag = 0xFF }; @@ -166,5 +167,8 @@ class TDigest : public SubKeyScanner { Centroid* centroid) const; rocksdb::Status prepareRankData(engine::Context& ctx, const Slice& digest_name, TDigestMetadata& metadata, std::vector& centroids); + + rocksdb::Status cdfUniqSorted(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, + std::vector* values); }; } // namespace redis diff --git a/tests/cppunit/types/tdigest_test.cc b/tests/cppunit/types/tdigest_test.cc index 26e04baa1b5..baeef4b1114 100644 --- a/tests/cppunit/types/tdigest_test.cc +++ b/tests/cppunit/types/tdigest_test.cc @@ -964,7 +964,7 @@ TEST_F(RedisTDigestTest, CDF_Test) { std::vector cdf_vals = {0, 1, 2, 3, 4, 5, 6}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, cdf_tdigest_name, cdf_vals, &result); + status = tdigest_->CDF(*ctx_, cdf_tdigest_name, cdf_vals, &result); ASSERT_TRUE(status.ok()) << status.ToString(); std::vector expected = {0.00, 0.03, 0.13, 0.29, 0.53, 0.83, 1.00}; @@ -986,7 +986,7 @@ TEST_F(RedisTDigestTest, CDF_returns_nan_on_empty_tdigest) { std::vector values = {0.0, 1.0, 2.0, 3.0}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, values, &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) { @@ -1007,7 +1007,7 @@ TEST_F(RedisTDigestTest, CDF_duplicate_values) { std::vector cdf_vals = {5, 10, 20, 25}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result); ASSERT_TRUE(status.ok()) << status.ToString(); std::vector expected = {0, 0.3, 0.8, 1}; @@ -1017,6 +1017,27 @@ TEST_F(RedisTDigestTest, CDF_duplicate_values) { } } +TEST_F(RedisTDigestTest, CDF_signed_zero_queries) { + 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 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, CDF_uniform_distribution) { std::string test_digest_name = "test_cdf_uniform" + std::to_string(util::GetTimeStampMS()); @@ -1033,7 +1054,7 @@ TEST_F(RedisTDigestTest, CDF_uniform_distribution) { std::vector cdf_vals = {1, 25, 50, 75, 100}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result); ASSERT_TRUE(status.ok()) << status.ToString(); std::vector expected = {0.01, 0.25, 0.50, 0.75, 1.00}; @@ -1061,7 +1082,7 @@ TEST_F(RedisTDigestTest, CDF_multiple_adds) { std::vector cdf_vals = {1, 5, 7, 10}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result); ASSERT_TRUE(status.ok()); std::vector expected = {0.10, 0.50, 0.70, 1.00}; @@ -1092,7 +1113,7 @@ TEST_F(RedisTDigestTest, CDF_skewed_distribution) { std::vector cdf_vals = {0, 1, 5, 10}; redis::TDigestCDFResult result; - status = tdigest_->CDFUniqSorted(*ctx_, test_digest_name, cdf_vals, &result); + status = tdigest_->CDF(*ctx_, test_digest_name, cdf_vals, &result); ASSERT_TRUE(status.ok()); std::vector expected = {0.4545, 0.91, 0.95, 1.00}; From d481a8b3275e5345745926a1d34e95c4bcdbd2e5 Mon Sep 17 00:00:00 2001 From: edward_xu Date: Tue, 14 Jul 2026 00:13:53 +0800 Subject: [PATCH 3/6] fix formatter --- src/types/redis_tdigest.cc | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc index 34e28717266..cc23e6f6616 100644 --- a/src/types/redis_tdigest.cc +++ b/src/types/redis_tdigest.cc @@ -581,7 +581,9 @@ rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, con } std::vector cdf_values; - if (auto status = cdfUniqSorted(ctx, digest_name, sorted_unique_inputs_with_idx | ranges::views::keys | ranges::to_vector, &cdf_values); !status.ok()) { + if (auto status = cdfUniqSorted(ctx, digest_name, + sorted_unique_inputs_with_idx | ranges::views::keys | ranges::to_vector, &cdf_values); + !status.ok()) { return status; } result->cdf_values.resize(inputs.size(), std::numeric_limits::quiet_NaN()); From cee4cf1e32e7357f3ba122a5cd6b8af400371071 Mon Sep 17 00:00:00 2001 From: edward_xu Date: Tue, 14 Jul 2026 00:37:06 +0800 Subject: [PATCH 4/6] fix copilot comments --- src/types/redis_tdigest.cc | 37 +++++++---------------------- tests/cppunit/types/tdigest_test.cc | 14 +++++------ 2 files changed, 16 insertions(+), 35 deletions(-) diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc index cc23e6f6616..efa849b7317 100644 --- a/src/types/redis_tdigest.cc +++ b/src/types/redis_tdigest.cc @@ -588,7 +588,7 @@ rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, con } result->cdf_values.resize(inputs.size(), std::numeric_limits::quiet_NaN()); - auto idx = 0; + size_t idx = 0; for (auto iter = sorted_unique_inputs_with_idx.cbegin(); iter != sorted_unique_inputs_with_idx.cend(); ++iter) { for (auto original_idx : iter->second) { result->cdf_values[original_idx] = cdf_values[idx]; @@ -615,27 +615,8 @@ rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest return rocksdb::Status::OK(); } - 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(); + if (auto status = mergeNodes(ctx, ns_key, &metadata); !status.ok()) { + return status; } } @@ -658,18 +639,18 @@ rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest return rocksdb::Status::InvalidArgument(current_centroid_result.Msg()); } auto& current_centroid = *current_centroid_result; - if (val < current_centroid.mean) { + + const int cmp = DoubleCompare(val, current_centroid.mean); + + if (cmp < 0) { break; } accum_weight += current_centroid.weight; - if (val > current_centroid.mean) { + if (cmp > 0) { weight += current_centroid.weight; continue; } - if (val == current_centroid.mean) { - weight += current_centroid.weight / 2; - continue; - } + weight += current_centroid.weight / 2; } double cdf_val = (weight / total_weight); results.push_back(cdf_val); diff --git a/tests/cppunit/types/tdigest_test.cc b/tests/cppunit/types/tdigest_test.cc index baeef4b1114..6ff51ea1110 100644 --- a/tests/cppunit/types/tdigest_test.cc +++ b/tests/cppunit/types/tdigest_test.cc @@ -950,7 +950,7 @@ TEST_F(RedisTDigestTest, MergeWithUserSpecifiedCompression) { EXPECT_EQ(metadata.total_observations, 2); } -TEST_F(RedisTDigestTest, CDF_Test) { +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); @@ -975,7 +975,7 @@ TEST_F(RedisTDigestTest, CDF_Test) { } } -TEST_F(RedisTDigestTest, CDF_returns_nan_on_empty_tdigest) { +TEST_F(RedisTDigestTest, CDFReturnsNaNOnEmptyTDigest) { std::string test_digest_name = "test_digest_cdf_nan" + std::to_string(util::GetTimeStampMS()); bool exists = false; @@ -994,7 +994,7 @@ TEST_F(RedisTDigestTest, CDF_returns_nan_on_empty_tdigest) { } } -TEST_F(RedisTDigestTest, CDF_duplicate_values) { +TEST_F(RedisTDigestTest, CDFDuplicateValues) { std::string test_digest_name = "test_cdf_duplicates" + std::to_string(util::GetTimeStampMS()); bool exists = false; @@ -1017,7 +1017,7 @@ TEST_F(RedisTDigestTest, CDF_duplicate_values) { } } -TEST_F(RedisTDigestTest, CDF_signed_zero_queries) { +TEST_F(RedisTDigestTest, CDFSignedZeroQueries) { std::string test_digest_name = "test_cdf_signed_zero" + std::to_string(util::GetTimeStampMS()); bool exists = false; @@ -1038,7 +1038,7 @@ TEST_F(RedisTDigestTest, CDF_signed_zero_queries) { EXPECT_NEAR(result.cdf_values[1], 0.5, 0.001); } -TEST_F(RedisTDigestTest, CDF_uniform_distribution) { +TEST_F(RedisTDigestTest, CDFUniformDistribution) { std::string test_digest_name = "test_cdf_uniform" + std::to_string(util::GetTimeStampMS()); bool exists = false; @@ -1065,7 +1065,7 @@ TEST_F(RedisTDigestTest, CDF_uniform_distribution) { } } -TEST_F(RedisTDigestTest, CDF_multiple_adds) { +TEST_F(RedisTDigestTest, CDFMultipleAdds) { std::string test_digest_name = "test_cdf_multiadd" + std::to_string(util::GetTimeStampMS()); bool exists = false; @@ -1094,7 +1094,7 @@ TEST_F(RedisTDigestTest, CDF_multiple_adds) { } } -TEST_F(RedisTDigestTest, CDF_skewed_distribution) { +TEST_F(RedisTDigestTest, CDFSkewedDistribution) { std::string test_digest_name = "test_cdf_skewed" + std::to_string(util::GetTimeStampMS()); bool exists = false; From a237dbc8bef4ad96db45f90bd0fad73c6a73aa47 Mon Sep 17 00:00:00 2001 From: edward_xu Date: Fri, 17 Jul 2026 00:18:27 +0800 Subject: [PATCH 5/6] fix the interpolation problem. --- src/types/redis_tdigest.cc | 62 ++----------- src/types/redis_tdigest.h | 3 - src/types/tdigest.h | 130 +++++++++++++++++++++++++++- tests/cppunit/types/tdigest_test.cc | 47 ++++++++++ 4 files changed, 180 insertions(+), 62 deletions(-) diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc index efa849b7317..38c836dad8b 100644 --- a/src/types/redis_tdigest.cc +++ b/src/types/redis_tdigest.cc @@ -37,7 +37,6 @@ #include #include #include -#include #include #include "commands/error_constants.h" @@ -575,32 +574,6 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest, rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, TDigestCDFResult* result) { - std::map> sorted_unique_inputs_with_idx; - for (size_t i = 0; i < inputs.size(); ++i) { - sorted_unique_inputs_with_idx[inputs[i]].push_back(i); - } - - std::vector cdf_values; - if (auto status = cdfUniqSorted(ctx, digest_name, - sorted_unique_inputs_with_idx | ranges::views::keys | ranges::to_vector, &cdf_values); - !status.ok()) { - return status; - } - result->cdf_values.resize(inputs.size(), std::numeric_limits::quiet_NaN()); - - size_t idx = 0; - for (auto iter = sorted_unique_inputs_with_idx.cbegin(); iter != sorted_unique_inputs_with_idx.cend(); ++iter) { - for (auto original_idx : iter->second) { - result->cdf_values[original_idx] = cdf_values[idx]; - } - ++idx; - } - - return rocksdb::Status::OK(); -} - -rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest_name, - const std::vector& inputs, std::vector* values) { auto ns_key = AppendNamespacePrefix(digest_name); TDigestMetadata metadata; { @@ -611,7 +584,7 @@ rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest } if (metadata.total_observations == 0) { - *values = std::vector(inputs.size(), std::numeric_limits::quiet_NaN()); + result->cdf_values = std::vector(inputs.size(), std::numeric_limits::quiet_NaN()); return rocksdb::Status::OK(); } @@ -626,36 +599,11 @@ rocksdb::Status TDigest::cdfUniqSorted(engine::Context& ctx, const Slice& digest } auto dump_centroids = DummyCentroids(metadata, centroids); - auto total_weight = dump_centroids.TotalWeight(); - auto iter = dump_centroids.Begin(); - double accum_weight = 0.; - std::vector results; - results.reserve(inputs.size()); - for (const auto val : inputs) { - double weight = accum_weight; - for (; iter->Valid(); iter->Next()) { - auto current_centroid_result = iter->GetCentroid(); - if (!current_centroid_result) { - return rocksdb::Status::InvalidArgument(current_centroid_result.Msg()); - } - auto& current_centroid = *current_centroid_result; - - const int cmp = DoubleCompare(val, current_centroid.mean); - - if (cmp < 0) { - break; - } - accum_weight += current_centroid.weight; - if (cmp > 0) { - weight += current_centroid.weight; - continue; - } - weight += current_centroid.weight / 2; - } - double cdf_val = (weight / total_weight); - results.push_back(cdf_val); + 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()); } - *values = std::move(results); return rocksdb::Status::OK(); } diff --git a/src/types/redis_tdigest.h b/src/types/redis_tdigest.h index 1c8758f08f0..445445c0f45 100644 --- a/src/types/redis_tdigest.h +++ b/src/types/redis_tdigest.h @@ -167,8 +167,5 @@ class TDigest : public SubKeyScanner { Centroid* centroid) const; rocksdb::Status prepareRankData(engine::Context& ctx, const Slice& digest_name, TDigestMetadata& metadata, std::vector& centroids); - - rocksdb::Status cdfUniqSorted(engine::Context& ctx, const Slice& digest_name, const std::vector& inputs, - std::vector* values); }; } // namespace redis diff --git a/src/types/tdigest.h b/src/types/tdigest.h index 531f693ed20..3e12110f168 100644 --- a/src/types/tdigest.h +++ b/src/types/tdigest.h @@ -22,10 +22,11 @@ #include +#include +#include +#include #include #include -#include -#include #include #include "common/status.h" @@ -172,6 +173,131 @@ struct DoubleComparator { bool operator()(const double& a, const double& b) const { return DoubleCompare(a, b) == -1; } }; +inline double InterpolateRank(double value, double lower_value, double upper_value, double lower_rank, + double upper_rank) { + if (DoubleEqual(lower_value, upper_value)) { + return (lower_rank + upper_rank) / 2; + } + return Lerp(lower_rank, upper_rank, (value - lower_value) / (upper_value - lower_value)); +} + +// 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. +inline Status TDigestCDF(const std::vector& centroids, double min, double max, double total_weight, + const std::vector& inputs, std::vector* result) { + if (centroids.empty() || total_weight <= 0) { + return Status{Status::InvalidArgument, "invalid or empty tdigest"}; + } + + std::map> 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 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; + + struct CentroidGroup { + double mean; + double weight; + double center_rank; + }; + + std::vector groups; + groups.reserve(centroids.size()); + double cumulative_weight = 0; + for (size_t i = 0; i < centroids.size();) { + double group_weight = 0; + const auto mean = centroids[i].mean; + do { + group_weight += centroids[i].weight; + ++i; + } while (i < centroids.size() && DoubleEqual(mean, centroids[i].mean)); + + groups.push_back({ + .mean = mean, + .weight = group_weight, + .center_rank = cumulative_weight + group_weight / 2, + }); + cumulative_weight += group_weight; + } + + std::vector sorted_results; + sorted_results.reserve(sorted_unique_inputs.size()); + size_t group_idx = 0; + for (const auto value : sorted_unique_inputs) { + if (value < min) { + sorted_results.push_back(0); + continue; + } + if (value > max) { + sorted_results.push_back(1); + continue; + } + + if (value == min) { + auto rank = DoubleEqual(groups.front().mean, min) ? groups.front().center_rank : kHalfSingletonBoundaryWeight; + sorted_results.push_back(rank / total_weight); + continue; + } + if (value == max) { + auto rank = DoubleEqual(groups.back().mean, max) ? groups.back().center_rank + : (total_weight - kHalfSingletonBoundaryWeight); + sorted_results.push_back(rank / total_weight); + continue; + } + + while (group_idx < groups.size() && DoubleCompare(groups[group_idx].mean, value) < 0) { + ++group_idx; + } + + double rank = 0; + if (group_idx == groups.size()) { + const auto& last = groups.back(); + rank = InterpolateRank(value, last.mean, max, last.center_rank, total_weight - kSingletonBoundaryWeight); + } else if (DoubleEqual(value, groups[group_idx].mean)) { + rank = groups[group_idx].center_rank; + } else if (group_idx == 0) { + rank = InterpolateRank(value, min, groups.front().mean, kSingletonBoundaryWeight, groups.front().center_rank); + } else { + const auto& left = groups[group_idx - 1]; + const auto& right = groups[group_idx]; + auto weight_before_left = left.center_rank - left.weight / 2; + // When both adjacent centroids are singletons, include the left exact sample and exclude the right one. + if (left.weight == kSingletonBoundaryWeight && right.weight == kSingletonBoundaryWeight) { + rank = weight_before_left + kSingletonBoundaryWeight; + } else { + // Exclude singleton half-weights from interpolation because singleton centroids are exact samples. + double left_excluded_weight = left.weight == kSingletonBoundaryWeight ? kHalfSingletonBoundaryWeight : 0; + double right_excluded_weight = right.weight == kSingletonBoundaryWeight ? kHalfSingletonBoundaryWeight : 0; + double rank_span = (left.weight + right.weight) / 2 - left_excluded_weight - right_excluded_weight; + double base_rank = weight_before_left + left.weight / 2 + left_excluded_weight; + rank = base_rank + rank_span * (value - left.mean) / (right.mean - left.mean); + } + } + + rank = std::clamp(rank, 0.0, total_weight); + sorted_results.push_back(rank / total_weight); + } + + result->clear(); + result->resize(inputs.size(), std::numeric_limits::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] = sorted_results[i]; + } + } + + return Status::OK(); +} + template inline Status TDigestByRank(TD&& td, const std::vector& inputs, std::vector* result) { result->clear(); diff --git a/tests/cppunit/types/tdigest_test.cc b/tests/cppunit/types/tdigest_test.cc index 6ff51ea1110..71d3bad3712 100644 --- a/tests/cppunit/types/tdigest_test.cc +++ b/tests/cppunit/types/tdigest_test.cc @@ -1065,6 +1065,53 @@ TEST_F(RedisTDigestTest, CDFUniformDistribution) { } } +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 samples = ranges::views::iota(0, 100) | + ranges::views::transform([](int i) { return static_cast(i); }) | + ranges::to>(); + status = tdigest_->Add(*ctx_, test_digest_name, samples); + ASSERT_TRUE(status.ok()); + + std::vector 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 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 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()); From 7d77515204cd33cc98417b9373ceb89cb85e850a Mon Sep 17 00:00:00 2001 From: edward_xu Date: Fri, 17 Jul 2026 00:38:31 +0800 Subject: [PATCH 6/6] make command output with double. --- src/commands/cmd_tdigest.cc | 6 ++-- .../gocase/unit/type/tdigest/tdigest_test.go | 36 +++++++++++++++---- 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc index a81c1475785..1492398595f 100644 --- a/src/commands/cmd_tdigest.cc +++ b/src/commands/cmd_tdigest.cc @@ -585,8 +585,10 @@ class CommandTDigestCDF : public Commander { return {Status::RedisExecErr, s.ToString()}; } - *output = - conn->MultiBulkString(result.cdf_values | ranges::views::transform(util::Float2String) | ranges::to_vector); + output->append(redis::MultiLen(result.cdf_values.size())); + for (auto const value : result.cdf_values) { + output->append(conn->Double(value)); + } return Status::OK(); } diff --git a/tests/gocase/unit/type/tdigest/tdigest_test.go b/tests/gocase/unit/type/tdigest/tdigest_test.go index 7144f741f2e..1cfa6bf0a17 100644 --- a/tests/gocase/unit/type/tdigest/tdigest_test.go +++ b/tests/gocase/unit/type/tdigest/tdigest_test.go @@ -1313,6 +1313,7 @@ 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) @@ -1343,7 +1344,7 @@ func tdigestTestsByRankAndByRevRank(t *testing.T, configs util.KvrocksServerConf require.NoError(t, err) require.Len(t, vals, 4) - // empty tdigest should return "nan" + // 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") @@ -1351,7 +1352,13 @@ func tdigestTestsByRankAndByRevRank(t *testing.T, configs util.KvrocksServerConf vals, err = rsp.Slice() require.NoError(t, err) require.Len(t, vals, 1) - require.Equal(t, "nan", vals[0]) + 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") @@ -1359,7 +1366,15 @@ func tdigestTestsByRankAndByRevRank(t *testing.T, configs util.KvrocksServerConf vals, err = rsp.Slice() require.NoError(t, err) require.Len(t, vals, 4) - require.Equal(t, []interface{}{"nan", "nan", "nan", "nan"}, vals) + 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 @@ -1443,10 +1458,17 @@ func tdigestTestsByRankAndByRevRank(t *testing.T, configs util.KvrocksServerConf require.NoError(t, err) require.Len(t, vals, len(cdfArgs)) for i, v := range vals { - 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) + 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) } }