You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kvrocks.apache.org by hu...@apache.org on 2023/06/18 16:28:58 UTC
[incubator-kvrocks] branch unstable updated: Add the support of the BZMPOP command (#1490)
This is an automated email from the ASF dual-hosted git repository.
hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/incubator-kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 7016bb66 Add the support of the BZMPOP command (#1490)
7016bb66 is described below
commit 7016bb6659a3484266acc0d21383bf0bf558e481
Author: Yangsx-1 <85...@users.noreply.github.com>
AuthorDate: Mon Jun 19 00:28:52 2023 +0800
Add the support of the BZMPOP command (#1490)
This PR adds support for bzmpop, bzpopmax, bzpopmin commands like Redis.
And I also add some tests for these three commands.
---
src/commands/cmd_zset.cc | 348 ++++++++++++++++++++++++++++++-
tests/gocase/unit/type/zset/zset_test.go | 93 ++++++++-
2 files changed, 431 insertions(+), 10 deletions(-)
diff --git a/src/commands/cmd_zset.cc b/src/commands/cmd_zset.cc
index 6ba6f1d9..42c9553d 100644
--- a/src/commands/cmd_zset.cc
+++ b/src/commands/cmd_zset.cc
@@ -73,6 +73,8 @@ class CommandZAdd : public Commander {
return {Status::RedisExecErr, s.ToString()};
}
+ svr->WakeupBlockingConns(args_[1], member_scores_.size());
+
if (flags_.HasIncr()) {
auto new_score = member_scores_[0].score;
if ((flags_.HasNX() || flags_.HasXX() || flags_.HasLT() || flags_.HasGT()) && old_score == new_score &&
@@ -273,6 +275,187 @@ class CommandZPopMax : public CommandZPop {
CommandZPopMax() : CommandZPop(false) {}
};
+static rocksdb::Status PopFromMultipleZsets(redis::ZSet *zset_db, const std::vector<std::string> &keys, bool min,
+ int count, std::string *user_key, std::vector<MemberScore> *member_scores) {
+ rocksdb::Status s;
+ for (auto &key : keys) {
+ s = zset_db->Pop(key, count, min, member_scores);
+ if (!s.ok()) {
+ return s;
+ }
+
+ if (!member_scores->empty()) {
+ *user_key = key;
+ break;
+ }
+ }
+
+ return rocksdb::Status::OK();
+}
+
+class CommandBZPop : public Commander,
+ private EvbufCallbackBase<CommandBZPop, false>,
+ private EventCallbackBase<CommandBZPop> {
+ public:
+ explicit CommandBZPop(bool min) : min_(min) {}
+
+ Status Parse(const std::vector<std::string> &args) override {
+ auto parse_result = ParseInt<int>(args[args.size() - 1], 10);
+ if (!parse_result) {
+ return {Status::RedisParseErr, "timeout is not an integer or out of range"};
+ }
+ if (*parse_result < 0) {
+ return {Status::RedisParseErr, errTimeoutIsNegative};
+ }
+ timeout_ = *parse_result;
+
+ keys_ = std::vector<std::string>(args.begin() + 1, args.end() - 1);
+ return Commander::Parse(args);
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ svr_ = svr;
+ conn_ = conn;
+
+ std::string user_key;
+ std::vector<MemberScore> member_scores;
+
+ redis::ZSet zset_db(svr->storage, conn->GetNamespace());
+ auto s = PopFromMultipleZsets(&zset_db, keys_, min_, 1, &user_key, &member_scores);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+
+ if (!member_scores.empty()) {
+ SendMembersWithScores(member_scores, user_key);
+ return Status::OK();
+ }
+
+ // all sorted sets are empty
+ if (conn->IsInExec()) {
+ *output = redis::MultiLen(-1);
+ return Status::OK(); // no blocking in multi-exec
+ }
+
+ for (const auto &key : keys_) {
+ svr_->BlockOnKey(key, conn_);
+ }
+
+ auto bev = conn->GetBufferEvent();
+ SetCB(bev);
+
+ if (timeout_) {
+ timer_.reset(NewTimer(bufferevent_get_base(bev)));
+ timeval tm = {timeout_, 0};
+ evtimer_add(timer_.get(), &tm);
+ }
+
+ return {Status::BlockingCmd};
+ }
+
+ void SendMembersWithScores(const std::vector<MemberScore> &member_scores, const std::string &user_key) {
+ std::string output;
+ output.append(redis::MultiLen(member_scores.size() * 2 + 1));
+ output.append(redis::BulkString(user_key));
+ for (const auto &ms : member_scores) {
+ output.append(redis::BulkString(ms.member));
+ output.append(redis::BulkString(util::Float2String(ms.score)));
+ }
+ conn_->Reply(output);
+ }
+
+ void OnWrite(bufferevent *bev) {
+ std::string user_key;
+ std::vector<MemberScore> member_scores;
+
+ redis::ZSet zset_db(svr_->storage, conn_->GetNamespace());
+ auto s = PopFromMultipleZsets(&zset_db, keys_, min_, 1, &user_key, &member_scores);
+ if (!s.ok()) {
+ conn_->Reply(redis::Error("ERR " + s.ToString()));
+ return;
+ }
+
+ if (member_scores.empty()) {
+ // The connection may be waked up but can't pop from a zset. For example, connection A is blocked on zset and
+ // connection B added a new element; then connection A was unblocked, but this element may be taken by
+ // another connection C. So we need to block connection A again and wait for the element being added
+ // by disabling the WRITE event.
+ bufferevent_disable(bev, EV_WRITE);
+ return;
+ }
+
+ SendMembersWithScores(member_scores, user_key);
+
+ if (timer_) {
+ timer_.reset();
+ }
+
+ unblockOnAllKeys();
+ conn_->SetCB(bev);
+ bufferevent_enable(bev, EV_READ);
+ // We need to manually trigger the read event since we will stop processing commands
+ // in connection after the blocking command, so there may have some commands to be processed.
+ // Related issue: https://github.com/apache/incubator-kvrocks/issues/831
+ bufferevent_trigger(bev, EV_READ, BEV_TRIG_IGNORE_WATERMARKS);
+ }
+
+ void OnEvent(bufferevent *bev, int16_t events) {
+ if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
+ if (timer_ != nullptr) {
+ timer_.reset();
+ }
+ unblockOnAllKeys();
+ }
+ conn_->OnEvent(bev, events);
+ }
+
+ void TimerCB(int, int16_t) {
+ conn_->Reply(redis::MultiLen(-1));
+ timer_.reset();
+ unblockOnAllKeys();
+ auto bev = conn_->GetBufferEvent();
+ conn_->SetCB(bev);
+ bufferevent_enable(bev, EV_READ);
+ }
+
+ private:
+ bool min_;
+ int timeout_;
+ std::vector<std::string> keys_;
+ Server *svr_ = nullptr;
+ Connection *conn_ = nullptr;
+ UniqueEvent timer_;
+
+ void unblockOnAllKeys() {
+ for (const auto &key : keys_) {
+ svr_->UnblockOnKey(key, conn_);
+ }
+ }
+};
+
+class CommandBZPopMin : public CommandBZPop {
+ public:
+ CommandBZPopMin() : CommandBZPop(true) {}
+};
+
+class CommandBZPopMax : public CommandBZPop {
+ public:
+ CommandBZPopMax() : CommandBZPop(false) {}
+};
+
+static void SendMembersWithScoresForZMpop(Connection *conn, const std::string &user_key,
+ const std::vector<MemberScore> &member_scores) {
+ std::string output;
+ output.append(redis::MultiLen(2));
+ output.append(redis::BulkString(user_key));
+ output.append(redis::MultiLen(member_scores.size() * 2));
+ for (const auto &ms : member_scores) {
+ output.append(redis::BulkString(ms.member));
+ output.append(redis::BulkString(util::Float2String(ms.score)));
+ }
+ conn->Reply(output);
+}
+
class CommandZMPop : public Commander {
public:
CommandZMPop() = default;
@@ -313,16 +496,10 @@ class CommandZMPop : public Commander {
continue;
}
- output->append(redis::MultiLen(2));
- output->append(redis::BulkString(user_key));
- output->append(redis::MultiLen(member_scores.size() * 2));
- for (const auto &ms : member_scores) {
- output->append(redis::BulkString(ms.member));
- output->append(redis::BulkString(util::Float2String(ms.score)));
- }
+ SendMembersWithScoresForZMpop(conn, user_key, member_scores);
return Status::OK();
}
- *output = redis::NilString();
+ *output = redis::MultiLen(-1);
return Status::OK();
}
@@ -338,6 +515,158 @@ class CommandZMPop : public Commander {
int count_ = 1;
};
+class CommandBZMPop : public Commander,
+ private EvbufCallbackBase<CommandBZMPop, false>,
+ private EventCallbackBase<CommandBZMPop> {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ CommandParser parser(args, 1);
+
+ timeout_ = GET_OR_RET(parser.TakeInt<int>(NumericRange<int>{0, std::numeric_limits<int>::max()}));
+ if (timeout_ < 0) {
+ return {Status::RedisParseErr, errTimeoutIsNegative};
+ }
+
+ num_keys_ = GET_OR_RET(parser.TakeInt<int>(NumericRange<int>{1, std::numeric_limits<int>::max()}));
+ for (int i = 0; i < num_keys_; ++i) {
+ keys_.emplace_back(GET_OR_RET(parser.TakeStr()));
+ }
+
+ while (parser.Good()) {
+ if (parser.EatEqICase("min")) {
+ flag_ = ZSET_MIN;
+ } else if (parser.EatEqICase("max")) {
+ flag_ = ZSET_MAX;
+ } else if (parser.EatEqICase("count")) {
+ count_ = GET_OR_RET(parser.TakeInt<int>(NumericRange<int>{1, std::numeric_limits<int>::max()}));
+ } else {
+ return parser.InvalidSyntax();
+ }
+ }
+
+ if (flag_ == ZSET_NONE) {
+ return parser.InvalidSyntax();
+ }
+
+ return Commander::Parse(args);
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ svr_ = svr;
+ conn_ = conn;
+
+ std::string user_key;
+ std::vector<MemberScore> member_scores;
+
+ redis::ZSet zset_db(svr->storage, conn->GetNamespace());
+ auto s = PopFromMultipleZsets(&zset_db, keys_, flag_ == ZSET_MIN, count_, &user_key, &member_scores);
+ if (!s.ok()) {
+ return {Status::RedisExecErr, s.ToString()};
+ }
+
+ if (!member_scores.empty()) {
+ SendMembersWithScoresForZMpop(conn_, user_key, member_scores);
+ return Status::OK();
+ }
+
+ // all sorted sets are empty
+ if (conn->IsInExec()) {
+ *output = redis::MultiLen(-1);
+ return Status::OK(); // no blocking in multi-exec
+ }
+
+ for (const auto &key : keys_) {
+ svr_->BlockOnKey(key, conn_);
+ }
+
+ auto bev = conn->GetBufferEvent();
+ SetCB(bev);
+
+ if (timeout_) {
+ timer_.reset(NewTimer(bufferevent_get_base(bev)));
+ timeval tm = {timeout_, 0};
+ evtimer_add(timer_.get(), &tm);
+ }
+
+ return {Status::BlockingCmd};
+ }
+
+ void OnWrite(bufferevent *bev) {
+ std::string user_key;
+ std::vector<MemberScore> member_scores;
+
+ redis::ZSet zset_db(svr_->storage, conn_->GetNamespace());
+ auto s = PopFromMultipleZsets(&zset_db, keys_, flag_ == ZSET_MIN, count_, &user_key, &member_scores);
+ if (!s.ok()) {
+ conn_->Reply(redis::Error("ERR " + s.ToString()));
+ return;
+ }
+
+ if (member_scores.empty()) {
+ // The connection may be waked up but can't pop from a zset. For example, connection A is blocked on zset and
+ // connection B added a new element; then connection A was unblocked, but this element may be taken by
+ // another connection C. So we need to block connection A again and wait for the element being added
+ // by disabling the WRITE event.
+ bufferevent_disable(bev, EV_WRITE);
+ return;
+ }
+
+ SendMembersWithScoresForZMpop(conn_, user_key, member_scores);
+
+ if (timer_) {
+ timer_.reset();
+ }
+
+ unblockOnAllKeys();
+ conn_->SetCB(bev);
+ bufferevent_enable(bev, EV_READ);
+ // We need to manually trigger the read event since we will stop processing commands
+ // in connection after the blocking command, so there may have some commands to be processed.
+ // Related issue: https://github.com/apache/incubator-kvrocks/issues/831
+ bufferevent_trigger(bev, EV_READ, BEV_TRIG_IGNORE_WATERMARKS);
+ }
+
+ void OnEvent(bufferevent *bev, int16_t events) {
+ if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
+ if (timer_ != nullptr) {
+ timer_.reset();
+ }
+ unblockOnAllKeys();
+ }
+ conn_->OnEvent(bev, events);
+ }
+
+ void TimerCB(int, int16_t events) {
+ conn_->Reply(redis::NilString());
+ timer_.reset();
+ unblockOnAllKeys();
+ auto bev = conn_->GetBufferEvent();
+ conn_->SetCB(bev);
+ bufferevent_enable(bev, EV_READ);
+ }
+
+ static CommandKeyRange Range(const std::vector<std::string> &args) {
+ int num_key = *ParseInt<int>(args[2], 10);
+ return {3, 1 + num_key, 1};
+ }
+
+ private:
+ int timeout_ = 0; // seconds
+ int num_keys_;
+ std::vector<std::string> keys_;
+ enum { ZSET_MIN, ZSET_MAX, ZSET_NONE } flag_ = ZSET_NONE;
+ int count_ = 1;
+ Server *svr_ = nullptr;
+ Connection *conn_ = nullptr;
+ UniqueEvent timer_;
+
+ void unblockOnAllKeys() {
+ for (const auto &key : keys_) {
+ svr_->UnblockOnKey(key, conn_);
+ }
+ }
+};
+
class CommandZRangeStore : public Commander {
public:
explicit CommandZRangeStore() : range_type_(kZRangeRank), direction_(kZRangeDirectionForward) {}
@@ -936,7 +1265,10 @@ REDIS_REGISTER_COMMANDS(MakeCmdAttr<CommandZAdd>("zadd", -4, "write", 1, 1, 1),
MakeCmdAttr<CommandZLexCount>("zlexcount", 4, "read-only", 1, 1, 1),
MakeCmdAttr<CommandZPopMax>("zpopmax", -2, "write", 1, 1, 1),
MakeCmdAttr<CommandZPopMin>("zpopmin", -2, "write", 1, 1, 1),
+ MakeCmdAttr<CommandBZPopMax>("bzpopmax", -3, "write", 1, -2, 1),
+ MakeCmdAttr<CommandBZPopMin>("bzpopmin", -3, "write", 1, -2, 1),
MakeCmdAttr<CommandZMPop>("zmpop", -4, "write", CommandZMPop::Range),
+ MakeCmdAttr<CommandBZMPop>("bzmpop", -5, "write", CommandBZMPop::Range),
MakeCmdAttr<CommandZRangeStore>("zrangestore", -4, "write", 1, 1, 1),
MakeCmdAttr<CommandZRange>("zrange", -4, "read-only", 1, 1, 1),
MakeCmdAttr<CommandZRevRange>("zrevrange", -4, "read-only", 1, 1, 1),
diff --git a/tests/gocase/unit/type/zset/zset_test.go b/tests/gocase/unit/type/zset/zset_test.go
index 94a16b61..ec215271 100644
--- a/tests/gocase/unit/type/zset/zset_test.go
+++ b/tests/gocase/unit/type/zset/zset_test.go
@@ -67,7 +67,7 @@ func createDefaultLexZset(rdb *redis.Client, ctx context.Context) {
{0, "omega"}})
}
-func basicTests(t *testing.T, rdb *redis.Client, ctx context.Context, encoding string) {
+func basicTests(t *testing.T, rdb *redis.Client, ctx context.Context, encoding string, srv *util.KvrocksServer) {
t.Run(fmt.Sprintf("Check encoding - %s", encoding), func(t *testing.T) {
rdb.Del(ctx, "ztmp")
rdb.ZAdd(ctx, "ztmp", redis.Z{Score: 10, Member: "x"})
@@ -318,6 +318,60 @@ func basicTests(t *testing.T, rdb *redis.Client, ctx context.Context, encoding s
require.Equal(t, []redis.Z{{Score: 10, Member: "a"}}, rdb.ZPopMax(ctx, "ztmp", 3).Val())
})
+ t.Run(fmt.Sprintf("BZPOPMIN basics - %s", encoding), func(t *testing.T) {
+ rdb.Del(ctx, "zseta")
+ rdb.Del(ctx, "zsetb")
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"}, redis.Z{Score: 2, Member: "b"}, redis.Z{Score: 3, Member: "c"})
+ rdb.ZAdd(ctx, "zsetb", redis.Z{Score: 1, Member: "d"}, redis.Z{Score: 2, Member: "e"})
+ require.EqualValues(t, 3, rdb.ZCard(ctx, "zseta").Val())
+ require.EqualValues(t, 2, rdb.ZCard(ctx, "zsetb").Val())
+ resultz := rdb.BZPopMin(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 1, Member: "a"}, resultz)
+ resultz = rdb.BZPopMin(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 2, Member: "b"}, resultz)
+ resultz = rdb.BZPopMin(ctx, 0, "zsetb", "zseta").Val().Z
+ require.Equal(t, redis.Z{Score: 1, Member: "d"}, resultz)
+ resultz = rdb.BZPopMin(ctx, 0, "zsetb", "zseta").Val().Z
+ require.Equal(t, redis.Z{Score: 2, Member: "e"}, resultz)
+ resultz = rdb.BZPopMin(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 3, Member: "c"}, resultz)
+ var err = rdb.BZPopMin(ctx, time.Millisecond*1000, "zseta", "zsetb").Err()
+ require.Equal(t, redis.Nil, err)
+
+ rd := srv.NewTCPClient()
+ defer func() { require.NoError(t, rd.Close()) }()
+ require.NoError(t, rd.WriteArgs("bzpopmin", "zseta", "0"))
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"})
+ rd.MustReadStrings(t, []string{"zseta", "a", "1"})
+ })
+
+ t.Run(fmt.Sprintf("BZPOPMAX basics - %s", encoding), func(t *testing.T) {
+ rdb.Del(ctx, "zseta")
+ rdb.Del(ctx, "zsetb")
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"}, redis.Z{Score: 2, Member: "b"}, redis.Z{Score: 3, Member: "c"})
+ rdb.ZAdd(ctx, "zsetb", redis.Z{Score: 1, Member: "d"}, redis.Z{Score: 2, Member: "e"})
+ require.EqualValues(t, 3, rdb.ZCard(ctx, "zseta").Val())
+ require.EqualValues(t, 2, rdb.ZCard(ctx, "zsetb").Val())
+ resultz := rdb.BZPopMax(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 3, Member: "c"}, resultz)
+ resultz = rdb.BZPopMax(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 2, Member: "b"}, resultz)
+ resultz = rdb.BZPopMax(ctx, 0, "zsetb", "zseta").Val().Z
+ require.Equal(t, redis.Z{Score: 2, Member: "e"}, resultz)
+ resultz = rdb.BZPopMax(ctx, 0, "zsetb", "zseta").Val().Z
+ require.Equal(t, redis.Z{Score: 1, Member: "d"}, resultz)
+ resultz = rdb.BZPopMax(ctx, 0, "zseta", "zsetb").Val().Z
+ require.Equal(t, redis.Z{Score: 1, Member: "a"}, resultz)
+ var err = rdb.BZPopMin(ctx, time.Millisecond*1000, "zseta", "zsetb").Err()
+ require.Equal(t, redis.Nil, err)
+
+ rd := srv.NewTCPClient()
+ defer func() { require.NoError(t, rd.Close()) }()
+ require.NoError(t, rd.WriteArgs("bzpopmax", "zseta", "0"))
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"})
+ rd.MustReadStrings(t, []string{"zseta", "a", "1"})
+ })
+
t.Run(fmt.Sprintf("ZMPOP basics - %s", encoding), func(t *testing.T) {
rdb.Del(ctx, "zseta")
rdb.Del(ctx, "zsetb")
@@ -338,6 +392,41 @@ func basicTests(t *testing.T, rdb *redis.Client, ctx context.Context, encoding s
require.EqualValues(t, 0, rdb.Exists(ctx, "zseta", "zsetb").Val())
})
+ t.Run(fmt.Sprintf("BZMPOP basics - %s", encoding), func(t *testing.T) {
+ rdb.Del(ctx, "zseta")
+ rdb.Del(ctx, "zsetb")
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"}, redis.Z{Score: 2, Member: "b"}, redis.Z{Score: 3, Member: "c"})
+ rdb.ZAdd(ctx, "zsetb", redis.Z{Score: 1, Member: "d"}, redis.Z{Score: 2, Member: "e"})
+ require.EqualValues(t, 3, rdb.ZCard(ctx, "zseta").Val())
+ require.EqualValues(t, 2, rdb.ZCard(ctx, "zsetb").Val())
+ var key, zset = rdb.BZMPop(ctx, 0, "min", 1, "zseta").Val()
+ require.Equal(t, "zseta", key)
+ require.Equal(t, []redis.Z{{Score: 1, Member: "a"}}, zset)
+ key, zset = rdb.BZMPop(ctx, 0, "max", 2, "zsetb").Val()
+ require.Equal(t, "zsetb", key)
+ require.Equal(t, []redis.Z{{Score: 2, Member: "e"}, {Score: 1, Member: "d"}}, zset)
+ key, zset = rdb.BZMPop(ctx, 0, "min", 3, "zseta").Val()
+ require.Equal(t, "zseta", key)
+ require.Equal(t, []redis.Z{{Score: 2, Member: "b"}, {Score: 3, Member: "c"}}, zset)
+ require.Equal(t, redis.Nil, rdb.BZMPop(ctx, time.Millisecond*1000, "max", 10, "zseta", "zsetb").Err())
+
+ rd := srv.NewClient()
+ defer func() { require.NoError(t, rd.Close()) }()
+ ch := make(chan *redis.ZSliceWithKeyCmd)
+ go func() {
+ ch <- rd.BZMPop(ctx, 0, "min", 10, "zseta")
+ }()
+ require.Eventually(t, func() bool {
+ cnt, _ := strconv.Atoi(util.FindInfoEntry(rdb, "blocked_clients"))
+ return cnt == 1
+ }, 5*time.Second, 100*time.Millisecond)
+ rdb.ZAdd(ctx, "zseta", redis.Z{Score: 1, Member: "a"}, redis.Z{Score: 2, Member: "b"})
+ r := <-ch
+ key, zset = r.Val()
+ require.Equal(t, "zseta", key)
+ require.Equal(t, []redis.Z{{Score: 1, Member: "a"}, {Score: 2, Member: "b"}}, zset)
+ })
+
t.Run(fmt.Sprintf("ZRANGESTORE basics - %s", encoding), func(t *testing.T) {
rdb.Del(ctx, "zsrc")
rdb.Del(ctx, "zdst")
@@ -1295,7 +1384,7 @@ func TestZset(t *testing.T) {
rdb := srv.NewClient()
defer func() { require.NoError(t, rdb.Close()) }()
- basicTests(t, rdb, ctx, "skiplist")
+ basicTests(t, rdb, ctx, "skiplist", srv)
t.Run("ZUNIONSTORE regression, should not create NaN in scores", func(t *testing.T) {
rdb.ZAdd(ctx, "z", redis.Z{Score: math.Inf(-1), Member: "neginf"})