You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "SharonIV0x86 (via GitHub)" <gi...@apache.org> on 2025/09/06 17:37:32 UTC
[PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
SharonIV0x86 opened a new pull request, #3163:
URL: https://github.com/apache/kvrocks/pull/3163
This closes: https://github.com/apache/kvrocks/issues/2807
This PR implements the support for ``TDIGEST.CDF`` command [redis docs](https://redis.io/docs/latest/commands/tdigest.cdf/). Following the implementation of the ``Quantile`` function for accessing tdigest centroids.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#issuecomment-3480242181
Hi @SharonIV0x86 ,
Do you mind that I continue work on this PR base on your commits? 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "SharonIV0x86 (via GitHub)" <gi...@apache.org>.
SharonIV0x86 commented on PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#issuecomment-3480251210
> Hi @SharonIV0x86 ,
>
> Do you mind that I continue work on this PR base on your commits? 😊
>
> Best Regards,
> Edward
Yeah no issues, all yours!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2335959183
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
You can refer to PR #2620 for further understanding of the `MakeCmdAttr` flags.
https://github.com/apache/kvrocks/blob/unstable/src/server/redis_connection.cc#L528-L546
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#issuecomment-3284927506
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3163) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3163&metric=new_coverage&view=list) (required ≥ 50%)
 [37.7% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3163&metric=new_duplicated_lines_density&view=list) (required ≤ 20%)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3163)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "SharonIV0x86 (via GitHub)" <gi...@apache.org>.
SharonIV0x86 closed pull request #3163: feat(tdigest): Implement TDIGEST.CDF command
URL: https://github.com/apache/kvrocks/pull/3163
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "SharonIV0x86 (via GitHub)" <gi...@apache.org>.
SharonIV0x86 commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2340677201
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
I have made the changes.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
I have made this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2407046539
##########
tests/gocase/unit/type/tdigest/tdigest_test.go:
##########
@@ -518,4 +519,55 @@ func tdigestTests(t *testing.T, configs util.KvrocksServerConfigs) {
validation(newDestKey1)
validation(newDestKey2)
})
+ 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())
Review Comment:
Hi @SharonIV0x86 ,
We'd better add some tests with duplicated values to create different weights for some centroids.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,56 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
+ 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());
+ std::vector<std::string> cdf_result;
+ TDigestCDFResult result;
+ TDigestMetadata metadata;
+ auto meta_status = tdigest.GetMetaData(ctx, key_name_, &metadata);
+ std::vector<std::string> nan_results(values_.size(), "nan");
Review Comment:
Hi @SharonIV0x86 ,
We could move this `nan_results` construction into the empty element branch.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,56 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
+ 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());
+ std::vector<std::string> cdf_result;
+ TDigestCDFResult result;
+ TDigestMetadata metadata;
+ auto meta_status = tdigest.GetMetaData(ctx, key_name_, &metadata);
+ std::vector<std::string> nan_results(values_.size(), "nan");
Review Comment:
The `nan_results` could be constructed in the empty element branch.
##########
src/types/redis_tdigest.cc:
##########
@@ -414,6 +414,66 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest,
return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector<double>& inputs,
+ TDigestCDFResult* result) {
+ auto ns_key = AppendNamespacePrefix(digest_name);
+ TDigestMetadata metadata;
+ {
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+
+ if (auto status = getMetaDataByNsKey(ctx, ns_key, &metadata); !status.ok()) {
+ return status;
+ }
+
+ if (metadata.unmerged_nodes > 0) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisTDigest);
+ if (auto status = batch->PutLogData(log_data.Encode()); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = mergeCurrentBuffer(ctx, ns_key, batch, &metadata); !status.ok()) {
+ return status;
+ }
+
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ if (auto status = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); !status.ok()) {
+ return status;
+ }
+ ctx.RefreshLatestSnapshot();
+ }
+ }
+ std::vector<Centroid> centroids;
+ if (auto status = dumpCentroids(ctx, ns_key, metadata, ¢roids); !status.ok()) {
+ return status;
+ }
+ auto dump_centroids = DummyCentroids(metadata, centroids);
+ double total_weight = dump_centroids.TotalWeight();
+ std::vector<double> results;
+ for (double val : inputs) {
+ auto iter_begin = dump_centroids.Begin();
+ auto iter_end = dump_centroids.End();
+ double eq_count = 0;
+ double smaller_count = 0;
+ for (; iter_begin->Valid(); iter_begin->Next()) {
+ auto current_centroid = iter_begin->GetCentroid();
+ if (val > current_centroid->mean) {
+ smaller_count++;
+ } else if (val == current_centroid->mean) {
+ eq_count++;
+ }
+ }
+ double cdf_val = (smaller_count / total_weight) + ((eq_count / 2) / total_weight);
Review Comment:
Hi @SharonIV0x86 ,
It seems that we mistake the count with weight here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#issuecomment-3269073130
Hi @SharonIV0x86 ,
Thanks for your effort!
I will review this PR later today. 😊
Best Regards,
Edward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "SharonIV0x86 (via GitHub)" <gi...@apache.org>.
SharonIV0x86 commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2407060971
##########
src/types/redis_tdigest.cc:
##########
@@ -414,6 +414,66 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest,
return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector<double>& inputs,
+ TDigestCDFResult* result) {
+ auto ns_key = AppendNamespacePrefix(digest_name);
+ TDigestMetadata metadata;
+ {
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+
+ if (auto status = getMetaDataByNsKey(ctx, ns_key, &metadata); !status.ok()) {
+ return status;
+ }
+
+ if (metadata.unmerged_nodes > 0) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisTDigest);
+ if (auto status = batch->PutLogData(log_data.Encode()); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = mergeCurrentBuffer(ctx, ns_key, batch, &metadata); !status.ok()) {
+ return status;
+ }
+
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ if (auto status = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); !status.ok()) {
+ return status;
+ }
+ ctx.RefreshLatestSnapshot();
+ }
+ }
+ std::vector<Centroid> centroids;
+ if (auto status = dumpCentroids(ctx, ns_key, metadata, ¢roids); !status.ok()) {
+ return status;
+ }
+ auto dump_centroids = DummyCentroids(metadata, centroids);
+ double total_weight = dump_centroids.TotalWeight();
+ std::vector<double> results;
+ for (double val : inputs) {
+ auto iter_begin = dump_centroids.Begin();
+ auto iter_end = dump_centroids.End();
+ double eq_count = 0;
+ double smaller_count = 0;
+ for (; iter_begin->Valid(); iter_begin->Next()) {
+ auto current_centroid = iter_begin->GetCentroid();
+ if (val > current_centroid->mean) {
+ smaller_count++;
+ } else if (val == current_centroid->mean) {
+ eq_count++;
+ }
+ }
+ double cdf_val = (smaller_count / total_weight) + ((eq_count / 2) / total_weight);
Review Comment:
Okay, I will take a look into it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2333953913
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,51 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ key_name_ = args[1];
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
+ values_.reserve(args.size() - 2);
+ for (size_t i = 2; i < args.size(); i++) {
+ auto value = ParseFloat(args[i]);
+ if (!value) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ values_.push_back(*value);
+ }
+ return Status::OK();
+ }
+ Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+ TDigest tdigest(srv->storage, conn->GetNamespace());
+ std::vector<std::string> cdf_result;
+ TDigestCDFResult result;
+ TDigestMetadata metadata;
+ auto meta_status = tdigest.GetMetaData(ctx, key_name_, &metadata);
+ if (!meta_status.ok()) {
+ if (meta_status.IsNotFound()) {
+ return {Status::RedisExecErr, errKeyNotFound};
+ }
+ return {Status::RedisExecErr, meta_status.ToString()};
+ }
+ if (metadata.total_observations == 0) {
Review Comment:
I have tested with Redis Docker, it should be the ["nan"] vector with the same size as the input.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
It is a read-only command, but with a lock guard in merge action.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,51 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ key_name_ = args[1];
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
+ values_.reserve(args.size() - 2);
+ for (size_t i = 2; i < args.size(); i++) {
+ auto value = ParseFloat(args[i]);
+ if (!value) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ values_.push_back(*value);
+ }
+ return Status::OK();
+ }
+ Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+ TDigest tdigest(srv->storage, conn->GetNamespace());
+ std::vector<std::string> cdf_result;
+ TDigestCDFResult result;
+ TDigestMetadata metadata;
+ auto meta_status = tdigest.GetMetaData(ctx, key_name_, &metadata);
+ if (!meta_status.ok()) {
+ if (meta_status.IsNotFound()) {
+ return {Status::RedisExecErr, errKeyNotFound};
+ }
+ return {Status::RedisExecErr, meta_status.ToString()};
+ }
+ if (metadata.total_observations == 0) {
+ *output = redis::MultiBulkString(RESP::v2, cdf_result);
+ return Status::OK();
+ }
+ auto s = tdigest.CDF(ctx, key_name_, values_, &result);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ for (const auto &val : result.cdf_values) {
+ cdf_result.push_back(std::to_string(val));
Review Comment:
We could use `util::Float2String` as the `quantile` command.
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,51 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ key_name_ = args[1];
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
Review Comment:
Maybe we could check the vector size at beginning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2332163165
##########
src/types/redis_tdigest.cc:
##########
@@ -414,6 +414,66 @@ rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest,
return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}
+rocksdb::Status TDigest::CDF(engine::Context& ctx, const Slice& digest_name, const std::vector<double>& inputs,
+ TDigestCDFResult* result) {
+ auto ns_key = AppendNamespacePrefix(digest_name);
+ TDigestMetadata metadata;
+ {
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+
+ if (auto status = getMetaDataByNsKey(ctx, ns_key, &metadata); !status.ok()) {
+ return status;
+ }
+
+ if (metadata.unmerged_nodes > 0) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisTDigest);
+ if (auto status = batch->PutLogData(log_data.Encode()); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = mergeCurrentBuffer(ctx, ns_key, batch, &metadata); !status.ok()) {
+ return status;
+ }
+
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ if (auto status = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); !status.ok()) {
+ return status;
+ }
+ ctx.RefreshLatestSnapshot();
+ }
+ }
+ std::vector<Centroid> centroids;
+ if (auto status = dumpCentroids(ctx, ns_key, metadata, ¢roids); !status.ok()) {
+ return status;
+ }
+ auto dump_centroids = DummyCentroids(metadata, centroids);
+ double total_weight = dump_centroids.TotalWeight();
+ std::vector<double> results;
+ for (double val : inputs) {
+ auto iter_begin = dump_centroids.Begin();
+ auto iter_end = dump_centroids.End();
+ double eq_count = 0;
+ double smaller_count = 0;
+ for (; iter_begin->Valid(); iter_begin->Next()) {
+ auto current_centroid = iter_begin->GetCentroid();
+ if (val > current_centroid->mean) {
+ smaller_count++;
+ } else if (val == current_centroid->mean) {
+ eq_count++;
+ }
+ }
+ double cdf_val = (smaller_count / total_weight) + ((eq_count / 2) / total_weight);
Review Comment:
The CDF calculation is incorrect. It's counting the number of centroids rather than their weights. It should accumulate `current_centroid->weight` instead of incrementing by 1 for proper cumulative distribution calculation.
```suggestion
double eq_weight = 0;
double smaller_weight = 0;
for (; iter_begin->Valid(); iter_begin->Next()) {
auto current_centroid = iter_begin->GetCentroid();
if (val > current_centroid->mean) {
smaller_weight += current_centroid->weight;
} else if (val == current_centroid->mean) {
eq_weight += current_centroid->weight;
}
}
double cdf_val = (smaller_weight / total_weight) + ((eq_weight / 2) / total_weight);
```
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
The TDIGEST.CDF command should be marked as 'read-only' instead of 'write' since it only reads data and doesn't modify the T-Digest structure.
```suggestion
MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "read-only", 1, 1, 1));
```
##########
src/commands/cmd_tdigest.cc:
##########
@@ -358,7 +358,51 @@ class CommandTDigestMerge : public Commander {
std::vector<std::string> source_keys_;
TDigestMergeOptions options_;
};
+class CommandTDigestCDF : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ key_name_ = args[1];
+ if (args.size() == 2) return {Status::RedisParseErr, errWrongNumOfArguments};
Review Comment:
The command registration specifies minimum 4 arguments (-4), but this validation only checks for exactly 2 arguments. It should validate that there are at least 3 arguments (command + key + at least one value).
```suggestion
if (args.size() < 3) return {Status::RedisParseErr, errWrongNumOfArguments};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "LindaSummer (via GitHub)" <gi...@apache.org>.
LindaSummer commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2335959183
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
You can refer to PR #2620 for further understanding of the `MakeCmdAttr` flags.
https://github.com/apache/kvrocks/blob/cf28ab1c451c4788d782f2c4b178934bd0210cf0/src/server/redis_connection.cc#L528-L546
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2406949634
##########
tests/gocase/unit/type/tdigest/tdigest_test.go:
##########
@@ -518,4 +519,55 @@ func tdigestTests(t *testing.T, configs util.KvrocksServerConfigs) {
validation(newDestKey1)
validation(newDestKey2)
})
+ 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])
+
+ // testing with a empry digest with multi-valued CDF
Review Comment:
Corrected spelling of 'empry' to 'empty'.
```suggestion
// testing with an empty digest with multi-valued CDF
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#issuecomment-3269402114
## [](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3163) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3163&metric=new_coverage&view=list) (required ≥ 50%)
 [37.8% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_kvrocks&pullRequest=3163&metric=new_duplicated_lines_density&view=list) (required ≤ 20%)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_kvrocks&pullRequest=3163)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(tdigest): Implement TDIGEST.CDF command [kvrocks]
Posted by "SharonIV0x86 (via GitHub)" <gi...@apache.org>.
SharonIV0x86 commented on code in PR #3163:
URL: https://github.com/apache/kvrocks/pull/3163#discussion_r2334228889
##########
src/commands/cmd_tdigest.cc:
##########
@@ -371,5 +415,6 @@ REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.crea
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange),
+ MakeCmdAttr<CommandTDigestCDF>("tdigest.cdf", -4, "write", 1, 1, 1));
Review Comment:
Yes, i will work on it, i was unsure about how the command registration works and what exactly those parameters do in the ``MakeCmdAttr`` so it will be helpful if you could explain it in a little detail.
Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org