You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kvrocks.apache.org by ed...@apache.org on 2025/08/04 12:43:22 UTC
(kvrocks) branch 2.13 created (now 154804c43)
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a change to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
at 154804c43 release: prepare source release apache-kvrocks-2.13.0
This branch includes the following new commits:
new 4ffbed4e9 chore(common): refactor parse utils via std::from_chars (#3052)
new d80a1f73e chore(ci): using golang-lint v2 (#3055)
new 3a1b63101 chore(deps): bump cpptrace to v1.0.3 (#3056)
new 65f6027b2 feat(tdigest): add `TDIGEST.MERGE` command implementation (#3054)
new 3f0ad06a6 build: disable LTO in debug mode (#3069)
new 9fa734e9e test: re-enable skipped TLS replica cases (#3068)
new 39575eba4 chore(ci): bump golangci-lint to 2.3.0 (#3071)
new f57e019f6 fix(config): the default log-dir should include stdout (#3085)
new 65681ebf8 Comment out the tdigest and its test cases due to it's not production ready
new 154804c43 release: prepare source release apache-kvrocks-2.13.0
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(kvrocks) 05/10: build: disable LTO in debug mode (#3069)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 3f0ad06a6fc01cbb2d868e34f2f1f149e75e5a46
Author: Twice <tw...@gmail.com>
AuthorDate: Tue Jul 22 13:49:03 2025 +0800
build: disable LTO in debug mode (#3069)
---
CMakeLists.txt | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 2fdff566a..26213b60b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -30,12 +30,17 @@ option(ASAN_WITH_LSAN "enable leak sanitizer while address sanitizer is enabled"
option(ENABLE_STATIC_LIBSTDCXX "link kvrocks with static library of libstd++ instead of shared library" ON)
option(ENABLE_LUAJIT "enable use of luaJIT instead of lua" ON)
option(ENABLE_OPENSSL "enable openssl to support tls connection" OFF)
-option(ENABLE_IPO "enable interprocedural optimization" ON)
+option(ENABLE_LTO "enable link-time optimization" ON)
set(SYMBOLIZE_BACKEND "" CACHE STRING "symbolization backend library for cpptrace (libbacktrace, libdwarf, or empty)")
set(PORTABLE 0 CACHE STRING "build a portable binary (disable arch-specific optimizations)")
# TODO: set ENABLE_NEW_ENCODING to ON when we are ready
option(ENABLE_NEW_ENCODING "enable new encoding (#1033) for storing 64bit size and expire time in milliseconds" ON)
+# to save build time in debug mode
+if (CMAKE_BUILD_TYPE STREQUAL "Debug")
+ set(ENABLE_LTO OFF)
+endif()
+
if (CMAKE_VERSION VERSION_GREATER_EQUAL "3.24.0")
cmake_policy(SET CMP0135 NEW)
endif()
@@ -136,7 +141,7 @@ endif()
if (CMAKE_HOST_APPLE)
set(DISABLE_JEMALLOC ON)
- set(ENABLE_IPO OFF)
+ set(ENABLE_LTO OFF)
endif ()
if(NOT DISABLE_JEMALLOC)
@@ -271,10 +276,10 @@ endif()
# disable LTO on GCC <= 9 due to an ICE
if((CMAKE_CXX_COMPILER_ID STREQUAL "GNU") AND (CMAKE_CXX_COMPILER_VERSION VERSION_LESS 10))
- set(ENABLE_IPO OFF)
+ set(ENABLE_LTO OFF)
endif()
-if(ENABLE_IPO)
+if(ENABLE_LTO)
include(CheckIPOSupported)
check_ipo_supported(RESULT ipo_result OUTPUT ipo_output LANGUAGES CXX)
@@ -284,7 +289,7 @@ if(ENABLE_IPO)
target_link_libraries(kvrocks_objs PUBLIC "-fuse-ld=lld")
endif()
else()
- message(WARNING "IPO is not supported: ${ipo_output}")
+ message(WARNING "LTO is not supported: ${ipo_output}")
endif()
endif()
(kvrocks) 02/10: chore(ci): using golang-lint v2 (#3055)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit d80a1f73ed229b647b3a13aa9180b38b5ff90bb6
Author: Aleks Lozovyuk <al...@gmail.com>
AuthorDate: Wed Jul 16 04:55:21 2025 +0300
chore(ci): using golang-lint v2 (#3055)
---
tests/gocase/.golangci.yml | 37 ++++++++++++++++++++++------
tests/gocase/unit/type/bitmap/bitmap_test.go | 4 +--
tests/gocase/unit/type/zset/zset_test.go | 13 ++++++----
tests/gocase/util/server.go | 2 +-
x.py | 2 +-
5 files changed, 41 insertions(+), 17 deletions(-)
diff --git a/tests/gocase/.golangci.yml b/tests/gocase/.golangci.yml
index eb6014735..1f34dbe31 100644
--- a/tests/gocase/.golangci.yml
+++ b/tests/gocase/.golangci.yml
@@ -15,28 +15,49 @@
# specific language governing permissions and limitations
# under the License.
+version: "2"
run:
- timeout: 10m
concurrency: 4
allow-parallel-runners: true
allow-serial-runners: true
linters:
- disable-all: true
+ default: none
enable:
- asciicheck
- bodyclose
+ - copyloopvar
- durationcheck
- errcheck
- - copyloopvar
- - gofmt
- - goimports
- - gosimple
- ineffassign
- makezero
- prealloc
- predeclared
- rowserrcheck
- staticcheck
- - stylecheck
- - typecheck
- unused
+ exclusions:
+ generated: lax
+ presets:
+ - comments
+ - common-false-positives
+ - legacy
+ - std-error-handling
+ paths:
+ - third_party$
+ - builtin$
+ - examples$
+formatters:
+ enable:
+ - gofmt
+ - goimports
+ exclusions:
+ generated: lax
+ paths:
+ - third_party$
+ - builtin$
+ - examples$
+linters-settings:
+ staticcheck:
+ checks:
+ - all
+ - '-QF1012' # disable the rule QF1012: Use fmt.Fprintf(...) instead of WriteString(fmt.Sprintf(...))
\ No newline at end of file
diff --git a/tests/gocase/unit/type/bitmap/bitmap_test.go b/tests/gocase/unit/type/bitmap/bitmap_test.go
index 04012f791..6211ae3c8 100644
--- a/tests/gocase/unit/type/bitmap/bitmap_test.go
+++ b/tests/gocase/unit/type/bitmap/bitmap_test.go
@@ -48,7 +48,7 @@ const (
func Set2SetBit(t *testing.T, rdb *redis.Client, ctx context.Context, key string, bs []byte) {
buf := bytes.NewBuffer([]byte{})
for _, v := range bs {
- buf.WriteString(fmt.Sprintf("%08b", v))
+ _, _ = fmt.Fprintf(buf, "%08b", v)
}
for index, value := range buf.String() {
require.NoError(t, rdb.SetBit(ctx, key, int64(index), int(value)-int('0')).Err())
@@ -74,7 +74,7 @@ func SimulateBitOp(op BITOP, values ...[]byte) string {
for _, value := range values {
buf := bytes.NewBuffer([]byte{})
for _, v := range value {
- buf.WriteString(fmt.Sprintf("%08b", v))
+ _, _ = fmt.Fprintf(buf, "%08b", v)
}
tmp := buf.String() + strings.Repeat("0", maxlen-len(buf.String()))
binaryArray = append(binaryArray, tmp)
diff --git a/tests/gocase/unit/type/zset/zset_test.go b/tests/gocase/unit/type/zset/zset_test.go
index d4a7a83f9..774155a02 100644
--- a/tests/gocase/unit/type/zset/zset_test.go
+++ b/tests/gocase/unit/type/zset/zset_test.go
@@ -1661,13 +1661,16 @@ func basicTests(t *testing.T, rdb *redis.Client, ctx context.Context, enabledRES
func stressTests(t *testing.T, rdb *redis.Client, ctx context.Context, encoding string) {
var elements int
- if encoding == "ziplist" {
+ switch encoding {
+ case "ziplist":
elements = 128
- } else if encoding == "skiplist" {
+ case "skiplist":
elements = 100
- } else {
- fmt.Println("Unknown sorted set encoding")
- return
+ default:
+ {
+ fmt.Println("Unknown sorted set encoding")
+ return
+ }
}
t.Run(fmt.Sprintf("ZSCORE - %s", encoding), func(t *testing.T) {
rdb.Del(ctx, "zscoretest")
diff --git a/tests/gocase/util/server.go b/tests/gocase/util/server.go
index 2e020c15e..ea8ee4742 100644
--- a/tests/gocase/util/server.go
+++ b/tests/gocase/util/server.go
@@ -244,7 +244,7 @@ func StartServerWithCLIOptions(
defer func() { require.NoError(t, f.Close()) }()
for k := range configs {
- _, err := f.WriteString(fmt.Sprintf("%s %s\n", k, configs[k]))
+ _, err := fmt.Fprintf(f, "%s %s\n", k, configs[k])
require.NoError(t, err)
}
cmd.Args = append(cmd.Args, "-c", f.Name())
diff --git a/x.py b/x.py
index 9caf09f09..c7c690ec5 100755
--- a/x.py
+++ b/x.py
@@ -31,7 +31,7 @@ from shutil import which
CMAKE_REQUIRE_VERSION = (3, 16, 0)
CLANG_FORMAT_REQUIRED_VERSION = (12, 0, 0)
CLANG_TIDY_REQUIRED_VERSION = (12, 0, 0)
-GOLANGCI_LINT_REQUIRED_VERSION = (1, 64, 3)
+GOLANGCI_LINT_REQUIRED_VERSION = (2, 2, 2)
SEMVER_REGEX = re.compile(
r"""
(kvrocks) 01/10: chore(common): refactor parse utils via std::from_chars (#3052)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 4ffbed4e912561ef1955b490ec0ad134cf253714
Author: Twice <tw...@gmail.com>
AuthorDate: Tue Jul 15 14:49:50 2025 +0800
chore(common): refactor parse utils via std::from_chars (#3052)
Co-authored-by: hulk <hu...@gmail.com>
---
src/common/parse_util.cc | 9 +--
src/common/parse_util.h | 100 +++++--------------------
tests/cppunit/parse_util.cc | 4 +-
tests/gocase/unit/type/hash/hash_test.go | 2 +-
tests/gocase/unit/type/strings/strings_test.go | 2 +-
5 files changed, 26 insertions(+), 91 deletions(-)
diff --git a/src/common/parse_util.cc b/src/common/parse_util.cc
index a7359c78a..9cb9790b3 100644
--- a/src/common/parse_util.cc
+++ b/src/common/parse_util.cc
@@ -20,14 +20,13 @@
#include "parse_util.h"
-#include <limits>
-
#include "bit_util.h"
+#include "string_util.h"
-StatusOr<std::uint64_t> ParseSizeAndUnit(const std::string &v) {
- auto [num, rest] = GET_OR_RET(TryParseInt<std::uint64_t>(v.c_str(), 10));
+StatusOr<std::uint64_t> ParseSizeAndUnit(std::string_view v) {
+ auto [num, rest] = GET_OR_RET(TryParseInt<std::uint64_t>(v, 10));
- if (*rest == 0) {
+ if (rest == v.data() + v.size()) {
return num;
} else if (util::EqualICase(rest, "k")) {
return util::CheckedShiftLeft(num, 10);
diff --git a/src/common/parse_util.h b/src/common/parse_util.h
index 3e3cf116c..201f9415a 100644
--- a/src/common/parse_util.h
+++ b/src/common/parse_util.h
@@ -20,70 +20,11 @@
#pragma once
-#include <cstdlib>
-#include <limits>
-#include <string>
+#include <charconv>
#include <tuple>
+#include <type_traits>
#include "status.h"
-#include "string_util.h"
-
-namespace details {
-
-template <typename>
-struct ParseIntFunc;
-
-template <>
-struct ParseIntFunc<char> { // NOLINT
- constexpr static const auto value = std::strtol;
-};
-
-template <>
-struct ParseIntFunc<short> { // NOLINT
- constexpr static const auto value = std::strtol;
-};
-
-template <>
-struct ParseIntFunc<int> { // NOLINT
- constexpr static const auto value = std::strtol;
-};
-
-template <>
-struct ParseIntFunc<long> { // NOLINT
- constexpr static const auto value = std::strtol;
-};
-
-template <>
-struct ParseIntFunc<long long> { // NOLINT
- constexpr static const auto value = std::strtoll;
-};
-
-template <>
-struct ParseIntFunc<unsigned char> { // NOLINT
- constexpr static const auto value = std::strtoul;
-};
-
-template <>
-struct ParseIntFunc<unsigned short> { // NOLINT
- constexpr static const auto value = std::strtoul;
-};
-
-template <>
-struct ParseIntFunc<unsigned> { // NOLINT
- constexpr static const auto value = std::strtoul;
-};
-
-template <>
-struct ParseIntFunc<unsigned long> { // NOLINT
- constexpr static const auto value = std::strtoul;
-};
-
-template <>
-struct ParseIntFunc<unsigned long long> { // NOLINT
- constexpr static const auto value = std::strtoull;
-};
-
-} // namespace details
template <typename T>
using ParseResultAndPos = std::tuple<T, const char *>;
@@ -94,24 +35,20 @@ using ParseResultAndPos = std::tuple<T, const char *>;
// e.g. TryParseInt("100MB") -> {100, "MB"}
// if no integer can be parsed or out of type range, an error will be returned
// base can be in {0, 2, ..., 36}, refer to strto* in standard c for more details
-template <typename T = long long> // NOLINT
-StatusOr<ParseResultAndPos<T>> TryParseInt(const char *v, int base = 0) {
- char *end = nullptr;
-
- errno = 0;
- auto res = details::ParseIntFunc<T>::value(v, &end, base);
+template <typename T = long long, std::enable_if_t<std::is_integral_v<T>, int> = 0> // NOLINT
+StatusOr<ParseResultAndPos<T>> TryParseInt(std::string_view v, int base = 10) {
+ T res = 0;
+ auto [end, ec] = std::from_chars(v.data(), v.data() + v.size(), res, base);
- if (v == end) {
+ if (v.data() == end) {
return {Status::NotOK, "not started as an integer"};
}
- if (errno) {
- return Status::FromErrno();
- }
-
- if (!std::is_same<T, decltype(res)>::value &&
- (res < std::numeric_limits<T>::min() || res > std::numeric_limits<T>::max())) {
- return {Status::NotOK, "out of range of integer type"};
+ if (ec != std::errc()) {
+ if (ec == std::errc::result_out_of_range) {
+ return {Status::NotOK, "out of range of integer type"};
+ }
+ return {Status::NotOK, std::make_error_code(ec).message()};
}
return ParseResultAndPos<T>{res, end};
@@ -121,13 +58,12 @@ StatusOr<ParseResultAndPos<T>> TryParseInt(const char *v, int base = 0) {
// not like TryParseInt, the whole string need to be parsed as an integer,
// e.g. ParseInt("100MB") -> error status
template <typename T = long long> // NOLINT
-StatusOr<T> ParseInt(const std::string &v, int base = 0) {
- const char *begin = v.c_str();
- auto res = TryParseInt<T>(begin, base);
+StatusOr<T> ParseInt(std::string_view v, int base = 10) {
+ auto res = TryParseInt<T>(v, base);
if (!res) return res;
- if (std::get<1>(*res) != begin + v.size()) {
+ if (std::get<1>(*res) != v.data() + v.size()) {
return {Status::NotOK, "encounter non-integer characters"};
}
@@ -140,7 +76,7 @@ using NumericRange = std::tuple<T, T>;
// this overload accepts a range {min, max},
// integer out of the range will trigger an error status
template <typename T = long long> // NOLINT
-StatusOr<T> ParseInt(const std::string &v, NumericRange<T> range, int base = 0) {
+StatusOr<T> ParseInt(std::string_view v, NumericRange<T> range, int base = 10) {
auto res = ParseInt<T>(v, base);
if (!res) return res;
@@ -153,8 +89,10 @@ StatusOr<T> ParseInt(const std::string &v, NumericRange<T> range, int base = 0)
}
// available units: K, M, G, T, P
-StatusOr<std::uint64_t> ParseSizeAndUnit(const std::string &v);
+StatusOr<std::uint64_t> ParseSizeAndUnit(std::string_view v);
+// we cannot use std::from_chars for floating-point numbers,
+// since it is available since gcc/libstdc++ 11 and libc++ 20.
template <typename>
struct ParseFloatFunc;
diff --git a/tests/cppunit/parse_util.cc b/tests/cppunit/parse_util.cc
index d0364b7b9..3dd00de64 100644
--- a/tests/cppunit/parse_util.cc
+++ b/tests/cppunit/parse_util.cc
@@ -31,13 +31,10 @@ TEST(ParseUtil, TryParseInt) {
ASSERT_EQ(TryParseInt("hello").Msg(), "not started as an integer");
ASSERT_FALSE(TryParseInt("9999999999999999999999999999999999"));
- ASSERT_FALSE(TryParseInt("1", 100));
}
TEST(ParseUtil, ParseInt) {
ASSERT_EQ(*ParseInt("-2333"), -2333);
- ASSERT_EQ(*ParseInt("0x1a"), 26);
- ASSERT_EQ(*ParseInt("011"), 9);
ASSERT_EQ(*ParseInt("111", 2), 7);
ASSERT_EQ(*ParseInt("11", 010), 9);
ASSERT_EQ(*ParseInt("11", 10), 11);
@@ -50,6 +47,7 @@ TEST(ParseUtil, ParseInt) {
ASSERT_EQ(*ParseInt<short>("30000"), 30000);
ASSERT_EQ(*ParseInt<int>("99999"), 99999);
ASSERT_EQ(ParseInt<int>("3000000000").Msg(), "out of range of integer type");
+ ASSERT_EQ(ParseInt<unsigned>("-1").Msg(), "not started as an integer");
ASSERT_EQ(*ParseInt("123", {0, 123}), 123);
ASSERT_EQ(ParseInt("124", {0, 123}).Msg(), "out of numeric range");
diff --git a/tests/gocase/unit/type/hash/hash_test.go b/tests/gocase/unit/type/hash/hash_test.go
index 29db4747d..8d1536796 100644
--- a/tests/gocase/unit/type/hash/hash_test.go
+++ b/tests/gocase/unit/type/hash/hash_test.go
@@ -631,7 +631,7 @@ var testHash = func(t *testing.T, configs util.KvrocksServerConfigs) {
t.Run("HINCRBY fails against hash value with spaces (left)", func(t *testing.T) {
rdb.HSet(ctx, "samllhash", "str", " 11")
rdb.HSet(ctx, "bighash", "str", " 11")
- pattern := "ERR.*not an integer.*"
+ pattern := "ERR.*not.*an integer.*"
util.ErrorRegexp(t, rdb.HIncrBy(ctx, "samllhash", "str", 1).Err(), pattern)
util.ErrorRegexp(t, rdb.HIncrBy(ctx, "bighash", "str", 1).Err(), pattern)
})
diff --git a/tests/gocase/unit/type/strings/strings_test.go b/tests/gocase/unit/type/strings/strings_test.go
index 305ab3975..56b67cb00 100644
--- a/tests/gocase/unit/type/strings/strings_test.go
+++ b/tests/gocase/unit/type/strings/strings_test.go
@@ -378,7 +378,7 @@ func testString(t *testing.T, configs util.KvrocksServerConfigs) {
t.Run("SETBIT with out of range bit offset", func(t *testing.T) {
require.NoError(t, rdb.Del(ctx, "mykey").Err())
require.ErrorContains(t, rdb.SetBit(ctx, "mykey", 4*1024*1024*1024+2, 1).Err(), "out of range")
- require.ErrorContains(t, rdb.SetBit(ctx, "mykey", -1, 1).Err(), "out of range")
+ require.ErrorContains(t, rdb.SetBit(ctx, "mykey", -1, 1).Err(), "an integer")
})
t.Run("SETBIT with non-bit argument", func(t *testing.T) {
(kvrocks) 04/10: feat(tdigest): add `TDIGEST.MERGE` command implementation (#3054)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 65f6027b247b4d370c85418093c7ae93502a7b80
Author: Edward Xu <xu...@gmail.com>
AuthorDate: Thu Jul 17 16:42:38 2025 +0800
feat(tdigest): add `TDIGEST.MERGE` command implementation (#3054)
Co-authored-by: Twice <tw...@gmail.com>
---
src/commands/cmd_tdigest.cc | 86 +++++++++++++++-
src/commands/error_constants.h | 3 +
src/types/redis_tdigest.cc | 137 ++++++++++++++++++++++++-
src/types/redis_tdigest.h | 12 ++-
src/types/tdigest.cc | 50 +++++++--
src/types/tdigest.h | 4 +-
tests/gocase/unit/type/tdigest/tdigest_test.go | 115 +++++++++++++++++++--
7 files changed, 386 insertions(+), 21 deletions(-)
diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc
index 4c04e8619..e7815256d 100644
--- a/src/commands/cmd_tdigest.cc
+++ b/src/commands/cmd_tdigest.cc
@@ -23,6 +23,8 @@
#include "command_parser.h"
#include "commander.h"
+#include "commands/error_constants.h"
+#include "parse_util.h"
#include "server/redis_reply.h"
#include "server/server.h"
#include "status.h"
@@ -32,6 +34,7 @@
namespace redis {
namespace {
constexpr auto kCompressionArg = "compression";
+constexpr auto kOverrideArg = "override";
constexpr auto kInfoCompression = "Compression";
constexpr auto kInfoCapacity = "Capacity";
@@ -281,11 +284,92 @@ class CommandTDigestQuantile : public Commander {
std::string key_name_;
std::vector<double> quantiles_;
};
+
+class CommandTDigestMerge : public Commander {
+ Status Parse(const std::vector<std::string> &args) override {
+ CommandParser parser(args, 1);
+ dest_key_ = GET_OR_RET(parser.TakeStr());
+ auto numkeys = parser.TakeInt();
+ if (!numkeys) {
+ return {Status::RedisParseErr, errParsingNumkeys};
+ }
+
+ if (*numkeys <= 0) {
+ return {Status::RedisParseErr, errNumkeysMustBePositive};
+ }
+
+ if (static_cast<int64_t>(args.size()) < (3 + *numkeys)) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ source_keys_.reserve(*numkeys);
+
+ for (auto i = 3; i < (3 + *numkeys); i++) {
+ auto src_digest = GET_OR_RET(parser.TakeStr());
+ source_keys_.emplace_back(std::move(src_digest));
+ }
+
+ while (parser.Good()) {
+ // more arguments than expected compression and override
+ if (options_.compression > 0 && options_.override_flag) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ if (parser.EatEqICase(kCompressionArg)) {
+ // compression already set or without a compression value
+ if (options_.compression > 0 || !parser.Good()) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ if (auto compression = parser.TakeInt<uint32_t>(); !compression) {
+ return {Status::RedisParseErr, errParseCompression};
+ } else if (*compression <= 0 || *compression > kTDigestMaxCompression) {
+ return {Status::RedisParseErr, errCompressionOutOfRange};
+ } else {
+ options_.compression = *compression;
+ }
+ }
+
+ if (parser.EatEqICase(kOverrideArg)) {
+ if (options_.override_flag) { // override already set
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+ options_.override_flag = true;
+ } else {
+ return {Status::RedisParseErr, errWrongKeyword};
+ }
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+ TDigest tdigest(srv->storage, conn->GetNamespace());
+ auto s = tdigest.Merge(ctx, dest_key_, source_keys_, options_);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.IsNotFound() ? errKeyNotFound : s.ToString()};
+ }
+ *output = redis::RESP_OK;
+ return Status::OK();
+ }
+
+ private:
+ std::string dest_key_;
+ std::vector<std::string> source_keys_;
+ TDigestMergeOptions options_;
+};
+
+std::vector<CommandKeyRange> GetMergeKeyRange(const std::vector<std::string> &args) {
+ auto numkeys = ParseInt<int>(args[2], 10).ValueOr(0);
+ return {{1, 1, 1}, {3, 2 + numkeys, 1}};
+}
+
REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.create", -2, "write", 1, 1, 1),
MakeCmdAttr<CommandTDigestInfo>("tdigest.info", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestAdd>("tdigest.add", -3, "write", 1, 1, 1),
MakeCmdAttr<CommandTDigestMax>("tdigest.max", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
MakeCmdAttr<CommandTDigestQuantile>("tdigest.quantile", -3, "read-only", 1, 1, 1),
- MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1));
+ MakeCmdAttr<CommandTDigestReset>("tdigest.reset", 2, "write", 1, 1, 1),
+ MakeCmdAttr<CommandTDigestMerge>("tdigest.merge", -4, "write", GetMergeKeyRange));
} // namespace redis
diff --git a/src/commands/error_constants.h b/src/commands/error_constants.h
index 26b2e6921..71235b6cc 100644
--- a/src/commands/error_constants.h
+++ b/src/commands/error_constants.h
@@ -50,4 +50,7 @@ inline constexpr const char *errCompressionMustBePositive = "compression paramet
inline constexpr const char *errCompressionOutOfRange = "compression must be between 1 and 1000";
inline constexpr const char *errKeyNotFound = "key does not exist";
inline constexpr const char *errKeyAlreadyExists = "key already exists";
+inline constexpr const char *errParsingNumkeys = "error parsing numkeys";
+inline constexpr const char *errNumkeysMustBePositive = "numkeys need to be a positive integer";
+inline constexpr const char *errWrongKeyword = "wrong keyword";
} // namespace redis
diff --git a/src/types/redis_tdigest.cc b/src/types/redis_tdigest.cc
index 5f112a410..f506ad92e 100644
--- a/src/types/redis_tdigest.cc
+++ b/src/types/redis_tdigest.cc
@@ -37,6 +37,7 @@
#include <range/v3/view/transform.hpp>
#include <vector>
+#include "commands/error_constants.h"
#include "db_util.h"
#include "encoding.h"
#include "status.h"
@@ -246,6 +247,7 @@ rocksdb::Status TDigest::Quantile(engine::Context& ctx, const Slice& digest_name
return rocksdb::Status::OK();
}
+
rocksdb::Status TDigest::Reset(engine::Context& ctx, const Slice& digest_name) {
auto ns_key = AppendNamespacePrefix(digest_name);
@@ -285,6 +287,134 @@ rocksdb::Status TDigest::Reset(engine::Context& ctx, const Slice& digest_name) {
auto status = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
return status;
}
+
+rocksdb::Status TDigest::Merge(engine::Context& ctx, const Slice& dest_digest,
+ const std::vector<std::string>& source_digests, const TDigestMergeOptions& options) {
+ if (options.compression != 0 && options.compression > kTDigestMaxCompression) {
+ return rocksdb::Status::InvalidArgument(fmt::format("compression should be less than {}", kTDigestMaxCompression));
+ }
+
+ auto dest_ns_key = AppendNamespacePrefix(dest_digest);
+
+ bool dest_digest_existed = false;
+ TDigestMetadata dest_metadata;
+ if (auto status = getMetaDataByNsKey(ctx, dest_ns_key, &dest_metadata); !status.ok() && !status.IsNotFound()) {
+ return status;
+ } else if (status.ok()) {
+ dest_digest_existed = true;
+ if (!options.override_flag) {
+ return rocksdb::Status::InvalidArgument(fmt::format("{}: {}", errKeyAlreadyExists, dest_digest.ToString()));
+ }
+ }
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisTDigest);
+ if (auto status = batch->PutLogData(log_data.Encode()); !status.ok()) {
+ return status;
+ }
+
+ uint32_t compression = 0;
+ uint64_t total_observations = 0;
+ std::vector<CentroidsWithDelta> source_centroids_data;
+ source_centroids_data.reserve(source_digests.size());
+ // use map to avoid duplicate processing of the same tdigest
+ std::map<std::string, const CentroidsWithDelta*> unique_source_centroids;
+ for (const auto& tdigest : source_digests) {
+ if (auto it = unique_source_centroids.find(tdigest); it != unique_source_centroids.end()) {
+ // skip if the tdigest has been processed
+ if (it->second != nullptr) { // only store non-empty centroids
+ source_centroids_data.emplace_back(*it->second);
+ }
+ continue;
+ }
+
+ TDigestMetadata metadata;
+ std::vector<Centroid> source_centroids;
+ auto source_ns_key = AppendNamespacePrefix(tdigest);
+ if (auto status = getMetaDataByNsKey(ctx, source_ns_key, &metadata); !status.ok()) {
+ if (status.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument(fmt::format("{}: {}", errKeyNotFound, tdigest));
+ }
+ return status;
+ }
+
+ if (metadata.unmerged_nodes > 0) {
+ if (auto status = mergeCurrentBuffer(ctx, source_ns_key, batch, &metadata, nullptr, &source_centroids);
+ !status.ok()) {
+ return status;
+ }
+
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ if (auto status = batch->Put(metadata_cf_handle_, source_ns_key, metadata_bytes); !status.ok()) {
+ return status;
+ }
+ } else if (metadata.merged_nodes > 0) {
+ if (auto status = dumpCentroids(ctx, source_ns_key, metadata, &source_centroids); !status.ok()) {
+ return status;
+ }
+ }
+
+ if (!source_centroids.empty()) {
+ unique_source_centroids[tdigest] = &source_centroids_data.emplace_back(CentroidsWithDelta{
+ .centroids = std::move(source_centroids),
+ .delta = metadata.compression,
+ .min = metadata.minimum,
+ .max = metadata.maximum,
+ .total_weight = static_cast<double>(metadata.merged_weight),
+ });
+ } else {
+ unique_source_centroids[tdigest] = nullptr; // use nullptr as a marker for empty centroids
+ }
+
+ total_observations += metadata.total_observations;
+ compression = std::max(compression, metadata.compression);
+ }
+
+ if (options.compression != 0) {
+ compression = options.compression;
+ }
+
+ auto merged_data = TDigestMerge(source_centroids_data, compression);
+ if (!merged_data.IsOK()) {
+ return rocksdb::Status::InvalidArgument(merged_data.Msg());
+ }
+
+ if (dest_digest_existed) {
+ auto start_key = internalSegmentGuardPrefixKey(dest_metadata, dest_ns_key, SegmentType::kBuffer);
+ auto guard_key = internalSegmentGuardPrefixKey(dest_metadata, dest_ns_key, SegmentType::kGuardFlag);
+
+ if (auto status = batch->DeleteRange(cf_handle_, start_key, guard_key); !status.ok()) {
+ return status;
+ }
+ }
+
+ auto capacity = compression * 6 + 10;
+ capacity = ((capacity < kMaxElements) ? capacity : kMaxElements);
+ dest_metadata.compression = compression;
+ dest_metadata.capacity = capacity;
+ dest_metadata.unmerged_nodes = 0;
+ dest_metadata.merged_nodes = merged_data->centroids.size();
+ dest_metadata.total_weight = static_cast<uint64_t>(merged_data->total_weight);
+ dest_metadata.merged_weight = static_cast<uint64_t>(merged_data->total_weight);
+ dest_metadata.minimum = merged_data->min;
+ dest_metadata.maximum = merged_data->max;
+ dest_metadata.merge_times = 0;
+ dest_metadata.total_observations = total_observations;
+
+ std::string metadata_bytes;
+ dest_metadata.Encode(&metadata_bytes);
+ if (auto status = batch->Put(metadata_cf_handle_, dest_ns_key, metadata_bytes); !status.ok()) {
+ return status;
+ }
+
+ if (auto status = applyNewCentroids(batch, dest_ns_key, dest_metadata, merged_data->centroids); !status.ok()) {
+ return status;
+ }
+
+ return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
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);
@@ -296,7 +426,8 @@ rocksdb::Status TDigest::getMetaDataByNsKey(engine::Context& context, const Slic
rocksdb::Status TDigest::mergeCurrentBuffer(engine::Context& ctx, const std::string& ns_key,
ObserverOrUniquePtr<rocksdb::WriteBatchBase>& batch,
- TDigestMetadata* metadata, const std::vector<double>* additional_buffer) {
+ TDigestMetadata* metadata, const std::vector<double>* additional_buffer,
+ std::vector<Centroid>* dump_centroids) {
std::vector<Centroid> centroids;
std::vector<double> buffer;
centroids.reserve(metadata->merged_nodes);
@@ -330,6 +461,10 @@ rocksdb::Status TDigest::mergeCurrentBuffer(engine::Context& ctx, const std::str
metadata->unmerged_nodes = 0;
metadata->merged_weight = static_cast<uint64_t>(merged_centroids->total_weight);
+ if (dump_centroids != nullptr) {
+ *dump_centroids = std::move(merged_centroids->centroids);
+ }
+
return rocksdb::Status::OK();
}
diff --git a/src/types/redis_tdigest.h b/src/types/redis_tdigest.h
index b33a0387e..2026cf94d 100644
--- a/src/types/redis_tdigest.h
+++ b/src/types/redis_tdigest.h
@@ -44,6 +44,11 @@ struct TDigestCreateOptions {
uint32_t compression;
};
+struct TDigestMergeOptions {
+ uint32_t compression = 0;
+ bool override_flag = false;
+};
+
struct TDigestQuantitleResult {
std::optional<std::vector<double>> quantiles;
};
@@ -69,6 +74,10 @@ class TDigest : public SubKeyScanner {
TDigestQuantitleResult* result);
rocksdb::Status Reset(engine::Context& ctx, const Slice& digest_name);
+
+ rocksdb::Status Merge(engine::Context& ctx, const Slice& dest_digest, const std::vector<std::string>& source_digests,
+ const TDigestMergeOptions& options);
+
rocksdb::Status GetMetaData(engine::Context& context, const Slice& digest_name, TDigestMetadata* metadata);
private:
@@ -110,7 +119,8 @@ class TDigest : public SubKeyScanner {
rocksdb::Status mergeCurrentBuffer(engine::Context& ctx, const std::string& ns_key,
ObserverOrUniquePtr<rocksdb::WriteBatchBase>& batch, TDigestMetadata* metadata,
- const std::vector<double>* additional_buffer = nullptr);
+ const std::vector<double>* additional_buffer = nullptr,
+ std::vector<Centroid>* dump_centroids = nullptr);
std::string internalBufferKey(const std::string& ns_key, const TDigestMetadata& metadata) const;
std::string internalKeyFromCentroid(const std::string& ns_key, const TDigestMetadata& metadata,
const Centroid& centroid, uint32_t seq) const;
diff --git a/src/types/tdigest.cc b/src/types/tdigest.cc
index 3c6080975..afbc45ab1 100644
--- a/src/types/tdigest.cc
+++ b/src/types/tdigest.cc
@@ -32,7 +32,6 @@ refer to https://github.com/apache/arrow/blob/27bbd593625122a4a25d9471c8aaf5df54
#include <queue>
#include "common/status.h"
-#include "logging.h"
namespace {
// scale function K1
@@ -397,30 +396,59 @@ CentroidsWithDelta TDigest::DumpCentroids() const {
void TDigest::Add(const std::vector<double>& items) { impl_.MergeInput(items); }
-StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<CentroidsWithDelta>& centroids_list) {
+StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<CentroidsWithDelta>& centroids_list, uint64_t delta) {
if (centroids_list.empty()) {
- return Status{Status::InvalidArgument, "centroids_list is empty"};
+ return CentroidsWithDelta{.delta = delta};
}
if (centroids_list.size() == 1) {
- return centroids_list.front();
+ if (centroids_list.front().delta == delta) {
+ return centroids_list.front();
+ }
+ if (centroids_list.front().centroids.empty()) {
+ return CentroidsWithDelta{.delta = delta};
+ }
}
- TDigest digest{centroids_list.front().delta};
- digest.Reset(centroids_list.front());
+ TDigest digest{delta};
+ std::vector<TDigest> others;
+ others.reserve(centroids_list.size());
+ for (const auto& centroids : centroids_list) {
+ if (centroids.centroids.empty()) {
+ continue; // skip empty centroids
+ }
+ TDigest d{centroids.delta};
+ d.Reset(centroids);
+ others.emplace_back(std::move(d));
+ }
+
+ digest.Merge(others);
+
+ return digest.DumpCentroids();
+}
+
+StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<double>& buffer,
+ const std::vector<CentroidsWithDelta>& centroids_lists, uint64_t delta) {
+ TDigest digest{delta};
+
+ digest.Reset(CentroidsWithDelta{});
+ digest.Add(buffer);
std::vector<TDigest> others;
- others.reserve(centroids_list.size() - 1);
+ others.reserve(centroids_lists.size());
- for (size_t i = 1; i < centroids_list.size(); ++i) {
- TDigest d{centroids_list[i].delta};
- digest.Reset(centroids_list[i]);
+ for (const auto& centroids : centroids_lists) {
+ if (centroids.centroids.empty()) {
+ continue; // skip empty centroids
+ }
+ TDigest d{centroids.delta};
+ d.Reset(centroids);
others.emplace_back(std::move(d));
}
-
digest.Merge(others);
return digest.DumpCentroids();
}
+
StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<double>& buffer, const CentroidsWithDelta& centroid_list) {
TDigest digest{centroid_list.delta};
digest.Reset(centroid_list);
diff --git a/src/types/tdigest.h b/src/types/tdigest.h
index 0f088cb0f..1f416b48e 100644
--- a/src/types/tdigest.h
+++ b/src/types/tdigest.h
@@ -50,7 +50,9 @@ struct CentroidsWithDelta {
double total_weight;
};
-StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<CentroidsWithDelta>& centroids_list);
+StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<CentroidsWithDelta>& centroids_list, uint64_t delta);
+StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<double>& buffer,
+ const std::vector<CentroidsWithDelta>& centroids_lists, uint64_t delta);
StatusOr<CentroidsWithDelta> TDigestMerge(const std::vector<double>& buffer, const CentroidsWithDelta& centroid_list);
/**
diff --git a/tests/gocase/unit/type/tdigest/tdigest_test.go b/tests/gocase/unit/type/tdigest/tdigest_test.go
index 62efacb9e..3c2565ac0 100644
--- a/tests/gocase/unit/type/tdigest/tdigest_test.go
+++ b/tests/gocase/unit/type/tdigest/tdigest_test.go
@@ -31,12 +31,14 @@ import (
)
const (
- errMsgWrongNumberArg = "wrong number of arguments"
- errMsgParseCompression = "error parsing compression parameter"
- errMsgNeedToBePositive = "compression parameter needs to be a positive integer"
- errMsgMustInRange = "compression must be between 1 and 1000"
- errMsgKeyAlreadyExists = "key already exists"
- errMsgKeyNotExist = "key does not exist"
+ errMsgWrongNumberArg = "wrong number of arguments"
+ errMsgParseCompression = "error parsing compression parameter"
+ errMsgNeedToBePositive = "compression parameter needs to be a positive integer"
+ errMsgMustInRange = "compression must be between 1 and 1000"
+ errMsgKeyAlreadyExists = "key already exists"
+ errMsgKeyNotExist = "key does not exist"
+ errNumkeysMustBePositive = "numkeys need to be a positive integer"
+ errCompressionParameterMustBePositive = "compression parameter needs to be a positive integer"
)
type tdigestInfo struct {
@@ -415,4 +417,105 @@ func tdigestTests(t *testing.T, configs util.KvrocksServerConfigs) {
}
}
})
+
+ t.Run("tdigest.merge with different arguments", func(t *testing.T) {
+ keyPrefix := "tdigest_merge_"
+
+ // no arguments
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE").Err(), errMsgWrongNumberArg)
+
+ // merge with no source keys
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key1").Err(), errMsgWrongNumberArg)
+
+ // merge with invalid number of source keys
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key2", "hahah").Err(), errMsgWrongNumberArg)
+
+ // merge with not matching number of source keys
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key3", 3, "hahah").Err(), errMsgWrongNumberArg)
+
+ // merge with negative number of source keys
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key4", -1, "hahah").Err(), errNumkeysMustBePositive)
+
+ // merge with non-existent source key
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key5", 1, keyPrefix+"nonexistent").Err(), errMsgKeyNotExist)
+
+ // merge with invalid compression keyword
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key6", 1, keyPrefix+"nonexistent", "compression").Err(), errMsgWrongNumberArg)
+
+ // merge with invalid compression value
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key7", 1, keyPrefix+"nonexistent", "compression", "hahah").Err(), errMsgParseCompression)
+
+ // merge with more than one override
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key8", 1, keyPrefix+"nonexistent", "compression", "100", "override", "override").Err(), errMsgWrongNumberArg)
+
+ // create a source digest and add some data
+ sourceKey1 := keyPrefix + "source1"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", sourceKey1, "compression", "101").Err())
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", sourceKey1, "1.0", "2.0", "3.0").Err())
+
+ sourceKey2 := keyPrefix + "source2"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", sourceKey2, "compression", "30").Err())
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", sourceKey2, "4.0", "5.0", "6.0", "100", "-200").Err())
+
+ // create a destination digest
+ destKey := keyPrefix + "dest"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", destKey, "compression", "100").Err())
+
+ // merge the source into the destination without override
+ require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", destKey, 2, sourceKey1, sourceKey2).Err(), errMsgKeyAlreadyExists)
+
+ // merge the source into the destination with override
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", destKey, 2, sourceKey1, sourceKey2, "override").Err())
+
+ // merge to a new destination key
+ newDestKey1 := keyPrefix + "new_dest"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", newDestKey1, 2, sourceKey1, sourceKey2).Err())
+
+ // merge with same source keys
+ newDestKey2 := keyPrefix + "new_dest2"
+ require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", newDestKey2, 4, sourceKey1, sourceKey2, sourceKey1, sourceKey2).Err())
+
+ validation := func(destMergeKey string) {
+ rsp := rdb.Do(ctx, "TDIGEST.INFO", destMergeKey)
+ require.NoError(t, rsp.Err())
+ info := toTdigestInfo(t, rsp.Val())
+ require.EqualValues(t, 101, info.Compression)
+ require.EqualValues(t, 8, info.Observations)
+
+ rsp = rdb.Do(ctx, "TDIGEST.MAX", destMergeKey)
+ require.NoError(t, rsp.Err())
+ {
+ rspval, err := rsp.Float64()
+ require.NoError(t, err)
+ require.InEpsilon(t, 100, rspval, 0.001)
+ }
+
+ rsp = rdb.Do(ctx, "TDIGEST.MIN", destMergeKey)
+ require.NoError(t, rsp.Err())
+ {
+ rspval, err := rsp.Float64()
+ require.NoError(t, err)
+ require.InEpsilon(t, -200, rspval, 0.001)
+ }
+
+ rsp = rdb.Do(ctx, "TDIGEST.QUANTILE", destMergeKey, "0.1", "0.5", "0.75", "0.9", "0.99", "1")
+ require.NoError(t, rsp.Err())
+ vals, err := rsp.Slice()
+ require.NoError(t, err)
+ require.Len(t, vals, 6)
+ expected := []float64{-200.0, 4.0, 6.0, 100.0, 100.0, 100.0}
+ for i, v := range vals {
+ str, ok := v.(string)
+ require.True(t, ok, "expected string but got %T at index %d", v, i)
+
+ got, err := strconv.ParseFloat(str, 64)
+ require.NoError(t, err, "could not parse value at index %d", i)
+
+ require.InEpsilon(t, expected[i], got, 0.2, "mismatch at index %d", i)
+ }
+ }
+ validation(destKey)
+ validation(newDestKey1)
+ validation(newDestKey2)
+ })
}
(kvrocks) 09/10: Comment out the tdigest and its test cases due to it's not production ready
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 65681ebf8053ac49a93a28f41ad5c01fbc3fde2f
Author: edward_xu <xu...@gmail.com>
AuthorDate: Mon Aug 4 20:38:57 2025 +0800
Comment out the tdigest and its test cases due to it's not production ready
---
src/commands/cmd_tdigest.cc | 16 +-
tests/gocase/unit/type/tdigest/tdigest_test.go | 521 -------------------------
2 files changed, 8 insertions(+), 529 deletions(-)
diff --git a/src/commands/cmd_tdigest.cc b/src/commands/cmd_tdigest.cc
index e7815256d..b780586d9 100644
--- a/src/commands/cmd_tdigest.cc
+++ b/src/commands/cmd_tdigest.cc
@@ -364,12 +364,12 @@ std::vector<CommandKeyRange> GetMergeKeyRange(const std::vector<std::string> &ar
return {{1, 1, 1}, {3, 2 + numkeys, 1}};
}
-REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.create", -2, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestInfo>("tdigest.info", 2, "read-only", 1, 1, 1),
- MakeCmdAttr<CommandTDigestAdd>("tdigest.add", -3, "write", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMax>("tdigest.max", 2, "read-only", 1, 1, 1),
- MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
- MakeCmdAttr<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));
+// REDIS_REGISTER_COMMANDS(TDigest, MakeCmdAttr<CommandTDigestCreate>("tdigest.create", -2, "write", 1, 1, 1),
+// MakeCmdAttr<CommandTDigestInfo>("tdigest.info", 2, "read-only", 1, 1, 1),
+// MakeCmdAttr<CommandTDigestAdd>("tdigest.add", -3, "write", 1, 1, 1),
+// MakeCmdAttr<CommandTDigestMax>("tdigest.max", 2, "read-only", 1, 1, 1),
+// MakeCmdAttr<CommandTDigestMin>("tdigest.min", 2, "read-only", 1, 1, 1),
+// MakeCmdAttr<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));
} // namespace redis
diff --git a/tests/gocase/unit/type/tdigest/tdigest_test.go b/tests/gocase/unit/type/tdigest/tdigest_test.go
deleted file mode 100644
index 3c2565ac0..000000000
--- a/tests/gocase/unit/type/tdigest/tdigest_test.go
+++ /dev/null
@@ -1,521 +0,0 @@
-//go:build !ignore_when_tsan
-
-/*
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements. See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership. The ASF licenses this file
-* to you under the Apache License, Version 2.0 (the
-* "License"); you may not use this file except in compliance
-* with the License. You may obtain a copy of the License at
-*
-* http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing,
-* software distributed under the License is distributed on an
-* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-* KIND, either express or implied. See the License for the
-* specific language governing permissions and limitations
-* under the License.
- */
-
-package tdigest
-
-import (
- "context"
- "strconv"
- "testing"
-
- "github.com/apache/kvrocks/tests/gocase/util"
- "github.com/stretchr/testify/require"
-)
-
-const (
- errMsgWrongNumberArg = "wrong number of arguments"
- errMsgParseCompression = "error parsing compression parameter"
- errMsgNeedToBePositive = "compression parameter needs to be a positive integer"
- errMsgMustInRange = "compression must be between 1 and 1000"
- errMsgKeyAlreadyExists = "key already exists"
- errMsgKeyNotExist = "key does not exist"
- errNumkeysMustBePositive = "numkeys need to be a positive integer"
- errCompressionParameterMustBePositive = "compression parameter needs to be a positive integer"
-)
-
-type tdigestInfo struct {
- Compression int64
- Capacity int64
- MergedNodes int64
- UnmergedNodes int64
- MergedWeight int64
- UnmergedWeight int64
- Observations int64
- TotalCompressions int64
- // memory usgae is not useful, we do not support it now
-}
-
-func toTdigestInfo(t *testing.T, value interface{}) tdigestInfo {
- require.IsType(t, map[interface{}]interface{}{}, value)
- v := value.(map[interface{}]interface{})
- return tdigestInfo{
- Compression: v["Compression"].(int64),
- Capacity: v["Capacity"].(int64),
- MergedNodes: v["Merged nodes"].(int64),
- UnmergedNodes: v["Unmerged nodes"].(int64),
- MergedWeight: v["Merged weight"].(int64),
- UnmergedWeight: v["Unmerged weight"].(int64),
- Observations: v["Observations"].(int64),
- TotalCompressions: v["Total compressions"].(int64),
- }
-}
-
-func TestTDigest(t *testing.T) {
- configOptions := []util.ConfigOptions{
- {
- Name: "txn-context-enabled",
- Options: []string{"yes", "no"},
- ConfigType: util.YesNo,
- },
- }
-
- configsMatrix, err := util.GenerateConfigsMatrix(configOptions)
- require.NoError(t, err)
-
- for _, configs := range configsMatrix {
- tdigestTests(t, configs)
- }
-}
-
-func tdigestTests(t *testing.T, configs util.KvrocksServerConfigs) {
- srv := util.StartServer(t, configs)
- defer srv.Close()
- ctx := context.Background()
- rdb := srv.NewClient()
- defer func() { require.NoError(t, rdb.Close()) }()
-
- t.Run("tdigest.create with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_create_"
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "hahah").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "1", "hahah").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression", "hahah").Err(), errMsgParseCompression)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression", "0").Err(), errMsgNeedToBePositive)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression", "-1").Err(), errMsgNeedToBePositive)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression", "0.1").Err(), errMsgParseCompression)
-
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key0", "compression", "1").Err())
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key0", "compression", "1").Err(), errMsgKeyAlreadyExists)
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key1", "compression", "1000").Err())
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key", "compression", "1001").Err(), errMsgMustInRange)
- })
-
- t.Run("tdigest.info with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_info_"
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.INFO").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.INFO", keyPrefix+"key", "hahah").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.INFO", keyPrefix+"key").Err(), errMsgKeyNotExist)
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key0", "compression", "1").Err())
- {
- rsp := rdb.Do(ctx, "TDIGEST.INFO", keyPrefix+"key0")
- require.NoError(t, rsp.Err())
- info := toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 1, info.Compression)
- require.EqualValues(t, 1*6+10, info.Capacity)
- require.EqualValues(t, 0, info.MergedNodes)
- require.EqualValues(t, 0, info.UnmergedNodes)
- require.EqualValues(t, 0, info.MergedWeight)
- require.EqualValues(t, 0, info.UnmergedWeight)
- require.EqualValues(t, 0, info.Observations)
- require.EqualValues(t, 0, info.TotalCompressions)
- }
-
- {
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"key1", "compression", "1000").Err())
- rsp := rdb.Do(ctx, "TDIGEST.INFO", keyPrefix+"key1")
- require.NoError(t, rsp.Err())
- info := toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 1000, info.Compression)
- require.EqualValues(t, 1024, info.Capacity) // max is 1024
- require.EqualValues(t, 0, info.MergedNodes)
- require.EqualValues(t, 0, info.UnmergedNodes)
- require.EqualValues(t, 0, info.MergedWeight)
- require.EqualValues(t, 0, info.UnmergedWeight)
- require.EqualValues(t, 0, info.Observations)
- require.EqualValues(t, 0, info.TotalCompressions)
- }
- })
-
- t.Run("tdigest.add with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_add_"
-
- // Satisfy the number of parameters
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.ADD").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.ADD", keyPrefix+"key").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.ADD", keyPrefix+"key", "abc").Err(), "not a valid float")
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.ADD", keyPrefix+"nonexistent", "1.0").Err(), errMsgKeyNotExist)
-
- // Test adding values to a key
- key := keyPrefix + "test1"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key, "compression", "100").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "42.0").Err())
-
- rsp := rdb.Do(ctx, "TDIGEST.INFO", key)
- require.NoError(t, rsp.Err())
- info := toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 1, info.UnmergedNodes)
- require.EqualValues(t, 1, info.Observations)
-
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "1.0", "2.0", "3.0", "4.0", "5.0").Err())
-
- rsp = rdb.Do(ctx, "TDIGEST.INFO", key)
- require.NoError(t, rsp.Err())
- info = toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 6, info.Observations)
-
- // Test adding values to a key with compression
- key2 := keyPrefix + "test2"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key2, "compression", "100").Err())
-
- args := []interface{}{key2}
- for i := 1; i <= 1000; i++ {
- args = append(args, float64(i))
- }
- require.NoError(t, rdb.Do(ctx, append([]interface{}{"TDIGEST.ADD"}, args...)...).Err())
-
- rsp = rdb.Do(ctx, "TDIGEST.INFO", key2)
- require.NoError(t, rsp.Err())
- info = toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 1000, info.Observations)
-
- // Test adding values to a key with compression and merge node
- key3 := keyPrefix + "test3"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key3, "compression", "10").Err())
-
- args = []interface{}{key3}
- for i := 1; i <= 100; i++ {
- args = append(args, float64(i%10))
- }
- require.NoError(t, rdb.Do(ctx, append([]interface{}{"TDIGEST.ADD"}, args...)...).Err())
-
- rsp = rdb.Do(ctx, "TDIGEST.INFO", key3)
- require.NoError(t, rsp.Err())
- info = toTdigestInfo(t, rsp.Val())
-
- require.Greater(t, info.MergedNodes, int64(0))
- require.Greater(t, info.MergedWeight, int64(0))
- require.EqualValues(t, 100, info.Observations)
- require.Greater(t, info.TotalCompressions, int64(0))
- })
-
- t.Run("tdigest.max with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_max_"
-
- // Test invalid arguments
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MAX").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MAX", keyPrefix+"nonexistent").Err(), errMsgKeyNotExist)
-
- // Test with empty tdigest
- key := keyPrefix + "test1"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key, "compression", "100").Err())
- rsp := rdb.Do(ctx, "TDIGEST.MAX", key)
- require.NoError(t, rsp.Err())
- require.EqualValues(t, rsp.Val(), "nan")
-
- // Test with single value
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "42.5").Err())
- rsp = rdb.Do(ctx, "TDIGEST.MAX", key)
- require.NoError(t, rsp.Err())
- require.Equal(t, "42.5", rsp.Val())
-
- // Test with multiple values
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "1.0", "100.5", "50.5", "-10.5").Err())
- rsp = rdb.Do(ctx, "TDIGEST.MAX", key)
- require.NoError(t, rsp.Err())
- require.Equal(t, "100.5", rsp.Val())
- })
-
- t.Run("tdigest.min with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_min_"
-
- // Test invalid arguments
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MIN").Err(), errMsgWrongNumberArg)
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MIN", keyPrefix+"nonexistent").Err(), errMsgKeyNotExist)
-
- // Test with empty tdigest
- key := keyPrefix + "test1"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key, "compression", "100").Err())
- rsp := rdb.Do(ctx, "TDIGEST.MIN", key)
- require.NoError(t, rsp.Err())
- require.EqualValues(t, rsp.Val(), "nan")
-
- // Test with single value
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "42.5").Err())
- rsp = rdb.Do(ctx, "TDIGEST.MIN", key)
- require.NoError(t, rsp.Err())
- require.Equal(t, "42.5", rsp.Val())
-
- // Test with multiple values
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "1.0", "100.5", "50.5", "-10.5").Err())
- rsp = rdb.Do(ctx, "TDIGEST.MIN", key)
- require.NoError(t, rsp.Err())
- require.Equal(t, "-10.5", rsp.Val())
- })
- t.Run("tdigest.reset with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_reset_"
-
- // Testing with no arguments to .RESET
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.RESET").Err(), errMsgWrongNumberArg)
-
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", keyPrefix+"mydigest", "compression", "101").Err())
-
- key := keyPrefix + "mydigest"
- // Adding some data to digest
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key, "-84.3", "199.3", "343.34", "12.34").Err())
-
- // Checking MIN value to ensure data was added
- rsp := rdb.Do(ctx, "TDIGEST.MIN", key)
- require.NoError(t, rsp.Err())
- require.EqualValues(t, rsp.Val(), "-84.3")
-
- // Reset on a non-existent key
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.RESET", keyPrefix+"notexist").Err(), errMsgKeyNotExist)
-
- // Get TDIGEST.INFO before reset
- rsp = rdb.Do(ctx, "TDIGEST.INFO", key)
- require.NoError(t, rsp.Err())
- infoBeforeReset := toTdigestInfo(t, rsp.Val())
-
- // Perform the reset
- require.NoError(t, rdb.Do(ctx, "TDIGEST.RESET", key).Err())
-
- // Get TDIGEST.INFO after reset
- rsp = rdb.Do(ctx, "TDIGEST.INFO", key)
- require.NoError(t, rsp.Err())
- infoAfterReset := toTdigestInfo(t, rsp.Val())
-
- // Ensure capacity remains unchanged
- require.EqualValues(t, infoBeforeReset.Capacity, infoAfterReset.Capacity)
- require.EqualValues(t, 101, infoAfterReset.Compression)
- require.EqualValues(t, 0, infoAfterReset.MergedNodes)
- require.EqualValues(t, 0, infoAfterReset.UnmergedNodes)
- require.EqualValues(t, 0, infoAfterReset.Observations)
- require.EqualValues(t, 0, infoAfterReset.TotalCompressions)
-
- // Reset on an empty digest
- emptyDigestKey := keyPrefix + "empty"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", emptyDigestKey, "COMPRESSION", "100").Err())
- rsp = rdb.Do(ctx, "TDIGEST.RESET", emptyDigestKey)
- require.NoError(t, rsp.Err())
-
- // Ensure empty digest's capacity remains the same
- rsp = rdb.Do(ctx, "TDIGEST.INFO", emptyDigestKey)
- require.NoError(t, rsp.Err())
- infoAfterEmptyReset := toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 100, infoAfterEmptyReset.Compression)
- })
- t.Run("tdigest.quantile with different arguments", func(t *testing.T) {
- keyPrefix := "t_qt_"
-
- // No arguments
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.QUANTILE").Err(), errMsgWrongNumberArg)
-
- // Non-existent key
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.QUANTILE", keyPrefix+"iDoNotExist", "0.5").Err(), errMsgKeyNotExist)
-
- {
- // Quantile on empry tdigest
- key0 := keyPrefix + "00"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key0, "compression", "100").Err())
- rsp := rdb.Do(ctx, "TDIGEST.QUANTILE", key0, "0.4", "0.1", "0.2", "0.3", "0.7", "0.5")
- require.NoError(t, rsp.Err())
- vals, err := rsp.Slice()
- require.NoError(t, err)
- require.Equal(t, 6, len(vals))
- for _, v := range vals {
- s, ok := v.(string)
- require.True(t, ok, "Expected string but got %T", v)
- require.Equal(t, "nan", s, "Expected value to be 'nan'")
- }
- }
- {
- // Quantiles on positive data
- key1 := keyPrefix + "01"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key1, "compression", "100").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key1, "1", "2", "2", "3", "3", "3", "4", "4", "4", "4", "5", "5", "5", "5", "5").Err())
- rsp := rdb.Do(ctx, "TDIGEST.QUANTILE", key1, "0", "0.1", "0.2", "0.3", "0.4", "0.5", "0.6", "0.7", "0.8", "0.9", "1")
- require.NoError(t, rsp.Err())
- vals, err := rsp.Slice()
- require.NoError(t, err)
- require.Len(t, vals, 11)
- expected := []float64{1.0, 2.0, 2.5, 3.0, 3.5, 4.0, 4.0, 5.0, 5.0, 5.0, 5.0}
- for i, v := range vals {
- str, ok := v.(string)
- require.True(t, ok, "expected string but got %T at index %d", v, i)
-
- got, err := strconv.ParseFloat(str, 64)
- require.NoError(t, err, "could not parse value at index %d", i)
-
- require.InEpsilon(t, expected[i], got, 0.0001, "mismatch at index %d", i)
- }
- }
- {
- // Quantiles on negative data
- key2 := keyPrefix + "02"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key2, "compression", "100").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key2, "-1", "-2", "-3", "-4", "-5", "-6", "-7", "-8", "-9", "-10").Err())
- rsp := rdb.Do(ctx, "TDIGEST.QUANTILE", key2, "0", "0.25", "0.5", "0.75", "1")
- require.NoError(t, rsp.Err())
-
- vals, err := rsp.Slice()
- require.NoError(t, err)
- require.Len(t, vals, 5)
-
- expected := []float64{-10.0, -8.0, -5.5, -3.0, -1.0}
- for i, v := range vals {
- str, ok := v.(string)
- require.True(t, ok, "expected string but got %T at index %d", v, i)
-
- got, err := strconv.ParseFloat(str, 64)
- require.NoError(t, err, "could not parse value at index %d", i)
-
- require.InEpsilon(t, expected[i], got, 0.0001, "mismatch at index %d", i)
- }
- }
- {
- // Query with unordered quantiles
- key3 := keyPrefix + "03"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", key3, "compression", "100").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", key3,
- "3", "12", "-3", "-19", "13", "4", "14", "18", "-1", "-5", "15", "-10", "33", "17", "-20",
- ).Err())
- rsp := rdb.Do(ctx, "TDIGEST.QUANTILE", key3,
- "0.9", "0.1", "0.7", "0.3", "0.6", "0.0", "0.55", "0.65", "0.34", "0.88",
- )
- require.NoError(t, rsp.Err())
- vals, err := rsp.Slice()
- require.NoError(t, err)
- require.Equal(t, 10, len(vals))
-
- expected := []float64{
- 18.0,
- -19.0,
- 14.0,
- -3.0,
- 12.5,
- -20.0,
- 12.0,
- 13.0,
- -1.0,
- 18.0,
- }
- for i, v := range vals {
- strVal, ok := v.(string)
- require.True(t, ok, "Expected string at index %d but got %T", i, v)
- numVal, err := strconv.ParseFloat(strVal, 64)
- require.NoError(t, err, "Failed to parse value at index %d: %s", i, strVal)
- require.InDelta(t, expected[i], numVal, 1e-6, "Mismatch at index %d", i)
- }
- }
- })
-
- t.Run("tdigest.merge with different arguments", func(t *testing.T) {
- keyPrefix := "tdigest_merge_"
-
- // no arguments
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE").Err(), errMsgWrongNumberArg)
-
- // merge with no source keys
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key1").Err(), errMsgWrongNumberArg)
-
- // merge with invalid number of source keys
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key2", "hahah").Err(), errMsgWrongNumberArg)
-
- // merge with not matching number of source keys
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key3", 3, "hahah").Err(), errMsgWrongNumberArg)
-
- // merge with negative number of source keys
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key4", -1, "hahah").Err(), errNumkeysMustBePositive)
-
- // merge with non-existent source key
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key5", 1, keyPrefix+"nonexistent").Err(), errMsgKeyNotExist)
-
- // merge with invalid compression keyword
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key6", 1, keyPrefix+"nonexistent", "compression").Err(), errMsgWrongNumberArg)
-
- // merge with invalid compression value
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key7", 1, keyPrefix+"nonexistent", "compression", "hahah").Err(), errMsgParseCompression)
-
- // merge with more than one override
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", keyPrefix+"key8", 1, keyPrefix+"nonexistent", "compression", "100", "override", "override").Err(), errMsgWrongNumberArg)
-
- // create a source digest and add some data
- sourceKey1 := keyPrefix + "source1"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", sourceKey1, "compression", "101").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", sourceKey1, "1.0", "2.0", "3.0").Err())
-
- sourceKey2 := keyPrefix + "source2"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", sourceKey2, "compression", "30").Err())
- require.NoError(t, rdb.Do(ctx, "TDIGEST.ADD", sourceKey2, "4.0", "5.0", "6.0", "100", "-200").Err())
-
- // create a destination digest
- destKey := keyPrefix + "dest"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.CREATE", destKey, "compression", "100").Err())
-
- // merge the source into the destination without override
- require.ErrorContains(t, rdb.Do(ctx, "TDIGEST.MERGE", destKey, 2, sourceKey1, sourceKey2).Err(), errMsgKeyAlreadyExists)
-
- // merge the source into the destination with override
- require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", destKey, 2, sourceKey1, sourceKey2, "override").Err())
-
- // merge to a new destination key
- newDestKey1 := keyPrefix + "new_dest"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", newDestKey1, 2, sourceKey1, sourceKey2).Err())
-
- // merge with same source keys
- newDestKey2 := keyPrefix + "new_dest2"
- require.NoError(t, rdb.Do(ctx, "TDIGEST.MERGE", newDestKey2, 4, sourceKey1, sourceKey2, sourceKey1, sourceKey2).Err())
-
- validation := func(destMergeKey string) {
- rsp := rdb.Do(ctx, "TDIGEST.INFO", destMergeKey)
- require.NoError(t, rsp.Err())
- info := toTdigestInfo(t, rsp.Val())
- require.EqualValues(t, 101, info.Compression)
- require.EqualValues(t, 8, info.Observations)
-
- rsp = rdb.Do(ctx, "TDIGEST.MAX", destMergeKey)
- require.NoError(t, rsp.Err())
- {
- rspval, err := rsp.Float64()
- require.NoError(t, err)
- require.InEpsilon(t, 100, rspval, 0.001)
- }
-
- rsp = rdb.Do(ctx, "TDIGEST.MIN", destMergeKey)
- require.NoError(t, rsp.Err())
- {
- rspval, err := rsp.Float64()
- require.NoError(t, err)
- require.InEpsilon(t, -200, rspval, 0.001)
- }
-
- rsp = rdb.Do(ctx, "TDIGEST.QUANTILE", destMergeKey, "0.1", "0.5", "0.75", "0.9", "0.99", "1")
- require.NoError(t, rsp.Err())
- vals, err := rsp.Slice()
- require.NoError(t, err)
- require.Len(t, vals, 6)
- expected := []float64{-200.0, 4.0, 6.0, 100.0, 100.0, 100.0}
- for i, v := range vals {
- str, ok := v.(string)
- require.True(t, ok, "expected string but got %T at index %d", v, i)
-
- got, err := strconv.ParseFloat(str, 64)
- require.NoError(t, err, "could not parse value at index %d", i)
-
- require.InEpsilon(t, expected[i], got, 0.2, "mismatch at index %d", i)
- }
- }
- validation(destKey)
- validation(newDestKey1)
- validation(newDestKey2)
- })
-}
(kvrocks) 07/10: chore(ci): bump golangci-lint to 2.3.0 (#3071)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 39575eba4c73a77574e754a41329f7940a02ff4b
Author: Aleks Lozovyuk <al...@gmail.com>
AuthorDate: Tue Jul 22 18:40:42 2025 +0300
chore(ci): bump golangci-lint to 2.3.0 (#3071)
---
x.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/x.py b/x.py
index c7c690ec5..be2246289 100755
--- a/x.py
+++ b/x.py
@@ -31,7 +31,7 @@ from shutil import which
CMAKE_REQUIRE_VERSION = (3, 16, 0)
CLANG_FORMAT_REQUIRED_VERSION = (12, 0, 0)
CLANG_TIDY_REQUIRED_VERSION = (12, 0, 0)
-GOLANGCI_LINT_REQUIRED_VERSION = (2, 2, 2)
+GOLANGCI_LINT_REQUIRED_VERSION = (2, 3, 0)
SEMVER_REGEX = re.compile(
r"""
(kvrocks) 03/10: chore(deps): bump cpptrace to v1.0.3 (#3056)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 3a1b631014d746acf7554677b0acad92975ed472
Author: Aleks Lozovyuk <al...@gmail.com>
AuthorDate: Thu Jul 17 06:44:50 2025 +0300
chore(deps): bump cpptrace to v1.0.3 (#3056)
---
cmake/cpptrace.cmake | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/cmake/cpptrace.cmake b/cmake/cpptrace.cmake
index 3750d8dc9..3ed9d7f82 100644
--- a/cmake/cpptrace.cmake
+++ b/cmake/cpptrace.cmake
@@ -20,8 +20,8 @@ include_guard()
include(cmake/utils.cmake)
FetchContent_DeclareGitHubWithMirror(cpptrace
- jeremy-rifkin/cpptrace v1.0.2
- MD5=9174091a2fe1dbceb8e181ae83b6abd7
+ jeremy-rifkin/cpptrace v1.0.3
+ MD5=b23b986fb2b4398adf0bba2e1e894b8d
)
if (SYMBOLIZE_BACKEND STREQUAL "libbacktrace")
(kvrocks) 10/10: release: prepare source release apache-kvrocks-2.13.0
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 154804c4318b32f811c1486ab97543e0f856d61d
Author: edward_xu <xu...@gmail.com>
AuthorDate: Mon Aug 4 20:40:52 2025 +0800
release: prepare source release apache-kvrocks-2.13.0
---
src/VERSION.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/VERSION.txt b/src/VERSION.txt
index 76cff7f1f..a3ebb9f51 100644
--- a/src/VERSION.txt
+++ b/src/VERSION.txt
@@ -1 +1 @@
-unstable
\ No newline at end of file
+2.13.0
\ No newline at end of file
(kvrocks) 08/10: fix(config): the default log-dir should include stdout (#3085)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit f57e019f6acdf0a879a0a25c246859ff83c528b7
Author: Twice <tw...@gmail.com>
AuthorDate: Thu Jul 31 18:58:39 2025 +0800
fix(config): the default log-dir should include stdout (#3085)
---
src/config/config.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/config/config.cc b/src/config/config.cc
index 874e36886..f93a09439 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -892,7 +892,7 @@ Status Config::finish() {
return {Status::NotOK, "replication doesn't support unix socket"};
}
if (db_dir.empty()) db_dir = dir + "/db";
- if (log_dir.empty()) log_dir = dir;
+ if (log_dir.empty()) log_dir = dir + ",stdout";
std::vector<std::string> create_dirs = {dir};
for (const auto &name : create_dirs) {
auto s = rocksdb::Env::Default()->CreateDirIfMissing(name);
(kvrocks) 06/10: test: re-enable skipped TLS replica cases (#3068)
Posted by ed...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
edwardxu pushed a commit to branch 2.13
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit 9fa734e9e0739d8ae3fe781e7a4f709d3f1f566e
Author: Twice <tw...@gmail.com>
AuthorDate: Tue Jul 22 15:36:52 2025 +0800
test: re-enable skipped TLS replica cases (#3068)
Co-authored-by: hulk <hu...@gmail.com>
---
tests/gocase/tls/tls_test.go | 2 --
1 file changed, 2 deletions(-)
diff --git a/tests/gocase/tls/tls_test.go b/tests/gocase/tls/tls_test.go
index 8adba8781..16c0400aa 100644
--- a/tests/gocase/tls/tls_test.go
+++ b/tests/gocase/tls/tls_test.go
@@ -140,8 +140,6 @@ func TestTLS(t *testing.T) {
}
func TestTLSReplica(t *testing.T) {
- t.Skip("FIXME: flaky test with a high frequency of failure")
-
if !util.TLSEnable() {
t.Skip("TLS tests run only if tls enabled.")
}