feat(config): implement integer config option with unit (apache#2448)

kvrocks.conf

@@ -63,6 +63,7 @@ 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.
+# It can be just an integer (e.g. 10000000), or an integer followed by a unit (e.g. 12M, 7G, 2T).
 #
 # proto-max-bulk-len 536870912
 

src/common/bit_util.h

@@ -20,6 +20,14 @@
 
 #pragma once
 
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+
+#include "encoding.h"
+#include "status.h"
+#include "type_util.h"
+
 namespace util {
 
 /* Count number of bits set in the binary array pointed by 's' and long
@@ -146,4 +154,14 @@ inline int64_t RawBitpos(const uint8_t *c, int64_t count, bool bit) {
 
 }  // namespace msb
 
+// num << bit <= MAX  ->  num <= MAX >> bit
+template <typename T, typename U>
+StatusOr<T> CheckedShiftLeft(T num, U bit) {
+  if (num <= std::numeric_limits<T>::max() >> bit) {
+    return num << bit;
+  }
+
+  return {Status::NotOK, "arithmetic overflow"};
+}
+
 }  // namespace util

src/common/parse_util.cc

@@ -22,31 +22,23 @@
 
 #include <limits>
 
-// num << bit <= MAX  ->  num <= MAX >> bit
-template <typename T, typename U>
-StatusOr<T> CheckedShiftLeft(T num, U bit) {
-  if (num <= std::numeric_limits<T>::max() >> bit) {
-    return num << bit;
-  }
-
-  return {Status::NotOK, "arithmetic overflow"};
-}
+#include "bit_util.h"
 
 StatusOr<std::uint64_t> ParseSizeAndUnit(const std::string &v) {
   auto [num, rest] = GET_OR_RET(TryParseInt<std::uint64_t>(v.c_str(), 10));
 
   if (*rest == 0) {
     return num;
   } else if (util::EqualICase(rest, "k")) {
-    return CheckedShiftLeft(num, 10);
+    return util::CheckedShiftLeft(num, 10);
   } else if (util::EqualICase(rest, "m")) {
-    return CheckedShiftLeft(num, 20);
+    return util::CheckedShiftLeft(num, 20);
   } else if (util::EqualICase(rest, "g")) {
-    return CheckedShiftLeft(num, 30);
+    return util::CheckedShiftLeft(num, 30);
   } else if (util::EqualICase(rest, "t")) {
-    return CheckedShiftLeft(num, 40);
+    return util::CheckedShiftLeft(num, 40);
   } else if (util::EqualICase(rest, "p")) {
-    return CheckedShiftLeft(num, 50);
+    return util::CheckedShiftLeft(num, 50);
   }
 
   return {Status::NotOK, "encounter unexpected unit"};

src/config/config.cc

@@ -25,6 +25,7 @@
 #include <strings.h>
 
 #include <algorithm>
+#include <cstdint>
 #include <cstring>
 #include <fstream>
 #include <iostream>
@@ -187,7 +188,8 @@ Config::Config() {
       {"redis-cursor-compatible", false, new YesNoField(&redis_cursor_compatible, true)},
       {"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)},
+      {"proto-max-bulk-len", false,
+       new IntWithUnitField<uint64_t>(&proto_max_bulk_len, std::to_string(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)},
@@ -885,7 +887,7 @@ Status Config::Set(Server *srv, std::string key, const std::string &value) {
     if (!s.IsOK()) return s.Prefixed("invalid value");
   }
 
-  auto origin_value = field->ToString();
+  auto origin_value = field->ToStringForRewrite();
   auto s = field->Set(value);
   if (!s.IsOK()) return s.Prefixed("failed to set new value");
 
@@ -922,7 +924,7 @@ Status Config::Rewrite(const std::map<std::string, std::string> &tokens) {
       // so skip it here to avoid rewriting it as new item.
       continue;
     }
-    new_config[iter.first] = iter.second->ToString();
+    new_config[iter.first] = iter.second->ToStringForRewrite();
   }
 
   std::string namespace_prefix = "namespace.";

src/config/config_type.h

@@ -29,6 +29,7 @@
 #include <string>
 #include <utility>
 
+#include "bit_util.h"
 #include "parse_util.h"
 #include "status.h"
 #include "string_util.h"
@@ -62,6 +63,7 @@ class ConfigField {
   virtual ~ConfigField() = default;
   virtual std::string Default() const = 0;
   virtual std::string ToString() const = 0;
+  virtual std::string ToStringForRewrite() const { return ToString(); }
   virtual Status Set(const std::string &v) = 0;
   virtual Status ToNumber(int64_t *n) const { return {Status::NotOK, "not supported"}; }
   virtual Status ToBool(bool *b) const { return {Status::NotOK, "not supported"}; }
@@ -127,6 +129,7 @@ class IntegerField : public ConfigField {
  public:
   IntegerField(IntegerType *receiver, IntegerType n, IntegerType min, IntegerType max)
       : receiver_(receiver), default_(n), min_(min), max_(max) {
+    CHECK(min <= n && n <= max);
     *receiver_ = n;
   }
   ~IntegerField() override = default;
@@ -251,3 +254,92 @@ class EnumField : public ConfigField {
     return {};
   }
 };
+
+enum class IntUnit { None = 0, K = 10, M = 20, G = 30, T = 40, P = 50 };
+
+template <typename T>
+class IntWithUnitField : public ConfigField {
+ public:
+  IntWithUnitField(T *receiver, std::string default_val, T min, T max)
+      : receiver_(receiver), default_val_(std::move(default_val)), min_(min), max_(max) {
+    CHECK(ReadFrom(default_val_));
+  }
+
+  std::string Default() const override { return default_val_; }
+  std::string ToString() const override { return std::to_string(*receiver_); }
+  std::string ToStringForRewrite() const override { return ToString(*receiver_, current_unit_); }
+
+  Status ToNumber(int64_t *n) const override {
+    *n = static_cast<int64_t>(*receiver_);
+    return Status::OK();
+  }
+
+  Status Set(const std::string &v) override { return ReadFrom(v); }
+
+  Status ReadFrom(const std::string &val) {
+    auto [num, rest] = GET_OR_RET(TryParseInt<T>(val.c_str(), 10));
+
+    if (*rest == 0) {
+      *receiver_ = num;
+      current_unit_ = IntUnit::None;
+    } else if (util::EqualICase(rest, "k")) {
+      *receiver_ = GET_OR_RET(util::CheckedShiftLeft(num, 10));
+      current_unit_ = IntUnit::K;
+    } else if (util::EqualICase(rest, "m")) {
+      *receiver_ = GET_OR_RET(util::CheckedShiftLeft(num, 20));
+      current_unit_ = IntUnit::M;
+    } else if (util::EqualICase(rest, "g")) {
+      *receiver_ = GET_OR_RET(util::CheckedShiftLeft(num, 30));
+      current_unit_ = IntUnit::G;
+    } else if (util::EqualICase(rest, "t")) {
+      *receiver_ = GET_OR_RET(util::CheckedShiftLeft(num, 40));
+      current_unit_ = IntUnit::T;
+    } else if (util::EqualICase(rest, "p")) {
+      *receiver_ = GET_OR_RET(util::CheckedShiftLeft(num, 50));
+      current_unit_ = IntUnit::P;
+    } else {
+      return {Status::NotOK, fmt::format("encounter unexpected unit: `{}`", rest)};
+    }
+
+    if (min_ > *receiver_ || max_ < *receiver_) {
+      return {Status::NotOK, fmt::format("this config value should be between {} and {}", min_, max_)};
+    }
+
+    return Status::OK();
+  }
+
+  static std::string ToString(T val, IntUnit current_unit) {
+    std::string unit_str;
+
+    switch (current_unit) {
+      case IntUnit::None:
+        unit_str = "";
+        break;
+      case IntUnit::K:
+        unit_str = "K";
+        break;
+      case IntUnit::M:
+        unit_str = "M";
+        break;
+      case IntUnit::G:
+        unit_str = "G";
+        break;
+      case IntUnit::T:
+        unit_str = "T";
+        break;
+      case IntUnit::P:
+        unit_str = "P";
+        break;
+    }
+
+    return std::to_string(val >> int(current_unit)) + unit_str;
+  }
+
+ private:
+  // NOTE: the receiver here will get the converted integer (e.g. "10k" -> get 10240)
+  T *receiver_;
+  std::string default_val_;
+  T min_;
+  T max_;
+  IntUnit current_unit_ = IntUnit::None;
+};

src/types/redis_stream.cc

@@ -1759,7 +1759,8 @@ rocksdb::Status Stream::GetPendingEntries(StreamPendingOptions &options, StreamG
     }
 
     if (options.with_count) {
-      ext_results.push_back({entry_id, pel_entry.last_delivery_time_ms, pel_entry.last_delivery_count, consumer_name});
+      ext_results.push_back(
+          {entry_id, {pel_entry.last_delivery_time_ms, pel_entry.last_delivery_count, consumer_name}});
       ext_result_count++;
       continue;
     }

tests/gocase/unit/config/config_test.go

@@ -266,6 +266,12 @@ func TestChangeProtoMaxBulkLen(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, "2097152", vals["proto-max-bulk-len"])
 
+	// change to 100MB
+	require.NoError(t, rdb.ConfigSet(ctx, "proto-max-bulk-len", "100M").Err())
+	vals, err = rdb.ConfigGet(ctx, "proto-max-bulk-len").Result()
+	require.NoError(t, err)
+	require.EqualValues(t, "104857600", vals["proto-max-bulk-len"])
+
 	// Must be >= 1MB
 	require.Error(t, rdb.ConfigSet(ctx, "proto-max-bulk-len", "1024").Err())
 }