Merge branch 'unstable' of https://github.com/wanghan-sanechips/kvrocks into unstable

Author: wanghan-sanechips

cmake/rocksdb.cmake

@@ -26,8 +26,8 @@ endif()
 include(cmake/utils.cmake)
 
 FetchContent_DeclareGitHubWithMirror(rocksdb
-  facebook/rocksdb v10.2.1
-  MD5=6dc9686856ec407cd98d78be06d53fa7
+  facebook/rocksdb v10.4.2
+  MD5=2581c62b42cbb1c2d158bb2c54355b6e
 )
 
 FetchContent_GetProperties(jemalloc)

src/commands/cmd_server.cc

@@ -1527,6 +1527,18 @@ class CommandFlushMemTable : public Commander {
   rocksdb::FlushOptions flush_options_;
 };
 
+class CommandFlushBlockCache : public Commander {
+ public:
+  Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, [[maybe_unused]] Connection *conn,
+                 std::string *output) override {
+    srv->storage->FlushBlockCache();
+
+    *output = redis::RESP_OK;
+    info("FLUSHBLOCKCACHE is triggered and executed successfully");
+    return Status::OK();
+  }
+};
+
 REDIS_REGISTER_COMMANDS(Server, MakeCmdAttr<CommandAuth>("auth", 2, "read-only ok-loading auth", NO_KEY),
                         MakeCmdAttr<CommandPing>("ping", -1, "read-only", NO_KEY),
                         MakeCmdAttr<CommandSelect>("select", 2, "read-only", NO_KEY),
@@ -1570,5 +1582,6 @@ REDIS_REGISTER_COMMANDS(Server, MakeCmdAttr<CommandAuth>("auth", 2, "read-only o
                         MakeCmdAttr<CommandDump>("dump", 2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandPollUpdates>("pollupdates", -2, "read-only admin", NO_KEY),
                         MakeCmdAttr<CommandSST>("sst", -3, "write exclusive admin", 1, 1, 1),
-                        MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY), )
+                        MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY),
+                        MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY), )
 }  // namespace redis

src/storage/storage.cc

@@ -843,6 +843,8 @@ rocksdb::Status Storage::ingestSST(rocksdb::ColumnFamilyHandle *cf_handle,
   return db_->IngestExternalFile(cf_handle, sst_file_names, options);
 }
 
+void Storage::FlushBlockCache() { shared_block_cache_->EraseUnRefEntries(); }
+
 Status Storage::ReplicaApplyWriteBatch(rocksdb::WriteBatch *batch) {
   return applyWriteBatch(default_write_opts_, batch);
 }

src/storage/storage.h

@@ -276,6 +276,7 @@ class Storage {
                                               const rocksdb::FlushOptions &options);
   [[nodiscard]] StatusOr<int> IngestSST(const std::string &folder,
                                         const rocksdb::IngestExternalFileOptions &ingest_options);
+  void FlushBlockCache();
 
   rocksdb::DB *GetDB();
   bool IsClosing() const { return db_closing_; }

tests/gocase/unit/flushblockcache/flushblockcache_test.go

@@ -0,0 +1,82 @@
+/*
+ * 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 flushblockcache
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/apache/kvrocks/tests/gocase/util"
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/require"
+)
+
+func getBlockCacheSize(rdb *redis.Client) (int64, error) {
+	value := util.FindInfoEntry(rdb, "block_cache_usage", "rocksdb")
+	return strconv.ParseInt(value, 10, 64)
+}
+
+func TestFlushBlockCache(t *testing.T) {
+	configs := map[string]string{}
+	srv := util.StartServer(t, configs)
+	defer srv.Close()
+
+	rdb := srv.NewClient()
+	defer func() {
+		require.NoError(t, rdb.Close())
+	}()
+
+	ctx := context.Background()
+
+	t.Run("flushblockcache", func(t *testing.T) {
+		_, err := rdb.Do(ctx, "SET", "A", "KVROCKS").Result()
+		require.NoError(t, err)
+		_, err = rdb.Do(ctx, "FLUSHMEMTABLE").Result()
+		require.NoError(t, err)
+		_, err = rdb.Do(ctx, "GET", "A").Result()
+		require.NoError(t, err)
+		initCacheSize, err := getBlockCacheSize(rdb)
+		require.NoError(t, err)
+		_, err = rdb.Do(ctx, "FLUSHBLOCKCACHE").Result()
+		require.NoError(t, err)
+		cacheSize, err := getBlockCacheSize(rdb)
+		require.NoError(t, err)
+		require.Less(t, cacheSize, initCacheSize)
+		require.Equal(t, "KVROCKS", rdb.Do(ctx, "GET", "A").Val())
+	})
+}
+
+func TestFlushBlockCacheInvalid(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	rdb := srv.NewClient()
+	defer func() {
+		require.NoError(t, rdb.Close())
+	}()
+
+	ctx := context.Background()
+
+	t.Run("invalid arguments", func(t *testing.T) {
+		_, err := rdb.Do(ctx, "FLUSHBLOCKCACHE", "ARG").Result()
+		require.Contains(t, err.Error(), "wrong number of arguments")
+	})
+}