feat: Add DELPREFIX to delete keys by prefix

src/commands/cmd_key.cc

@@ -23,6 +23,10 @@
 #include "commander.h"
 #include "commands/ttl_util.h"
 #include "error_constants.h"
+#include "rocksdb/db.h"
+#include "rocksdb/iterator.h"
+#include "rocksdb/status.h"
+#include "rocksdb/write_batch.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
 #include "storage/redis_db.h"
@@ -336,6 +340,56 @@ class CommandPersist : public Commander {
   }
 };
 
+class CommandDelPrefix : public Commander {
+ public:
+  Status Execute(engine::Context &ctx, Server *srv, redis::Connection *conn, std::string *output) override {
+    // Suppress unused parameter warnings
+    (void)ctx;
+    (void)conn;
+
+    if (args_.size() < 2) return {Status::NotOK, "Missing prefix argument"};
+
+    // Placeholder for cluster mode check
+    // if (srv->IsClusterMode()) {
+    //   return {Status::NotOK, "delprefix command is disabled in cluster mode to avoid inconsistencies"};
+    // }
+
+    std::string prefix = args_[1];
+    auto db = srv->storage->GetDB();
+    if (!db) return {Status::NotOK, "DB not initialized"};
+
+    // Creating an iterator with ReadOptions
+    rocksdb::ReadOptions read_options;
+    std::unique_ptr<rocksdb::Iterator> it(db->NewIterator(read_options));
+    if (!it) return {Status::NotOK, "Failed to create iterator"};
+
+    rocksdb::WriteBatch batch;
+    int delete_count = 0;
+
+    for (it->Seek(prefix); it->Valid(); it->Next()) {
+      if (!it->key().starts_with(rocksdb::Slice(prefix))) break;
+      batch.Delete(it->key());
+      delete_count++;
+    }
+
+    if (!it->status().ok()) {
+      return {Status::NotOK, "Iterator error: " + it->status().ToString()};
+    }
+
+    // Only write batch if there are keys to delete
+    if (delete_count > 0) {
+      rocksdb::WriteOptions write_options;
+      rocksdb::Status s = db->Write(write_options, &batch);
+      if (!s.ok()) {
+        return {Status::NotOK, "Write batch error: " + s.ToString()};
+      }
+    }
+
+    *output = std::to_string(delete_count);
+    return Status::OK();
+  }
+};
+
 class CommandDel : public Commander {
  public:
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
@@ -553,8 +607,35 @@ class CommandSort : public Commander {
   SortArgument sort_argument_;
 };
 
+class CommandKMetadata : public Commander {
+ public:
+  Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    redis::Database redis(srv->storage, conn->GetNamespace());
+    std::string &key = args_[1];
+    std::string nskey = redis.AppendNamespacePrefix(key);
+
+    Metadata metadata(kRedisNone, false);
+    auto s = redis.GetMetadata(ctx, RedisTypes::All(), nskey, &metadata);
+    if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+
+    if (metadata.IsSingleKVType()) {
+      *output = conn->Map({{redis::BulkString("type"), redis::BulkString(metadata.TypeName())},
+                           {redis::BulkString("expire"), redis::Integer(metadata.expire)},
+                           {redis::BulkString("flags"), redis::Integer(metadata.flags)}});
+    } else {
+      *output = conn->Map({{redis::BulkString("type"), redis::BulkString(metadata.TypeName())},
+                           {redis::BulkString("size"), redis::Integer(metadata.size)},
+                           {redis::BulkString("expire"), redis::Integer(metadata.expire)},
+                           {redis::BulkString("flags"), redis::Integer(metadata.flags)},
+                           {redis::BulkString("version"), redis::Integer(metadata.version)}});
+    }
+    return Status::OK();
+  }
+};
+
 REDIS_REGISTER_COMMANDS(Key, MakeCmdAttr<CommandTTL>("ttl", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandPTTL>("pttl", 2, "read-only", 1, 1, 1),
+                        MakeCmdAttr<CommandDelPrefix>("delprefix", 2, "write", 1, 1, 1),
                         MakeCmdAttr<CommandType>("type", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandMove>("move", 3, "write", 1, 1, 1),
                         MakeCmdAttr<CommandMoveX>("movex", 3, "write", 1, 1, 1),
@@ -573,6 +654,7 @@ REDIS_REGISTER_COMMANDS(Key, MakeCmdAttr<CommandTTL>("ttl", 2, "read-only", 1, 1
                         MakeCmdAttr<CommandRenameNX>("renamenx", 3, "write", 1, 2, 1),
                         MakeCmdAttr<CommandCopy>("copy", -3, "write", 1, 2, 1),
                         MakeCmdAttr<CommandSort<false>>("sort", -2, "write slow", 1, 1, 1),
-                        MakeCmdAttr<CommandSort<true>>("sort_ro", -2, "read-only slow", 1, 1, 1))
+                        MakeCmdAttr<CommandSort<true>>("sort_ro", -2, "read-only slow", 1, 1, 1),
+                        MakeCmdAttr<CommandKMetadata>("kmetadata", 2, "read-only", 1, 1, 1))
 
 }  // namespace redis

tests/gocase/integration/cli/cmd_delprefix_test.go

@@ -0,0 +1,105 @@
+/*
+ * 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 cli
+
+import (
+	"context"
+	"testing"
+
+	"github.com/apache/kvrocks/tests/gocase/util" // Utilize common utilities
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDelPrefix(t *testing.T) {
+	ctx := context.Background()
+	client := redis.NewClient(&redis.Options{
+		Addr: "localhost:6666", // Ensure this matches the Kvrocks port
+	})
+	defer client.Close()
+
+	// Set up test keys
+	client.Set(ctx, "test:1", "value1", 0)
+	client.Set(ctx, "test:2", "value2", 0)
+	client.Set(ctx, "other:1", "value3", 0)
+
+	// Run the DELPREFIX command
+	res, err := client.Do(ctx, "DELPREFIX", "test:").Int()
+	assert.NoError(t, err)
+	assert.Equal(t, 2, res, "Expected to delete 2 keys")
+
+	// Ensure the prefixed keys are deleted, but others remain
+	existsTest1, _ := client.Exists(ctx, "test:1").Result()
+	existsTest2, _ := client.Exists(ctx, "test:2").Result()
+	existsOther1, _ := client.Exists(ctx, "other:1").Result()
+
+	assert.Equal(t, int64(0), existsTest1, "test:1 should be deleted")
+	assert.Equal(t, int64(0), existsTest2, "test:2 should be deleted")
+	assert.Equal(t, int64(1), existsOther1, "other:1 should still exist")
+}
+
+func TestDelPrefixClusterMode(t *testing.T) {
+	t.Parallel()
+	// Start server with cluster mode enabled
+	srv := util.StartServer(t, map[string]string{"cluster-enabled": "yes"})
+	defer srv.Close()
+
+	ctx := context.Background()
+	client := srv.NewClient()
+	defer func() { require.NoError(t, client.Close()) }()
+
+	// Simulate cluster mode by adding keys with hash prefixes
+	client.Set(ctx, "{hash1}test:1", "value1", 0)
+	client.Set(ctx, "{hash1}test:2", "value2", 0)
+	client.Set(ctx, "{hash2}other:1", "value3", 0)
+
+	// Run the DELPREFIX command (expect it to be disabled)
+	_, err := client.Do(ctx, "DELPREFIX", "{hash1}test:").Result()
+	require.ErrorContains(t, err, "disabled in cluster mode")
+
+	// Ensure no keys are deleted
+	existsTest1, _ := client.Exists(ctx, "{hash1}test:1").Result()
+	existsTest2, _ := client.Exists(ctx, "{hash1}test:2").Result()
+	existsOther1, _ := client.Exists(ctx, "{hash2}other:1").Result()
+
+	assert.Equal(t, int64(1), existsTest1, "{hash1}test:1 should still exist")
+	assert.Equal(t, int64(1), existsTest2, "{hash1}test:2 should still exist")
+	assert.Equal(t, int64(1), existsOther1, "{hash2}other:1 should still exist")
+}
+
+func TestDelPrefixDisabledInClusterMode(t *testing.T) {
+	t.Parallel()
+	// Start server with cluster mode enabled
+	srv := util.StartServer(t, map[string]string{"cluster-enabled": "yes"})
+	defer srv.Close()
+
+	ctx := context.Background()
+	client := srv.NewClient()
+	defer func() { require.NoError(t, client.Close()) }()
+
+	// Simulate cluster mode by adding keys with hash prefixes
+	client.Set(ctx, "{hash1}test:1", "value1", 0)
+	client.Set(ctx, "{hash1}test:2", "value2", 0)
+
+	// Assume DELPREFIX command is disabled in cluster mode, expect an error
+	_, err := client.Do(ctx, "DELPREFIX", "{hash1}test:").Result()
+	require.ErrorContains(t, err, "disabled in cluster mode")
+}