Merge branch 'unstable' into feat/command_exists_optimization

Author: cyningsun

cmake/fmt.cmake

@@ -20,8 +20,8 @@ include_guard()
 include(cmake/utils.cmake)
 
 FetchContent_DeclareGitHubWithMirror(fmt
-  fmtlib/fmt 11.2.0
-  MD5=feeba3828e393f7dec473052bf0eef97
+  fmtlib/fmt 12.0.0
+  MD5=3fa90363bce77d4ab9c229d4f6757fdf
 )
 
 FetchContent_MakeAvailableWithArgs(fmt)

cmake/jsoncons.cmake

@@ -20,8 +20,8 @@ include_guard()
 include(cmake/utils.cmake)
 
 FetchContent_DeclareGitHubWithMirror(jsoncons
-  danielaparker/jsoncons v1.4.0
-  MD5=9288495ca2798082e1e96c8b824e9331
+  danielaparker/jsoncons v1.4.1
+  MD5=6e860305aa0aaa1ca0066e23e42acda4
 )
 
 FetchContent_MakeAvailableWithArgs(jsoncons

kvrocks.conf

@@ -1124,5 +1124,56 @@ rocksdb.max_compaction_bytes 0
 # Default: 0
 rocksdb.sst_file_delete_rate_bytes_per_sec 0
 
+# Enable RocksDB periodic compaction to force full compaction of SST files older than the specified time (in seconds).
+# If a compaction filter is registered, it will be applied during these compactions. 
+# Set to 0 to disable this feature.
+#
+# Default: 18446744073709551614 (0xFFFFFFFFFFFFFFFE, UINT64_MAX - 1), a special value indicating RocksDB-controlled behavior.
+# Currently, RocksDB interprets this default as 30 days (2592000 seconds).
+#
+# Typical use cases:
+# - Enforcing data cleanup via compaction filters (e.g., TTL expiration)
+# - Automatically refreshing data encoding/compression formats without manual intervention
+#
+# Reference: https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#periodic-compaction
+#
+# rocksdb.periodic_compaction_seconds 2592000
+
+# Enable RocksDB Time-to-Live (TTL) to automatically schedule compaction for SST files containing expired data.
+# - Files containing data older than the TTL (in seconds) will be prioritized for background compaction.
+# - Requires a registered compaction filter (e.g., TTL filter) to identify and remove expired entries.
+# - Set to 0 to disable TTL-based compaction.
+#
+# Default: 18446744073709551614 (0xFFFFFFFFFFFFFFFE, UINT64_MAX - 1), delegating control to RocksDB.
+# Current RocksDB behavior interprets this default as 30 days (2592000 seconds).
+#
+# Use cases:
+# - Automatic expiration of ephemeral data (e.g., session tokens, temporary logs)
+# - Lifecycle management for time-series datasets
+#
+# Reference: https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#ttl
+#
+# rocksdb.ttl 2592000
+
+# Schedule RocksDB periodic compactions during daily off-peak windows to reduce operational impact.
+# 
+# Requirements:
+# - Periodic compaction must be enabled (`periodic-compaction-seconds > 0`)
+# - Time format: "HH:MM-HH:MM" in UTC (e.g., "02:00-04:30" for a 2.5-hour window)
+# - Empty string disables off-peak scheduling
+#
+# Behavior:
+# - RocksDB proactively triggers periodic compactions during the specified off-peak window
+# - Compactions are optimized to complete before the next peak period begins
+#
+# Default: "" (disabled)
+#
+# Typical use cases:
+# - Minimize compaction I/O during business hours for latency-sensitive workloads
+# - Align resource-heavy operations with maintenance windows
+#
+# Reference: https://github.com/facebook/rocksdb/wiki/Daily-Off%E2%80%90peak-Time-Option
+rocksdb.daily_offpeak_time_utc ""
+
 ################################ NAMESPACE #####################################
 # namespace.test change.me

src/commands/cmd_server.cc

@@ -1586,7 +1586,7 @@ REDIS_REGISTER_COMMANDS(
     MakeCmdAttr<CommandSlaveOf>("replicaof", 3, "read-only exclusive no-script admin", NO_KEY),
     MakeCmdAttr<CommandStats>("stats", 1, "read-only", NO_KEY),
     MakeCmdAttr<CommandRdb>("rdb", -3, "write exclusive admin", NO_KEY),
-    MakeCmdAttr<CommandReset>("reset", 1, "ok-loading bypass-multi no-script", NO_KEY),
+    MakeCmdAttr<CommandReset>("reset", 1, "ok-loading bypass-multi no-script admin", NO_KEY),
     MakeCmdAttr<CommandApplyBatch>("applybatch", -2, "write no-multi", NO_KEY),
     MakeCmdAttr<CommandDump>("dump", 2, "read-only", 1, 1, 1),
     MakeCmdAttr<CommandPollUpdates>("pollupdates", -2, "read-only admin", NO_KEY),

src/commands/cmd_stream.cc

@@ -932,12 +932,13 @@ class CommandXPending : public Commander {
 
   static Status SendExtResults([[maybe_unused]] Connection *conn, std::string *output,
                                std::vector<StreamNACK> &ext_results) {
+    auto now = util::GetTimeStampMS();
     output->append(redis::MultiLen(ext_results.size()));
     for (const auto &entry : ext_results) {
       output->append(redis::MultiLen(4));
       output->append(redis::BulkString(entry.id.ToString()));
       output->append(redis::BulkString(entry.pel_entry.consumer_name));
-      output->append(redis::Integer(entry.pel_entry.last_delivery_time_ms));
+      output->append(redis::Integer(now - entry.pel_entry.last_delivery_time_ms));
       output->append(redis::Integer(entry.pel_entry.last_delivery_count));
     }
 

src/commands/cmd_timeseries.cc

@@ -217,6 +217,15 @@ class KeywordCommandBase : public Commander {
 };
 
 class CommandTSCreateBase : public KeywordCommandBase {
+ public:
+  Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, [[maybe_unused]] Connection *conn,
+                 [[maybe_unused]] std::string *output) override {
+    if (srv->GetConfig()->cluster_enabled && getCreateOption().labels.size()) {
+      return {Status::RedisExecErr, "Specifying LABELS is not supported in cluster mode"};
+    }
+    return Status::OK();
+  }
+
  protected:
   const TSCreateOption &getCreateOption() const { return create_option_; }
 
@@ -308,6 +317,9 @@ class CommandTSCreate : public CommandTSCreateBase {
     return CommandTSCreateBase::Parse(args);
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    auto sc = CommandTSCreateBase::Execute(ctx, srv, conn, output);
+    if (!sc.IsOK()) return sc;
+
     auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
     auto s = timeseries_db.Create(ctx, args_[1], getCreateOption());
     if (!s.ok() && s.IsInvalidArgument()) return {Status::RedisExecErr, errKeyAlreadyExists};
@@ -387,6 +399,9 @@ class CommandTSAdd : public CommandTSCreateBase {
     return CommandTSCreateBase::Parse(args);
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    auto sc = CommandTSCreateBase::Execute(ctx, srv, conn, output);
+    if (!sc.IsOK()) return sc;
+
     auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
     const auto &option = getCreateOption();
 
@@ -851,6 +866,9 @@ class CommandTSMGet : public CommandTSMGetBase {
     return CommandTSMGetBase::Parse(args);
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    if (srv->GetConfig()->cluster_enabled) {
+      return {Status::RedisExecErr, "TS.MGet is not supported in cluster mode"};
+    }
     auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
     std::vector<TSMGetResult> results;
     auto s = timeseries_db.MGet(ctx, getMGetOption(), is_return_latest_, &results);
@@ -898,6 +916,9 @@ class CommandTSMRange : public CommandTSRangeBase, public CommandTSMGetBase {
     return Status::OK();
   }
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    if (srv->GetConfig()->cluster_enabled) {
+      return {Status::RedisExecErr, "TS.MRANGE is not supported in cluster mode"};
+    }
     auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
     std::vector<TSMRangeResult> results;
     auto s = timeseries_db.MRange(ctx, option_, &results);
@@ -978,6 +999,117 @@ class CommandTSMRange : public CommandTSRangeBase, public CommandTSMGetBase {
   TSMRangeOption option_;
 };
 
+class CommandTSIncrByDecrBy : public CommandTSCreateBase {
+ public:
+  CommandTSIncrByDecrBy() { registerDefaultHandlers(); }
+  Status Parse(const std::vector<std::string> &args) override {
+    CommandParser parser(args, 2);
+    auto value_parse = parser.TakeFloat<double>();
+    if (!value_parse.IsOK()) {
+      return {Status::RedisParseErr, errInvalidValue};
+    }
+    value_ = value_parse.GetValue();
+    if (util::ToUpper(args[0]) == "TS.DECRBY") {
+      value_ = -value_;
+    }
+    CommandTSCreateBase::setSkipNum(3);
+    return CommandTSCreateBase::Parse(args);
+  }
+  Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    auto sc = CommandTSCreateBase::Execute(ctx, srv, conn, output);
+    if (!sc.IsOK()) return sc;
+
+    auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
+    const auto &option = getCreateOption();
+
+    if (!is_ts_set_) {
+      // TODO: Should modify function `Add` and `IncrBy` to add a sample with current time
+    }
+    TSChunk::AddResult res;
+    auto s = timeseries_db.IncrBy(ctx, args_[1], {ts_, value_}, option, &res);
+    if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+
+    if (res.type == TSChunk::AddResultType::kOld) {
+      *output +=
+          redis::Error({Status::NotOK, "timestamp must be equal to or higher than the maximum existing timestamp"});
+    } else {
+      *output += FormatAddResultAsRedisReply(res);
+    }
+    return Status::OK();
+  }
+
+ protected:
+  void registerDefaultHandlers() override {
+    CommandTSCreateBase::registerDefaultHandlers();
+    registerHandler("TIMESTAMP", [this](TSOptionsParser &parser) {
+      auto s = handleTimeStamp(parser, ts_);
+      if (!s.IsOK()) return s;
+      is_ts_set_ = true;
+      return Status::OK();
+    });
+  }
+  static Status handleTimeStamp(TSOptionsParser &parser, uint64_t &ts) {
+    auto parse_timestamp = parser.TakeInt<uint64_t>();
+    if (!parse_timestamp.IsOK()) {
+      return {Status::RedisParseErr, errInvalidTimestamp};
+    }
+    ts = parse_timestamp.GetValue();
+    return Status::OK();
+  }
+
+ private:
+  bool is_ts_set_ = false;
+  uint64_t ts_ = 0;
+  double value_ = 0;
+};
+
+class CommandTSDel : public Commander {
+ public:
+  Status Parse(const std::vector<std::string> &args) override {
+    if (args.size() < 4) {
+      return {Status::RedisParseErr, "wrong number of arguments for 'ts.del' command"};
+    }
+    CommandParser parser(args, 2);
+    // Parse start timestamp
+    auto start_parse = parser.TakeInt<uint64_t>();
+    if (!start_parse.IsOK()) {
+      auto start_ts_str = parser.TakeStr();
+      if (!start_ts_str.IsOK() || start_ts_str.GetValue() != "-") {
+        return {Status::RedisParseErr, "wrong fromTimestamp"};
+      }
+      // "-" means use default start timestamp: 0
+    } else {
+      start_ts_ = start_parse.GetValue();
+    }
+    // Parse end timestamp
+    auto end_parse = parser.TakeInt<uint64_t>();
+    if (!end_parse.IsOK()) {
+      auto end_ts_str = parser.TakeStr();
+      if (!end_ts_str.IsOK() || end_ts_str.GetValue() != "+") {
+        return {Status::RedisParseErr, "wrong toTimestamp"};
+      }
+      // "+" means use default end timestamp: MAX_TIMESTAMP
+    } else {
+      end_ts_ = end_parse.GetValue();
+    }
+    return Commander::Parse(args);
+  }
+  Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    auto timeseries_db = TimeSeries(srv->storage, conn->GetNamespace());
+    uint64_t deleted_count = 0;
+    auto s = timeseries_db.Del(ctx, args_[1], start_ts_, end_ts_, &deleted_count);
+    if (!s.ok()) {
+      return {Status::RedisExecErr, s.ToString()};
+    }
+    *output = redis::Integer(deleted_count);
+    return Status::OK();
+  }
+
+ private:
+  uint64_t start_ts_ = 0;
+  uint64_t end_ts_ = TSSample::MAX_TIMESTAMP;
+};
+
 REDIS_REGISTER_COMMANDS(Timeseries, MakeCmdAttr<CommandTSCreate>("ts.create", -2, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTSAdd>("ts.add", -4, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTSMAdd>("ts.madd", -4, "write", 1, -3, 1),
@@ -986,6 +1118,9 @@ REDIS_REGISTER_COMMANDS(Timeseries, MakeCmdAttr<CommandTSCreate>("ts.create", -2
                         MakeCmdAttr<CommandTSGet>("ts.get", -2, "read-only", 1, 1, 1),
                         MakeCmdAttr<CommandTSCreateRule>("ts.createrule", -6, "write", 1, 2, 1),
                         MakeCmdAttr<CommandTSMGet>("ts.mget", -3, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandTSMRange>("ts.mrange", -5, "read-only", NO_KEY), );
+                        MakeCmdAttr<CommandTSMRange>("ts.mrange", -5, "read-only", NO_KEY),
+                        MakeCmdAttr<CommandTSIncrByDecrBy>("ts.incrby", -3, "write", 1, 1, 1),
+                        MakeCmdAttr<CommandTSIncrByDecrBy>("ts.decrby", -3, "write", 1, 1, 1),
+                        MakeCmdAttr<CommandTSDel>("ts.del", -4, "write", 1, 1, 1), );
 
 }  // namespace redis

src/config/config.cc

@@ -307,6 +307,10 @@ Config::Config() {
       {"rocksdb.max_compaction_bytes", false, new Int64Field(&rocks_db.max_compaction_bytes, 0, 0, INT64_MAX)},
       {"rocksdb.sst_file_delete_rate_bytes_per_sec", false,
        new Int64Field(&rocks_db.sst_file_delete_rate_bytes_per_sec, 0, 0, INT64_MAX)},
+      {"rocksdb.periodic_compaction_seconds", false,
+       new UInt64Field(&rocks_db.periodic_compaction_seconds, kDefaultRocksdbPeriodicCompactionSeconds, 0, UINT64_MAX)},
+      {"rocksdb.ttl", false, new UInt64Field(&rocks_db.ttl, kDefaultRocksdbTTL, 0, UINT64_MAX)},
+      {"rocksdb.daily_offpeak_time_utc", false, new StringField(&rocks_db.daily_offpeak_time_utc, "")},
 
       /* rocksdb write options */
       {"rocksdb.write_options.sync", true, new YesNoField(&rocks_db.write_options.sync, false)},
@@ -732,6 +736,9 @@ void Config::initFieldCallback() {
              srv->storage->SetSstFileDeleteRateBytesPerSecond(rocks_db.sst_file_delete_rate_bytes_per_sec);
              return Status::OK();
            }},
+          {"rocksdb.periodic_compaction_seconds", set_cf_option_cb},
+          {"rocksdb.ttl", set_cf_option_cb},
+          {"rocksdb.daily_offpeak_time_utc", set_db_option_cb},
           {"rocksdb.level0_slowdown_writes_trigger",
            [this, &set_cf_option_cb](Server *srv, const std::string &k,
                                      [[maybe_unused]] const std::string &v) -> Status {

src/config/config.h

@@ -57,6 +57,9 @@ constexpr const uint32_t kDefaultPort = 6666;
 constexpr const char *kDefaultNamespace = "__namespace";
 constexpr int KVROCKS_MAX_LSM_LEVEL = 7;
 
+constexpr const uint64_t kDefaultRocksdbTTL = UINT64_MAX - 1;
+constexpr const uint64_t kDefaultRocksdbPeriodicCompactionSeconds = UINT64_MAX - 1;
+
 const std::vector<ConfigEnum<spdlog::level::level_enum>> log_levels{
     {"debug", spdlog::level::debug}, {"info", spdlog::level::info},      {"warning", spdlog::level::warn},
     {"error", spdlog::level::err},   {"fatal", spdlog::level::critical},
@@ -243,6 +246,9 @@ struct Config {
     bool partition_filters;
     int64_t max_compaction_bytes;
     int64_t sst_file_delete_rate_bytes_per_sec = 0;
+    uint64_t periodic_compaction_seconds = kDefaultRocksdbPeriodicCompactionSeconds;
+    uint64_t ttl = kDefaultRocksdbTTL;
+    std::string daily_offpeak_time_utc;
 
     struct WriteOptions {
       bool sync;

src/server/server.cc

@@ -1024,6 +1024,26 @@ Server::InfoEntries Server::GetRocksDBInfo() {
                          cf_stats_map["memtable-limit-delays"]);
     entries.emplace_back("memtable_count_limit_stop[" + cf_handle->GetName() + "]",
                          cf_stats_map["memtable-limit-stops"]);
+
+    // Get the SST file count in all levels
+    std::string sst_file_at_level = "[";
+    for (int level = 0; level < KVROCKS_MAX_LSM_LEVEL; level++) {
+      std::string sst_file_count;
+      db->GetProperty(cf_handle, rocksdb::DB::Properties::kNumFilesAtLevelPrefix + std::to_string(level),
+                      &sst_file_count);
+      if (level != 0) {
+        sst_file_at_level += ",";
+      }
+      sst_file_at_level += sst_file_count;
+    }
+    entries.emplace_back("num_files_at_level[" + cf_handle->GetName() + "]", sst_file_at_level + "]");
+
+    // Get the estimate pending compaction bytes for the current column family
+    std::string estimate_pending_compaction_bytes;
+    db->GetProperty(cf_handle, rocksdb::DB::Properties::kEstimatePendingCompactionBytes,
+                    &estimate_pending_compaction_bytes);
+    entries.emplace_back("estimate_pending_compaction_bytes[" + cf_handle->GetName() + "]",
+                         estimate_pending_compaction_bytes);
   }
 
   auto rocksdb_stats = storage->GetDB()->GetDBOptions().statistics;