feat(ts): Add TS.GET command (#3142)

Part of #3048

Author: yezhizi

src/commands/cmd_timeseries.cc

@@ -782,4 +782,41 @@ rocksdb::Status TimeSeries::Range(engine::Context &ctx, const Slice &user_key, c
   return rocksdb::Status::OK();
 }
 
+rocksdb::Status TimeSeries::Get(engine::Context &ctx, const Slice &user_key, bool is_return_latest,
+                                std::vector<TSSample> *res) {
+  res->clear();
+  std::string ns_key = AppendNamespacePrefix(user_key);
+
+  TimeSeriesMetadata metadata(false);
+  rocksdb::Status s = getTimeSeriesMetadata(ctx, ns_key, &metadata);
+  if (!s.ok()) {
+    return s;
+  }
+
+  // In the emun `TSSubkeyType`, `LABEL` is the next of `CHUNK`
+  std::string chunk_upper_bound = internalKeyFromLabelKey(ns_key, metadata, "");
+  std::string end_key = internalKeyFromChunkID(ns_key, metadata, TSSample::MAX_TIMESTAMP);
+  std::string prefix = end_key.substr(0, end_key.size() - sizeof(uint64_t));
+
+  rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
+  rocksdb::Slice upper_bound(chunk_upper_bound);
+  read_options.iterate_upper_bound = &upper_bound;
+  rocksdb::Slice lower_bound(prefix);
+  read_options.iterate_lower_bound = &lower_bound;
+
+  // Get the latest chunk
+  auto iter = util::UniqueIterator(ctx, read_options);
+  iter->SeekForPrev(end_key);
+  if (!iter->Valid() || !iter->key().starts_with(prefix)) {
+    return rocksdb::Status::OK();
+  }
+  auto chunk = CreateTSChunkFromData(iter->value());
+
+  if (is_return_latest) {
+    // TODO: need process `latest` option
+  }
+  res->push_back(chunk->GetLatestSample(0));
+  return rocksdb::Status::OK();
+}
+
 }  // namespace redis

src/types/redis_timeseries.cc

@@ -168,6 +168,7 @@ class TimeSeries : public SubKeyScanner {
   rocksdb::Status Info(engine::Context &ctx, const Slice &user_key, TSInfoResult *res);
   rocksdb::Status Range(engine::Context &ctx, const Slice &user_key, const TSRangeOption &option,
                         std::vector<TSSample> *res);
+  rocksdb::Status Get(engine::Context &ctx, const Slice &user_key, bool is_return_latest, std::vector<TSSample> *res);
 
  private:
   rocksdb::Status getTimeSeriesMetadata(engine::Context &ctx, const Slice &ns_key, TimeSeriesMetadata *metadata);

src/types/redis_timeseries.h

@@ -527,3 +527,10 @@ std::string UncompTSChunk::UpdateSampleValue(uint64_t ts, double value, bool is_
 
   return new_buffer;
 }
+
+TSSample UncompTSChunk::GetLatestSample(uint32_t idx) const {
+  if (metadata_.count == 0 || idx >= metadata_.count) {
+    unreachable();
+  }
+  return samples_[metadata_.count - 1 - idx];
+}

src/types/timeseries.cc

@@ -184,6 +184,9 @@ class TSChunk {
   // Returns empty string if no changes
   virtual std::string UpdateSampleValue(uint64_t ts, double value, bool is_add_on) const = 0;
 
+  // Get idx-th latest sample, idx=0 means latest sample
+  virtual TSSample GetLatestSample(uint32_t idx) const = 0;
+
  protected:
   nonstd::span<const char> data_;
   MetaData metadata_;
@@ -202,6 +205,7 @@ class UncompTSChunk : public TSChunk {
                                                 bool is_fix_split_mode) const override;
   std::string RemoveSamplesBetween(uint64_t from, uint64_t to) const override;
   std::string UpdateSampleValue(uint64_t ts, double value, bool is_add_on) const override;
+  TSSample GetLatestSample(uint32_t idx) const override;
 
  private:
   nonstd::span<const TSSample> samples_;

src/types/timeseries.h

@@ -332,3 +332,35 @@ TEST_F(TimeSeriesTest, Range) {
   EXPECT_TRUE(s.ok());
   EXPECT_EQ(res.size(), 1);
 }
+
+TEST_F(TimeSeriesTest, Get) {
+  redis::TSCreateOption option;
+  auto s = ts_db_->Create(*ctx_, key_, option);
+  EXPECT_TRUE(s.ok());
+
+  std::vector<TSSample> res;
+  // Test empty timeseries
+  s = ts_db_->Get(*ctx_, key_, false, &res);
+  EXPECT_TRUE(s.ok());
+  EXPECT_EQ(res.size(), 0);
+
+  // Add multiple samples
+  std::vector<TSSample> samples = {{1, 10}, {2, 20}, {3, 30}};
+  std::vector<TSChunk::AddResultWithTS> results;
+  results.resize(samples.size());
+
+  s = ts_db_->MAdd(*ctx_, key_, samples, &results);
+  EXPECT_TRUE(s.ok());
+
+  // Test basic GET (returns latest sample)
+  s = ts_db_->Get(*ctx_, key_, false, &res);
+  EXPECT_TRUE(s.ok());
+  EXPECT_EQ(res.size(), 1);
+  EXPECT_EQ(res[0].ts, 3);
+  EXPECT_EQ(res[0].v, 30);
+
+  // Test GET with empty timeseries
+  std::vector<TSSample> empty_res;
+  s = ts_db_->Get(*ctx_, "nonexistent_key", false, &empty_res);
+  EXPECT_FALSE(s.ok());
+}

tests/cppunit/types/timeseries_test.cc

@@ -408,4 +408,27 @@ func testTimeSeries(t *testing.T, configs util.KvrocksServerConfigs) {
 		res = rdb.Do(ctx, "ts.range", key, "-", "+", "AGGREGATION", "MIN", 20, "COUNT", 1).Val().([]interface{})
 		assert.Equal(t, 1, len(res))
 	})
+
+	t.Run("TS.GET Basic", func(t *testing.T) {
+		key := "test_get_key"
+		require.NoError(t, rdb.Del(ctx, key).Err())
+		require.NoError(t, rdb.Do(ctx, "ts.create", key).Err())
+		// Test GET on empty timeseries
+		res := rdb.Do(ctx, "ts.get", key).Val().([]interface{})
+		require.Equal(t, 0, len(res))
+
+		// Add samples
+		require.Equal(t, int64(1000), rdb.Do(ctx, "ts.add", key, "1000", "12.3").Val())
+		require.Equal(t, int64(2000), rdb.Do(ctx, "ts.add", key, "2000", "15.6").Val())
+
+		// Test basic GET
+		res = rdb.Do(ctx, "ts.get", key).Val().([]interface{})
+		require.Equal(t, 1, len(res))
+		require.Equal(t, int64(2000), res[0].([]interface{})[0])
+		require.Equal(t, 15.6, res[0].([]interface{})[1])
+
+		// Test GET on non-existent key
+		_, err := rdb.Do(ctx, "ts.get", "nonexistent_key").Result()
+		require.ErrorContains(t, err, "key does not exist")
+	})
 }