This is an automated email from the ASF dual-hosted git repository.
twice pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new cc2bb3c7e feat(command): add FLUSHSLOTS command to clear keys in
specified slot ranges in cluster mode (#3375)
cc2bb3c7e is described below
commit cc2bb3c7e3d77c6b88af5f1b42e9018e0bb31ec8
Author: sryan yuan <[email protected]>
AuthorDate: Fri Feb 27 13:41:53 2026 +0800
feat(command): add FLUSHSLOTS command to clear keys in specified slot
ranges in cluster mode (#3375)
Add FLUSHSLOTS command to clear keys in specified slot ranges in cluster
mode
FLUSHSLOTS allows clearing keys from one or more cluster slots without
affecting other slots.
It operates on the current connection's namespace (cluster mode does not
support multiple namespaces).
The command returns an error if a slot migration is in progress to
prevent conflicts.
Slot ranges can be specified as:
- Single slot: FLUSHSLOTS 1234
- Range: FLUSHSLOTS 100-200
- Multiple ranges: FLUSHSLOTS “0 2 4 6-9”
Example:
FLUSHSLOTS "0 2 4 6-9"
Unit tests have been added to verify clearing multiple slots and
ensuring remaining keys match expectations.
---------
Co-authored-by: yxj25245 <[email protected]>
---
src/cluster/cluster.cc | 2 +-
src/commands/cmd_cluster.cc | 23 ++++++++++++-
tests/gocase/integration/cluster/cluster_test.go | 43 ++++++++++++++++++++++++
3 files changed, 66 insertions(+), 2 deletions(-)
diff --git a/src/cluster/cluster.cc b/src/cluster/cluster.cc
index 8579cbf0e..b8cfea020 100644
--- a/src/cluster/cluster.cc
+++ b/src/cluster/cluster.cc
@@ -55,7 +55,7 @@ Cluster::Cluster(Server *srv, std::vector<std::string> binds,
int port)
// cluster data, so these commands should be executed exclusively, and
ReadWriteLock
// also can guarantee accessing data is safe.
bool Cluster::SubCommandIsExecExclusive(const std::string &subcommand) {
- std::array subcommands = {"setnodes", "setnodeid", "setslot", "import",
"reset"};
+ std::array subcommands = {"setnodes", "setnodeid", "setslot", "import",
"reset", "flushslots"};
return std::any_of(std::begin(subcommands), std::end(subcommands),
[&subcommand](const std::string &val) { return
util::EqualICase(val, subcommand); });
diff --git a/src/commands/cmd_cluster.cc b/src/commands/cmd_cluster.cc
index 7a16ddd9a..87c620f42 100644
--- a/src/commands/cmd_cluster.cc
+++ b/src/commands/cmd_cluster.cc
@@ -237,7 +237,14 @@ class CommandClusterX : public Commander {
return Status::OK();
}
- return {Status::RedisParseErr, "CLUSTERX command, CLUSTERX
VERSION|MYID|SETNODEID|SETNODES|SETSLOT|MIGRATE"};
+ // CLUSTERX FLUSHSLOTS $SLOT_RANGES
+ if (subcommand_ == "flushslots") {
+ if (args.size() != 3) return {Status::RedisParseErr,
errWrongNumOfArguments};
+ return CommandTable::ParseSlotRanges(args.back(), slot_ranges_);
+ }
+
+ return {Status::RedisParseErr,
+ "CLUSTERX command, CLUSTERX
VERSION|MYID|SETNODEID|SETNODES|SETSLOT|MIGRATE|FLUSHSLOTS"};
}
Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv,
Connection *conn, std::string *output) override {
@@ -289,6 +296,20 @@ class CommandClusterX : public Commander {
} else {
return s;
}
+ } else if (subcommand_ == "flushslots") {
+ if (srv->slot_migrator->IsMigrationInProgress()) {
+ return {Status::RedisExecErr, "Cannot flush slot when migration is in
progress"};
+ }
+
+ std::string ns = conn->GetNamespace();
+ Database redis(srv->storage, ns);
+ for (const auto &slot_range : slot_ranges_) {
+ if (auto s = redis.ClearKeysOfSlotRange(ctx, ns, slot_range); !s.ok())
{
+ return {Status::RedisExecErr, s.ToString()};
+ }
+ }
+
+ *output = redis::RESP_OK;
} else {
return {Status::RedisExecErr, "Invalid cluster command options"};
}
diff --git a/tests/gocase/integration/cluster/cluster_test.go
b/tests/gocase/integration/cluster/cluster_test.go
index 79008bf4a..7e597ac86 100644
--- a/tests/gocase/integration/cluster/cluster_test.go
+++ b/tests/gocase/integration/cluster/cluster_test.go
@@ -602,3 +602,46 @@ func TestClusterReset(t *testing.T) {
require.NoError(t, rdb0.Do(ctx, "clusterx", "SETNODES",
clusterNodes, "1").Err())
})
}
+
+func TestClusterFlushSlots(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{"cluster-enabled": "yes"})
+ defer srv.Close()
+
+ rdb := srv.NewClient()
+ defer func() {
+ require.NoError(t, rdb.Close())
+ }()
+
+ ctx := context.Background()
+ nodeID := "YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY"
+ require.NoError(t, rdb.Do(ctx, "clusterx", "SETNODEID", nodeID).Err())
+ clusterNodes := fmt.Sprintf("%s %s %d master - 0-16383", nodeID,
srv.Host(), srv.Port())
+ require.NoError(t, rdb.Do(ctx, "clusterx", "SETNODES", clusterNodes,
"1").Err())
+
+ slotKeys := []string{"{3560}key", "{22179}key", "{48756}key",
"{2977}key", "{4569}key",
+ "{460}key", "{4384}key", "{41432}key", "{46920}key",
"{9073}key"}
+ initKeys := func(cli *redis.Client) error {
+ if err := cli.FlushDB(ctx).Err(); err != nil {
+ return err
+ }
+ for _, key := range slotKeys {
+ if err := cli.Set(ctx, key, "value", 0).Err(); err !=
nil {
+ return err
+ }
+ }
+ return nil
+ }
+
+ t.Run("Flush slots", func(t *testing.T) {
+ require.NoError(t, initKeys(rdb))
+
+ r, err := rdb.Do(ctx, "clusterx", "flushslots", "0 2 4
6-9").Result()
+ require.NoError(t, err)
+ require.Equal(t, "OK", r)
+
+ expectedRemainingKeys := []string{slotKeys[1], slotKeys[3],
slotKeys[5]}
+ keys, err := rdb.Keys(ctx, "*").Result()
+ require.NoError(t, err)
+ require.ElementsMatch(t, keys, expectedRemainingKeys)
+ })
+}
