diff --git a/src/base/test/value_manager_test.cpp b/src/base/test/value_manager_test.cpp
index c8cc5ed8ab..2474ffd12d 100644
--- a/src/base/test/value_manager_test.cpp
+++ b/src/base/test/value_manager_test.cpp
@@ -56,3 +56,30 @@ TEST(value_schema_manager, get_value_schema)
         }
     }
 }
+
+TEST(pegasus_value_manager, get_value_schema)
+{
+    struct test_case
+    {
+        uint32_t meta_store_data_version;
+        uint32_t value_schema_version;
+        data_version expect_version;
+    } tests[] = {
+        {0, 0, pegasus::data_version::VERSION_0},
+        {1, 0, pegasus::data_version::VERSION_1},
+        {0, 1, pegasus::data_version::VERSION_0},
+        {1, 1, pegasus::data_version::VERSION_1},
+        {0, 2, pegasus::data_version::VERSION_2},
+        {1, 2, pegasus::data_version::VERSION_2},
+    };
+
+    for (const auto &t : tests) {
+        auto generate_schema =
+            value_schema_manager::instance().get_value_schema(t.value_schema_version);
+        std::string raw_value = generate_value(generate_schema, 0, 0, "");
+
+        auto schema =
+            value_schema_manager::instance().get_value_schema(t.meta_store_data_version, raw_value);
+        ASSERT_EQ(t.expect_version, schema->version());
+    }
+}
diff --git a/src/base/value_schema_manager.cpp b/src/base/value_schema_manager.cpp
index c6f62c5fde..23d53cdf32 100644
--- a/src/base/value_schema_manager.cpp
+++ b/src/base/value_schema_manager.cpp
@@ -42,8 +42,25 @@ void value_schema_manager::register_schema(std::unique_ptr<value_schema> schema)
 value_schema *value_schema_manager::get_value_schema(uint32_t meta_cf_data_version,
                                                      dsn::string_view value) const
 {
-    /// TBD(zlw)
-    return nullptr;
+    dsn::data_input input(value);
+    uint8_t first_byte = input.read_u8();
+    // first bit = 1 means the data version is >= VERSION_2
+    if (first_byte & 0x80) {
+        // In order to keep backward compatibility, we should return latest version if the data
+        // version in value is not found. In other words, it will work well in future version if it
+        // is compatible with latest version in current
+        auto schema = get_value_schema(first_byte & 0x7F);
+        if (nullptr == schema) {
+            return get_latest_value_schema();
+        }
+        return schema;
+    } else {
+        auto schema = get_value_schema(meta_cf_data_version);
+        if (nullptr == schema) {
+            dassert_f(false, "data version({}) in meta cf is not supported", meta_cf_data_version);
+        }
+        return schema;
+    }
 }
 
 value_schema *value_schema_manager::get_value_schema(uint32_t version) const
diff --git a/src/base/value_schema_v2.cpp b/src/base/value_schema_v2.cpp
index 0f94cb4684..7b5ea9b9ec 100644
--- a/src/base/value_schema_v2.cpp
+++ b/src/base/value_schema_v2.cpp
@@ -46,7 +46,8 @@ std::unique_ptr<value_field> value_schema_v2::extract_field(dsn::string_view val
 dsn::blob value_schema_v2::extract_user_data(std::string &&value)
 {
     auto ret = dsn::blob::create_from_bytes(std::move(value));
-    return ret.range(sizeof(uint8_t) + sizeof(uint32_t) + sizeof(uint64_t));
+    static const auto USER_DATA_OFFSET = sizeof(uint8_t) + sizeof(uint32_t) + sizeof(uint64_t);
+    return ret.range(USER_DATA_OFFSET);
 }
 
 void value_schema_v2::update_field(std::string &value, std::unique_ptr<value_field> field)