feat(ts): add the support of TS.ALTER command (#3264)

var-nan · var-nan · yezhizi · web-flow · commit 2dc7e9ea623b · 2025-11-21T16:04:00.000+08:00
It closes #3214. --------- Co-authored-by: var-nan <krs.nandhan999@protonmail.com> Co-authored-by: DeEMO <yzzxrx@gmail.com>
diff --git a/src/commands/cmd_timeseries.cc b/src/commands/cmd_timeseries.cc
@@ -315,7 +315,6 @@ class CommandTSCreateBase : public KeywordCommandBase {
     return Status::OK();
   }
 
- private:
   TSCreateOption create_option_;
 };
 
@@ -342,6 +341,52 @@ class CommandTSCreate : public CommandTSCreateBase {
   }
 };
 
+class CommandTSAlter : public CommandTSCreateBase {
+ public:
+  CommandTSAlter() { registerDefaultHandlers(); }
+  Status Parse(const std::vector<std::string> &args) override {
+    if (args.size() < 2) {
+      return {Status::RedisParseErr, errWrongNumOfArguments};
+    }
+    CommandTSCreateBase::setSkipNum(2);
+    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.Alter(ctx, args_[1], getCreateOption(), mask_);
+    if (!s.ok() && s.IsInvalidArgument()) return {Status::RedisExecErr, errKeyNotFound};
+    if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+    *output = redis::RESP_OK;
+    return Status::OK();
+  }
+
+ private:
+  uint8_t mask_ = 0;
+
+  void registerDefaultHandlers() override {
+    using AlterMode = std::underlying_type<TSAlterMode>::type;
+    registerHandler("RETENTION", [this](TSOptionsParser &parser) {
+      mask_ |= static_cast<AlterMode>(TSAlterMode::RETENTION);
+      return handleRetention(parser, create_option_.retention_time);
+    });
+    registerHandler("CHUNK_SIZE", [this](TSOptionsParser &parser) {
+      mask_ |= static_cast<AlterMode>(TSAlterMode::CHUNK_SIZE);
+      return handleChunkSize(parser, create_option_.chunk_size);
+    });
+    registerHandler("DUPLICATE_POLICY", [this](TSOptionsParser &parser) {
+      mask_ |= static_cast<AlterMode>(TSAlterMode::DUPLICATE_POLICY);
+      return handleDuplicatePolicy(parser, create_option_.duplicate_policy);
+    });
+    registerHandler("LABELS", [this](TSOptionsParser &parser) {
+      mask_ |= static_cast<AlterMode>(TSAlterMode::LABELS);
+      return handleLabels(parser, create_option_.labels);
+    });
+  }
+};
+
 class CommandTSInfo : public Commander {
  public:
   Status Parse(const std::vector<std::string> &args) override { return Commander::Parse(args); }
@@ -1227,6 +1272,7 @@ class CommandTSQueryIndex : public Commander {
 };
 
 REDIS_REGISTER_COMMANDS(Timeseries, MakeCmdAttr<CommandTSCreate>("ts.create", -2, "write", 1, 1, 1),
+                        MakeCmdAttr<CommandTSAlter>("ts.alter", -2, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTSAdd>("ts.add", -4, "write", 1, 1, 1),
                         MakeCmdAttr<CommandTSMAdd>("ts.madd", -4, "write", 1, -3, 1),
                         MakeCmdAttr<CommandTSRange>("ts.range", -4, "read-only", 1, 1, 1),
diff --git a/src/types/redis_timeseries.cc b/src/types/redis_timeseries.cc
@@ -1860,6 +1860,68 @@ rocksdb::Status TimeSeries::Create(engine::Context &ctx, const Slice &user_key,
   return createTimeSeries(ctx, ns_key, &metadata, &option);
 }
 
+rocksdb::Status TimeSeries::Alter(engine::Context &ctx, const Slice &user_key, const TSCreateOption &option,
+                                  uint8_t mask) {
+  std::string ns_key = AppendNamespacePrefix(user_key);
+  TimeSeriesMetadata metadata;
+  if (auto s = getTimeSeriesMetadata(ctx, ns_key, &metadata); !s.ok()) {
+    return rocksdb::Status::InvalidArgument("key not exists");
+  }
+  auto batch = storage_->GetWriteBatchBase();
+  WriteBatchLogData log_data(kRedisTimeSeries, {"Alter"});
+  if (auto s = batch->PutLogData(log_data.Encode()); !s.ok()) return s;
+
+  using AlterMode = std::underlying_type<TSAlterMode>::type;
+  bool update_retention = mask & static_cast<AlterMode>(TSAlterMode::RETENTION);
+  bool update_chunk_size = mask & static_cast<AlterMode>(TSAlterMode::CHUNK_SIZE);
+  bool update_duplicate_policy = mask & static_cast<AlterMode>(TSAlterMode::DUPLICATE_POLICY);
+  bool update_labels = mask & static_cast<AlterMode>(TSAlterMode::LABELS);
+
+  if (update_retention || update_chunk_size || update_duplicate_policy) {
+    metadata.retention_time = update_retention ? option.retention_time : metadata.retention_time;
+    metadata.chunk_size = update_chunk_size ? option.chunk_size : metadata.chunk_size;
+    metadata.duplicate_policy = update_duplicate_policy ? option.duplicate_policy : metadata.duplicate_policy;
+    std::string bytes;
+    metadata.Encode(&bytes);
+    if (auto s = batch->Put(metadata_cf_handle_, ns_key, bytes); !s.ok()) return s;
+  }
+  if (update_labels) {
+    LabelKVList prev_labels;
+    if (auto s = getLabelKVList(ctx, ns_key, metadata, &prev_labels); !s.ok()) return s;
+
+    std::unordered_map<std::string, std::pair<std::string, bool>> labels_map;  // True : (key,val) should be updated
+    for (auto &label : option.labels) {
+      labels_map.insert(std::make_pair(label.k, std::make_pair(label.v, true)));
+    }
+    for (auto &label : prev_labels) {
+      if (auto it = labels_map.find(label.k); it != labels_map.end()) {
+        it->second.second = it->second.first != label.v;
+        if (it->second.second) {  // Remove reverse index key. Val is updated later.
+          auto rev_index_key = TSRevLabelKey(namespace_, label.k, label.v, user_key).Encode();
+          if (auto s = batch->Delete(index_cf_handle_, rev_index_key); !s.ok()) return s;
+        }  // Else key-val pair unchanged
+      } else {  // Remove label.
+        auto internal_key = internalKeyFromLabelKey(ns_key, metadata, label.k);
+        if (auto s = batch->Delete(internal_key); !s.ok()) return s;
+
+        auto rev_index_key = TSRevLabelKey(namespace_, label.k, label.v, user_key).Encode();
+        if (auto s = batch->Delete(index_cf_handle_, rev_index_key); !s.ok()) return s;
+      }
+    }
+
+    for (auto &[k, v] : labels_map) {
+      if (v.second) {  // Update label and insert reverse-index
+        auto internal_key = internalKeyFromLabelKey(ns_key, metadata, k);
+        if (auto s = batch->Put(internal_key, v.first); !s.ok()) return s;
+
+        auto rev_index_key = TSRevLabelKey(namespace_, k, v.first, user_key).Encode();
+        if (auto s = batch->Put(index_cf_handle_, rev_index_key, Slice()); !s.ok()) return s;
+      }
+    }
+  }
+  return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
 rocksdb::Status TimeSeries::Add(engine::Context &ctx, const Slice &user_key, TSSample sample,
                                 const TSCreateOption &option, AddResult *res, const DuplicatePolicy *on_dup_policy) {
   std::string ns_key = AppendNamespacePrefix(user_key);
diff --git a/src/types/redis_timeseries.h b/src/types/redis_timeseries.h
@@ -253,6 +253,14 @@ enum class TSCreateRuleResult : uint8_t {
   kSrcEqDst = 6,
 };
 
+enum class TSAlterMode : uint8_t {
+  RETENTION = 1,
+  CHUNK_SIZE = 1 << 1,
+  DUPLICATE_POLICY = 1 << 2,
+  IGNORE = 1 << 3,
+  LABELS = 1 << 4,
+};
+
 std::vector<TSSample> GroupSamplesAndReduce(const std::vector<std::vector<TSSample>> &all_samples,
                                             TSMRangeOption::GroupReducerType reducer_type);
 
@@ -267,6 +275,7 @@ class TimeSeries : public SubKeyScanner {
   TimeSeries(engine::Storage *storage, const std::string &ns)
       : SubKeyScanner(storage, ns), index_cf_handle_(storage->GetCFHandle(ColumnFamilyID::Index)) {}
   rocksdb::Status Create(engine::Context &ctx, const Slice &user_key, const TSCreateOption &option);
+  rocksdb::Status Alter(engine::Context &ctx, const Slice &user_key, const TSCreateOption &option, uint8_t mask);
   rocksdb::Status Add(engine::Context &ctx, const Slice &user_key, TSSample sample, const TSCreateOption &option,
                       AddResult *res, const DuplicatePolicy *on_dup_policy = nullptr);
   rocksdb::Status MAdd(engine::Context &ctx, const Slice &user_key, std::vector<TSSample> samples,
diff --git a/tests/cppunit/types/timeseries_test.cc b/tests/cppunit/types/timeseries_test.cc
@@ -53,6 +53,82 @@ TEST_F(TimeSeriesTest, Create) {
   EXPECT_EQ(s.ToString(), "Invalid argument: key already exists");
 }
 
+TEST_F(TimeSeriesTest, Alter) {
+  redis::TSCreateOption option;
+  option.retention_time = 3600;
+  option.chunk_size = 1024;
+  option.labels = {{"type", "runtime"}, {"compiler", "gcc"}, {"machine", "Linux"}};
+  key_ = "pa";
+
+  EXPECT_TRUE(ts_db_->Create(*ctx_, key_, option).ok());
+
+  redis::TSInfoResult res;
+  option.retention_time = 200;
+  EXPECT_TRUE(ts_db_->Alter(*ctx_, key_, option, static_cast<uint8_t>(redis::TSAlterMode::RETENTION)).ok());
+  EXPECT_TRUE(ts_db_->Info(*ctx_, key_, &res).ok());
+  EXPECT_EQ(res.metadata.retention_time, 200);
+  EXPECT_EQ(res.metadata.chunk_size, 1024);
+  EXPECT_EQ(res.labels.size(), 3);
+
+  // Update chunk size and verify other fields are unaffected.
+  option.chunk_size = 128;
+  EXPECT_TRUE(ts_db_->Alter(*ctx_, key_, option, static_cast<uint8_t>(redis::TSAlterMode::CHUNK_SIZE)).ok());
+  ts_db_->Info(*ctx_, key_, &res);
+  EXPECT_EQ(res.metadata.retention_time, 200);
+  EXPECT_EQ(res.metadata.chunk_size, 128);
+  EXPECT_EQ(res.labels.size(), 3);
+
+  // Verify records are properly inserted.
+  std::vector<TSSample> samples = {{10, 23}, {12, 24.5}};
+  std::vector<TSChunk::AddResult> results;
+  results.resize(samples.size());
+  EXPECT_TRUE(ts_db_->MAdd(*ctx_, key_, samples, &results).ok());
+  EXPECT_EQ(results[0].sample.ts, 10);
+  EXPECT_EQ(results[1].sample.ts, 12);
+
+  // Update labels and verify
+  res.labels.clear();
+  option.labels = {{"compiler", "gcc_12"}, {"version", "123"}};
+  EXPECT_TRUE(ts_db_->Alter(*ctx_, key_, option, static_cast<uint8_t>(redis::TSAlterMode::LABELS)).ok());
+  EXPECT_TRUE(ts_db_->Info(*ctx_, key_, &res).ok());
+  EXPECT_EQ(res.metadata.retention_time, 200);
+  EXPECT_EQ(res.metadata.chunk_size, 128);
+  EXPECT_EQ(res.labels.size(), 2);
+  redis::LabelKVPair first = {"version", "123"}, second = {"compiler", "gcc_12"};
+  EXPECT_TRUE(std::find_if(res.labels.begin(), res.labels.end(), [first](const auto &label) {
+                return first.k == label.k && first.v == label.v;
+              }) != res.labels.end());
+  EXPECT_TRUE(std::find_if(res.labels.begin(), res.labels.end(), [second](const auto &label) {
+                return second.k == label.k && second.v == label.v;
+              }) != res.labels.end());
+
+  // Verify reverse-indexes are properly updated.
+  key_ = "pavani";
+  redis::TSCreateOption second_option;
+  second_option.labels = {{"compiler", "gcc_12"}, {"Desktop", "Lubuntu"}};
+  EXPECT_TRUE(ts_db_->Create(*ctx_, key_, second_option).ok());
+
+  redis::TSMGetOption::FilterOption filters;
+  filters.labels_equals = {{"compiler", {"gcc_12"}}};
+  std::vector<std::string> query_res;
+  EXPECT_TRUE(ts_db_->QueryIndex(*ctx_, filters, &query_res).ok());
+  EXPECT_EQ(query_res.size(), 2);
+  EXPECT_TRUE(std::find(query_res.begin(), query_res.end(), "pa") != query_res.end());
+  EXPECT_TRUE(std::find(query_res.begin(), query_res.end(), "pavani") != query_res.end());
+
+  // old labels should be deleted.
+  filters.labels_equals.clear();
+  filters.labels_equals = {{"machine", {"Linux"}}};
+  query_res.clear();
+  EXPECT_TRUE(ts_db_->QueryIndex(*ctx_, filters, &query_res).ok());
+  EXPECT_TRUE(query_res.empty());
+
+  key_ = "pavni";
+  auto s = ts_db_->Alter(*ctx_, key_, option, 1);
+  EXPECT_FALSE(s.ok());
+  EXPECT_EQ(s.ToString(), "Invalid argument: key not exists");
+}
+
 TEST_F(TimeSeriesTest, Add) {
   redis::TSCreateOption option;
   option.retention_time = 3600;
diff --git a/tests/gocase/unit/type/timeseries/timeseries_test.go b/tests/gocase/unit/type/timeseries/timeseries_test.go
@@ -167,6 +167,16 @@ func testTimeSeries(t *testing.T, configs util.KvrocksServerConfigs) {
 		require.Equal(t, int64(2), vals[11])
 	})
 
+	t.Run("TS.ALTER Verify Updates", func(t *testing.T) {
+		require.NoError(t, rdb.Do(ctx, "ts.create", key, "retention", "3600", "chunk_size", "1024",
+			"LABELS", "type", "runtime", "compiler", "gcc", "machine", "gcc").Err())
+
+		require.NoError(t, rdb.Do(ctx, "ts.alter", key, "retention", "200").Err())
+		vals, err := rdb.Do(ctx, "ts.info", key).Slice()
+		require.NoError(t, err)
+		require.Equal(t, int64(200), vals[9])
+	})
+
 	t.Run("TS.ADD Basic Add", func(t *testing.T) {
 		require.NoError(t, rdb.Del(ctx, key).Err())
 		require.NoError(t, rdb.Do(ctx, "ts.create", key).Err())
