feat(config): allow to change the max length of the bulk string

Currently, the proto's max length of bulk string is hard code to 512MiB,
and it's impossible to create a string value more than that. And Redis
allows to change this value by the configuration `proto-max-bulk-len`,
so we also just add this to align the behavior.

kvrocks.conf

@@ -61,6 +61,11 @@ cluster-enabled no
 # Default: no
 repl-namespace-enabled no
 
+# By default, the max length of bulk string is limited to 512MB. If you want to
+# change this limit to a different value(must >= 1MiB), you can use the following configuration.
+#
+# proto-max-bulk-len 536870912
+
 # Persist the cluster nodes topology in local file($dir/nodes.conf). This configuration
 # takes effect only if the cluster mode was enabled.
 #

src/config/config.cc

@@ -187,6 +187,7 @@ Config::Config() {
       {"redis-cursor-compatible", false, new YesNoField(&redis_cursor_compatible, false)},
       {"resp3-enabled", false, new YesNoField(&resp3_enabled, false)},
       {"repl-namespace-enabled", false, new YesNoField(&repl_namespace_enabled, false)},
+      {"proto-max-bulk-len", false, new UInt64Field(&proto_max_bulk_len, 512 * MiB, 1 * MiB, UINT64_MAX)},
       {"json-max-nesting-depth", false, new IntField(&json_max_nesting_depth, 1024, 0, INT_MAX)},
       {"json-storage-format", false,
        new EnumField<JsonStorageFormat>(&json_storage_format, json_storage_formats, JsonStorageFormat::JSON)},

src/config/config.h

@@ -96,6 +96,7 @@ struct Config {
   int max_backup_keep_hours = 24;
   int slowlog_log_slower_than = 100000;
   int slowlog_max_len = 128;
+  uint64_t proto_max_bulk_len = 512 * 1024 * 1024;
   bool daemonize = false;
   SupervisedMode supervised_mode = kSupervisedNone;
   bool slave_readonly = true;

src/config/config_type.h

@@ -46,6 +46,7 @@ class IntegerField;
 using IntField = IntegerField<int>;
 using UInt32Field = IntegerField<uint32_t>;
 using Int64Field = IntegerField<int64_t>;
+using UInt64Field = IntegerField<uint64_t>;
 
 template <typename Enum>
 struct ConfigEnum {

src/server/redis_request.cc

@@ -108,7 +108,7 @@ Status Request::Tokenize(evbuffer *input) {
         }
 
         bulk_len_ = *parse_result;
-        if (bulk_len_ > PROTO_BULK_MAX_SIZE) {
+        if (bulk_len_ > srv_->GetConfig()->proto_max_bulk_len) {
           return {Status::NotOK, "Protocol error: invalid bulk length"};
         }
 

src/server/redis_request.h

@@ -33,7 +33,6 @@ class Server;
 namespace redis {
 
 constexpr size_t PROTO_INLINE_MAX_SIZE = 16 * 1024L;
-constexpr size_t PROTO_BULK_MAX_SIZE = 512 * 1024L * 1024L;
 constexpr size_t PROTO_MULTI_MAX_SIZE = 1024 * 1024L;
 
 using CommandTokens = std::vector<std::string>;

src/types/redis_string.cc

@@ -27,7 +27,6 @@
 #include <string>
 
 #include "parse_util.h"
-#include "server/redis_request.h"
 #include "storage/redis_metadata.h"
 #include "time_util.h"
 
@@ -553,7 +552,7 @@ rocksdb::Status String::LCS(const std::string &user_key1, const std::string &use
   // Allocate the LCS table.
   uint64_t dp_size = (alen + 1) * (blen + 1);
   uint64_t bulk_size = dp_size * sizeof(uint32_t);
-  if (bulk_size > PROTO_BULK_MAX_SIZE || bulk_size / dp_size != sizeof(uint32_t)) {
+  if (bulk_size > storage_->GetConfig()->proto_max_bulk_len || bulk_size / dp_size != sizeof(uint32_t)) {
     return rocksdb::Status::Aborted("Insufficient memory, transient memory for LCS exceeds proto-max-bulk-len");
   }
   std::vector<uint32_t> dp(dp_size, 0);

tests/gocase/unit/config/config_test.go

@@ -245,3 +245,27 @@ func TestDynamicChangeWorkerThread(t *testing.T) {
 		require.Equal(t, "bar", rdb.Get(ctx, "foo").Val())
 	})
 }
+
+func TestChangeProtoMaxBulkLen(t *testing.T) {
+	configs := map[string]string{}
+	srv := util.StartServer(t, configs)
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	// Default value is 512MB
+	vals, err := rdb.ConfigGet(ctx, "proto-max-bulk-len").Result()
+	require.NoError(t, err)
+	require.EqualValues(t, "536870912", vals["proto-max-bulk-len"])
+
+	// Change to 2MB
+	require.NoError(t, rdb.ConfigSet(ctx, "proto-max-bulk-len", "2097152").Err())
+	vals, err = rdb.ConfigGet(ctx, "proto-max-bulk-len").Result()
+	require.NoError(t, err)
+	require.EqualValues(t, "2097152", vals["proto-max-bulk-len"])
+
+	// Must be >= 1MB
+	require.Error(t, rdb.ConfigSet(ctx, "proto-max-bulk-len", "1024").Err())
+}