refactor: implement value_shcema_v0

src/base/pegasus_value_schema.h

@@ -238,13 +238,24 @@ class pegasus_value_generator
 
 enum data_version
 {
+    VERSION_0 = 0,
+    VERSION_COUNT,
+    VERSION_MAX = VERSION_0,
     /// TBD(zlw)
 };
 
 struct value_params
 {
+    value_params(std::string &buf, std::vector<rocksdb::Slice> &slices)
+        : write_buf(buf), write_slices(slices)
+    {
+    }
+
     std::map<value_field_type, std::unique_ptr<value_field>> fields;
-    /// TBD(zlw)
+    // write_buf and write_slices are transferred from `pegasus_value_generator`, Which are used to
+    // prevent data copy
+    std::string &write_buf;
+    std::vector<rocksdb::Slice> &write_slices;
 };
 
 class value_schema

src/base/test/value_manager_test.cpp

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "value_schema_manager.h"
+#include <gtest/gtest.h>
+
+using namespace pegasus;
+
+extern std::string generate_value(value_schema *schema,
+                                  uint32_t expire_ts,
+                                  uint64_t time_tag,
+                                  dsn::string_view user_data);
+
+TEST(value_schema_manager, get_latest_value_schema)
+{
+    auto schema = value_schema_manager::instance().get_latest_value_schema();
+    ASSERT_EQ(data_version::VERSION_MAX, schema->version());
+}
+
+TEST(value_schema_manager, get_value_schema)
+{
+    struct test_case
+    {
+        uint32_t version;
+        bool schema_exist;
+    } tests[] = {
+        {pegasus::data_version::VERSION_0, true}, {pegasus::data_version::VERSION_MAX + 1, false},
+    };
+
+    for (const auto &t : tests) {
+        auto schema = value_schema_manager::instance().get_value_schema(t.version);
+        if (t.schema_exist) {
+            ASSERT_NE(schema, nullptr);
+            ASSERT_EQ(t.version, schema->version());
+        } else {
+            ASSERT_EQ(schema, nullptr);
+        }
+    }
+}

src/base/test/value_schema_test.cpp

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "base/pegasus_value_schema.h"
+#include "base/value_schema_manager.h"
+
+#include <gtest/gtest.h>
+
+using namespace pegasus;
+
+uint32_t extract_expire_ts(value_schema *schema, const std::string &raw_value)
+{
+    auto segment = schema->extract_field(raw_value, pegasus::value_field_type::EXPIRE_TIMESTAMP);
+    auto expire_ts_segment = static_cast<expire_timestamp_field *>(segment.get());
+    return expire_ts_segment->expire_ts;
+}
+
+uint64_t extract_time_tag(value_schema *schema, const std::string &raw_value)
+{
+    auto segment = schema->extract_field(raw_value, pegasus::value_field_type::TIME_TAG);
+    auto time_tag_segment = static_cast<time_tag_field *>(segment.get());
+    return time_tag_segment->time_tag;
+}
+
+std::string generate_value(value_schema *schema,
+                           uint32_t expire_ts,
+                           uint64_t time_tag,
+                           dsn::string_view user_data)
+{
+    std::string write_buf;
+    std::vector<rocksdb::Slice> write_slices;
+    value_params params{write_buf, write_slices};
+    params.fields[value_field_type::EXPIRE_TIMESTAMP] =
+        dsn::make_unique<expire_timestamp_field>(expire_ts);
+    params.fields[value_field_type::TIME_TAG] = dsn::make_unique<time_tag_field>(time_tag);
+    params.fields[value_field_type::USER_DATA] = dsn::make_unique<user_data_field>(user_data);
+
+    rocksdb::SliceParts sparts = schema->generate_value(params);
+    std::string raw_value;
+    for (int i = 0; i < sparts.num_parts; i++) {
+        raw_value += sparts.parts[i].ToString();
+    }
+    return raw_value;
+}
+
+TEST(value_schema, generate_and_extract)
+{
+    struct test_case
+    {
+        uint32_t data_version;
+        uint32_t expire_ts;
+        uint64_t time_tag;
+        std::string user_data;
+    } tests[] = {
+        {0, 1000, 0, ""},
+        {0, std::numeric_limits<uint32_t>::max(), 0, "pegasus"},
+        {0, std::numeric_limits<uint32_t>::max(), 0, ""},
+        {0, 0, 0, "a"},
+    };
+
+    for (const auto &t : tests) {
+        auto schema = value_schema_manager::instance().get_value_schema(t.data_version);
+        std::string raw_value = generate_value(schema, t.expire_ts, t.time_tag, t.user_data);
+
+        ASSERT_EQ(t.expire_ts, extract_expire_ts(schema, raw_value));
+        if (t.data_version >= 1) {
+            ASSERT_EQ(t.time_tag, extract_time_tag(schema, raw_value));
+        }
+
+        dsn::blob user_data = schema->extract_user_data(std::move(raw_value));
+        ASSERT_EQ(t.user_data, user_data.to_string());
+    }
+}
+
+TEST(value_schema, update_expire_ts)
+{
+    struct test_case
+    {
+        uint32_t data_version;
+        uint32_t expire_ts;
+        uint32_t update_expire_ts;
+    } tests[] = {
+        {0, 1000, 10086},
+    };
+
+    for (const auto &t : tests) {
+        std::string write_buf;
+        std::vector<rocksdb::Slice> write_slices;
+        auto schema = value_schema_manager::instance().get_value_schema(t.data_version);
+        std::string raw_value = generate_value(schema, t.expire_ts, 0, "");
+
+        std::unique_ptr<value_field> segment =
+            dsn::make_unique<expire_timestamp_field>(t.update_expire_ts);
+        schema->update_field(raw_value, std::move(segment));
+        ASSERT_EQ(t.update_expire_ts, extract_expire_ts(schema, raw_value));
+    }
+}

src/base/value_field.h

@@ -23,12 +23,38 @@ namespace pegasus {
 
 enum value_field_type
 {
-    /// TBD(zlw)
+    EXPIRE_TIMESTAMP = 0,
+    TIME_TAG,
+    USER_DATA,
 };
 
 struct value_field
 {
     virtual ~value_field() = default;
     virtual value_field_type type() = 0;
 };
+
+struct expire_timestamp_field : public value_field
+{
+    explicit expire_timestamp_field(uint32_t timestamp) { expire_ts = timestamp; }
+    value_field_type type() { return value_field_type::EXPIRE_TIMESTAMP; };
+
+    uint32_t expire_ts;
+};
+
+struct time_tag_field : public value_field
+{
+    explicit time_tag_field(uint64_t tag) { time_tag = tag; }
+    value_field_type type() { return value_field_type::TIME_TAG; };
+
+    uint64_t time_tag;
+};
+
+struct user_data_field : public value_field
+{
+    explicit user_data_field(dsn::string_view data) { user_data = data; }
+    value_field_type type() { return value_field_type::USER_DATA; };
+
+    dsn::string_view user_data;
+};
 } // namespace pegasus

src/base/value_schema_manager.cpp

@@ -18,12 +18,12 @@
  */
 
 #include "value_schema_manager.h"
+#include "value_schema_v0.h"
 
 namespace pegasus {
-
-void value_schema_manager::register_schema(value_schema *schema)
+void value_schema_manager::register_schema(std::unique_ptr<value_schema> schema)
 {
-    /// TBD(zlw)
+    _schemas[schema->version()] = std::move(schema);
 }
 
 value_schema *value_schema_manager::get_value_schema(uint32_t meta_cf_data_version,
@@ -35,14 +35,15 @@ value_schema *value_schema_manager::get_value_schema(uint32_t meta_cf_data_versi
 
 value_schema *value_schema_manager::get_value_schema(uint32_t version) const
 {
-    /// TBD(zlw)
-    return nullptr;
+    if (version >= _schemas.size()) {
+        return nullptr;
+    }
+    return _schemas[version].get();
 }
 
 value_schema *value_schema_manager::get_latest_value_schema() const
 {
-    /// TBD(zlw)
-    return nullptr;
+    return _schemas.rbegin()->get();
 }
 
 /**
@@ -51,6 +52,7 @@ value_schema *value_schema_manager::get_latest_value_schema() const
  */
 void register_value_schemas()
 {
+    value_schema_manager::instance().register_schema(dsn::make_unique<value_schema_v0>());
     /// TBD(zlw)
 }
 

src/base/value_schema_manager.h

@@ -27,7 +27,7 @@ namespace pegasus {
 class value_schema_manager : public dsn::utils::singleton<value_schema_manager>
 {
 public:
-    void register_schema(value_schema *schema);
+    void register_schema(std::unique_ptr<value_schema> schema);
     /// using the raw value in rocksdb and data version stored in meta column family to get data
     /// version
     value_schema *get_value_schema(uint32_t meta_cf_data_version, dsn::string_view value) const;
@@ -38,6 +38,6 @@ class value_schema_manager : public dsn::utils::singleton<value_schema_manager>
     value_schema_manager() = default;
     friend class dsn::utils::singleton<value_schema_manager>;
 
-    std::array<value_schema *, 2> _schemas;
+    std::array<std::unique_ptr<value_schema>, data_version::VERSION_COUNT> _schemas;
 };
 } // namespace pegasus