diff --git a/src/common/IndexKeyUtils.cpp b/src/common/IndexKeyUtils.cpp
index a95d32f72..bd9469262 100644
--- a/src/common/IndexKeyUtils.cpp
+++ b/src/common/IndexKeyUtils.cpp
@@ -8,6 +8,63 @@
 
 namespace nebula {
 
+// static
+void IndexKeyUtils::indexRaw(const IndexValues &values, std::string& raw) {
+    std::vector<int32_t> colsLen;
+    for (auto& col : values) {
+        if (col.first == Value::Type::STRING) {
+            colsLen.emplace_back(col.second.size());
+        }
+        raw.append(col.second.data(), col.second.size());
+    }
+    for (auto len : colsLen) {
+        raw.append(reinterpret_cast<const char*>(&len), sizeof(int32_t));
+    }
+}
+
+// static
+std::string IndexKeyUtils::vertexIndexKey(size_t vIdLen, PartitionID partId,
+                                          IndexID indexId, VertexID vId,
+                                          const IndexValues& values) {
+    int32_t item = (partId << kPartitionOffset) | static_cast<uint32_t>(NebulaKeyType::kIndex);
+    std::string key;
+    key.reserve(256);
+    key.append(reinterpret_cast<const char*>(&item), sizeof(int32_t))
+       .append(reinterpret_cast<const char*>(&indexId), sizeof(IndexID));
+    indexRaw(values, key);
+    key.append(vId.data(), vId.size())
+       .append(vIdLen - vId.size(), '\0');
+    return key;
+}
+
+// static
+std::string IndexKeyUtils::edgeIndexKey(size_t vIdLen, PartitionID partId,
+                                        IndexID indexId, VertexID srcId,
+                                        EdgeRanking rank, VertexID dstId,
+                                        const IndexValues& values) {
+    int32_t item = (partId << kPartitionOffset) | static_cast<uint32_t>(NebulaKeyType::kIndex);
+    std::string key;
+    key.reserve(256);
+    key.append(reinterpret_cast<const char*>(&item), sizeof(int32_t))
+       .append(reinterpret_cast<const char*>(&indexId), sizeof(IndexID));
+    indexRaw(values, key);
+    key.append(srcId.data(), srcId.size())
+       .append(vIdLen - srcId.size(), '\0')
+       .append(reinterpret_cast<const char*>(&rank), sizeof(EdgeRanking))
+       .append(dstId.data(), dstId.size())
+       .append(vIdLen - dstId.size(), '\0');
+    return key;
+}
+
+// static
+std::string IndexKeyUtils::indexPrefix(PartitionID partId, IndexID indexId) {
+    PartitionID item = (partId << kPartitionOffset) | static_cast<uint32_t>(NebulaKeyType::kIndex);
+    std::string key;
+    key.reserve(sizeof(PartitionID) + sizeof(IndexID));
+    key.append(reinterpret_cast<const char*>(&item), sizeof(PartitionID))
+       .append(reinterpret_cast<const char*>(&indexId), sizeof(IndexID));
+    return key;
+}
 
 }  // namespace nebula
 
diff --git a/src/common/IndexKeyUtils.h b/src/common/IndexKeyUtils.h
index 6f42916df..46f093cfd 100644
--- a/src/common/IndexKeyUtils.h
+++ b/src/common/IndexKeyUtils.h
@@ -9,44 +9,59 @@
 
 #include "base/Base.h"
 #include "base/StatusOr.h"
-#include "thrift/ThriftTypes.h"
 #include "interface/gen-cpp2/meta_types.h"
 #include "common/Types.h"
 
-
 namespace nebula {
 
-using VariantType = boost::variant<int64_t, double, bool, std::string>;
-
-using OptVariantType = StatusOr<VariantType>;
-
-using IndexValues = std::vector<std::pair<nebula::meta::cpp2::PropertyType, std::string>>;
+using IndexValues = std::vector<std::pair<Value::Type, std::string>>;
 
 /**
- * This class supply some utils for transition between Vertex/Edge and key in kvstore.
+ * This class supply some utils for index in kvstore.
  * */
 class IndexKeyUtils final {
 public:
     ~IndexKeyUtils() = default;
 
-    static std::string encodeVariant(const VariantType& v)  {
-        switch (v.which()) {
-            case VAR_INT64:
-                return encodeInt64(boost::get<int64_t>(v));
-            case VAR_DOUBLE:
-                return encodeDouble(boost::get<double>(v));
-            case VAR_BOOL: {
-                auto val = boost::get<bool>(v);
+    static std::string encodeValue(const Value& v)  {
+        switch (v.type()) {
+            case Value::Type::INT :
+                return encodeInt64(v.getInt());
+            case Value::Type::FLOAT :
+                return encodeDouble(v.getFloat());
+            case Value::Type::BOOL: {
+                auto val = v.getBool();
                 std::string raw;
                 raw.reserve(sizeof(bool));
                 raw.append(reinterpret_cast<const char*>(&val), sizeof(bool));
                 return raw;
             }
-            case VAR_STR:
-                return boost::get<std::string>(v);
-            default:
-                std::string errMsg = folly::stringPrintf("Unknown VariantType: %d", v.which());
-                LOG(ERROR) << errMsg;
+            case Value::Type::STRING :
+                return v.getStr();
+            case Value::Type::DATE : {
+                std::string buf;
+                buf.reserve(sizeof(int8_t) * 2 + sizeof(int16_t));
+                buf.append(reinterpret_cast<const char*>(&v.getDate().year), sizeof(int16_t))
+                   .append(reinterpret_cast<const char*>(&v.getDate().month), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&v.getDate().day), sizeof(int8_t));
+                return buf;
+            }
+            case Value::Type::DATETIME : {
+                std::string buf;
+                buf.reserve(sizeof(int32_t) * 2 + sizeof(int16_t) + sizeof(int8_t) * 5);
+                auto dt = v.getDateTime();
+                buf.append(reinterpret_cast<const char*>(&dt.year), sizeof(int16_t))
+                   .append(reinterpret_cast<const char*>(&dt.month), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&dt.day), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&dt.hour), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&dt.minute), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&dt.sec), sizeof(int8_t))
+                   .append(reinterpret_cast<const char*>(&dt.microsec), sizeof(int32_t))
+                   .append(reinterpret_cast<const char*>(&dt.timezone), sizeof(int32_t));
+                return buf;
+            }
+            default :
+                LOG(ERROR) << "Unsupported default value type";
         }
         return "";
     }
@@ -118,53 +133,121 @@ class IndexKeyUtils final {
         return val;
     }
 
-    static OptVariantType decodeVariant(const folly::StringPiece& raw,
-                                        nebula::meta::cpp2::PropertyType type) {
+    static StatusOr<Value> decodeValue(const folly::StringPiece& raw, Value::Type type) {
+        Value v;
         switch (type) {
-            case nebula::meta::cpp2::PropertyType::BOOL : {
-                return *reinterpret_cast<const bool*>(raw.data());
+            case Value::Type::INT : {
+                v.setInt(decodeInt64(raw));
+                break;
+            }
+            case Value::Type::FLOAT : {
+                v.setFloat(decodeDouble(raw));
+                break;
             }
-            case nebula::meta::cpp2::PropertyType::INT64 :
-            case nebula::meta::cpp2::PropertyType::TIMESTAMP : {
-                return decodeInt64(raw);
+            case Value::Type::BOOL : {
+                v.setBool(*reinterpret_cast<const bool*>(raw.data()));
+                break;
             }
-            case nebula::meta::cpp2::PropertyType::DOUBLE :
-            case nebula::meta::cpp2::PropertyType::FLOAT : {
-                return decodeDouble(raw);
+            case Value::Type::STRING : {
+                v.setStr(raw.str());
+                break;
             }
-            case nebula::meta::cpp2::PropertyType::STRING : {
-                return raw.str();
+            case Value::Type::DATE: {
+                nebula::Date dt;
+                memcpy(reinterpret_cast<void*>(&dt.year), &raw[0], sizeof(int16_t));
+                memcpy(reinterpret_cast<void*>(&dt.month),
+                       &raw[sizeof(int16_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.day),
+                       &raw[sizeof(int16_t) + sizeof(int8_t)],
+                       sizeof(int8_t));
+                v.setDate(dt);
+                break;
+            }
+            case Value::Type::DATETIME: {
+                nebula::DateTime dt;
+                memcpy(reinterpret_cast<void*>(&dt.year), &raw[0], sizeof(int16_t));
+                memcpy(reinterpret_cast<void*>(&dt.month),
+                       &raw[sizeof(int16_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.day),
+                       &raw[sizeof(int16_t) + sizeof(int8_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.hour),
+                       &raw[sizeof(int16_t) + 2 * sizeof(int8_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.minute),
+                       &raw[sizeof(int16_t) + 3 * sizeof(int8_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.sec),
+                       &raw[sizeof(int16_t) + 4 * sizeof(int8_t)],
+                       sizeof(int8_t));
+                memcpy(reinterpret_cast<void*>(&dt.microsec),
+                       &raw[sizeof(int16_t) + 5 * sizeof(int8_t)],
+                       sizeof(int32_t));
+                memcpy(reinterpret_cast<void*>(&dt.timezone),
+                       &raw[sizeof(int16_t) + 5 * sizeof(int8_t) + sizeof(int32_t)],
+                       sizeof(int32_t));
+                v.setDateTime(dt);
+                break;
             }
             default:
-                return OptVariantType(Status::Error("Unknown type"));
+                return Status::Error("Unknown value type");
         }
+        return v;
     }
 
-    static VertexIntID getIndexVertexIntID(const folly::StringPiece& rawKey) {
-        CHECK_GE(rawKey.size(), kVertexIndexLen);
-        auto offset = rawKey.size() - sizeof(VertexIntID);
-        return *reinterpret_cast<const VertexIntID*>(rawKey.data() + offset);
+    static VertexIDSlice getIndexVertexID(size_t vIdLen, const folly::StringPiece& rawKey) {
+        CHECK_GE(rawKey.size(), kVertexIndexLen + vIdLen);
+        auto offset = rawKey.size() - vIdLen;
+        return rawKey.subpiece(offset, vIdLen);
      }
 
-    static VertexIntID getIndexSrcId(const folly::StringPiece& rawKey) {
-        CHECK_GE(rawKey.size(), kEdgeIndexLen);
-        auto offset = rawKey.size() -
-                      sizeof(VertexIntID) * 2 - sizeof(EdgeRanking);
-        return readInt<VertexIntID>(rawKey.data() + offset, sizeof(VertexIntID));
+    static VertexIDSlice getIndexSrcId(size_t vIdLen, const folly::StringPiece& rawKey) {
+        CHECK_GE(rawKey.size(), kEdgeIndexLen + vIdLen * 2);
+        auto offset = rawKey.size() - (vIdLen << 1) - sizeof(EdgeRanking);
+        return rawKey.subpiece(offset, vIdLen);
     }
 
-    static VertexIntID getIndexDstId(const folly::StringPiece& rawKey) {
-        CHECK_GE(rawKey.size(), kEdgeIndexLen);
-        auto offset = rawKey.size() - sizeof(VertexIntID);
-        return readInt<VertexIntID>(rawKey.data() + offset, sizeof(VertexIntID));
+    static VertexIDSlice getIndexDstId(size_t vIdLen, const folly::StringPiece& rawKey) {
+        CHECK_GE(rawKey.size(), kEdgeIndexLen + vIdLen * 2);
+        auto offset = rawKey.size() - vIdLen;
+        return rawKey.subpiece(offset, vIdLen);
     }
 
-    static EdgeRanking getIndexRank(const folly::StringPiece& rawKey) {
-        CHECK_GE(rawKey.size(), kEdgeIndexLen);
-        auto offset = rawKey.size() - sizeof(VertexIntID) - sizeof(EdgeRanking);
+    static EdgeRanking getIndexRank(size_t vIdLen, const folly::StringPiece& rawKey) {
+        CHECK_GE(rawKey.size(), kEdgeIndexLen + vIdLen * 2);
+        auto offset = rawKey.size() - vIdLen - sizeof(EdgeRanking);
         return readInt<EdgeRanking>(rawKey.data() + offset, sizeof(EdgeRanking));
     }
 
+    static bool isIndexKey(const folly::StringPiece& key) {
+        constexpr int32_t len = static_cast<int32_t>(sizeof(NebulaKeyType));
+        auto type = readInt<int32_t>(key.data(), len) & kTypeMask;
+        return static_cast<uint32_t>(NebulaKeyType::kIndex) == type;
+    }
+
+    static IndexID getIndexId(const folly::StringPiece& rawKey) {
+        auto offset = sizeof(PartitionID);
+        return readInt<IndexID>(rawKey.data() + offset, sizeof(IndexID));
+    }
+
+    /**
+     * Generate vertex|edge index key for kv store
+     **/
+    static void indexRaw(const IndexValues &values, std::string& raw);
+
+    static std::string vertexIndexKey(size_t vIdLen, PartitionID partId,
+                                      IndexID indexId, VertexID vId,
+                                      const IndexValues& values);
+
+    static std::string edgeIndexKey(size_t vIdLen, PartitionID partId,
+                                    IndexID indexId, VertexID srcId,
+                                    EdgeRanking rank, VertexID dstId,
+                                    const IndexValues& values);
+
+    static std::string indexPrefix(PartitionID partId, IndexID indexId);
+
 private:
     IndexKeyUtils() = delete;
 
diff --git a/src/common/NebulaKeyUtils.h b/src/common/NebulaKeyUtils.h
index 60795cdff..5cee60e08 100644
--- a/src/common/NebulaKeyUtils.h
+++ b/src/common/NebulaKeyUtils.h
@@ -11,8 +11,6 @@
 
 namespace nebula {
 
-using VertexIDSlice = folly::StringPiece;
-
 /**
  * VertexKeyUtils:
  * type(1) + partId(3) + vertexId(*) + tagId(4) + version(8)
@@ -202,17 +200,6 @@ class NebulaKeyUtils final {
         return rawKey.subpiece(0, rawKey.size() - sizeof(int64_t));
     }
 
-    static bool isIndexKey(const folly::StringPiece& key) {
-        constexpr int32_t len = static_cast<int32_t>(sizeof(NebulaKeyType));
-        auto type = readInt<int32_t>(key.data(), len) & kTypeMask;
-        return static_cast<uint32_t>(NebulaKeyType::kIndex) == type;
-    }
-
-    static IndexID getIndexId(const folly::StringPiece& rawKey) {
-        auto offset = sizeof(PartitionID);
-        return readInt<IndexID>(rawKey.data() + offset, sizeof(IndexID));
-    }
-
     static void dumpBadKey(const folly::StringPiece& rawKey, size_t expect, size_t vIdLen) {
         std::stringstream msg;
         msg << "rawKey.size() != expect size"
diff --git a/src/common/Types.h b/src/common/Types.h
index 621f9bf84..8cb123e2b 100644
--- a/src/common/Types.h
+++ b/src/common/Types.h
@@ -24,6 +24,7 @@ enum class NebulaSystemKeyType : uint32_t {
     kSystemPart        = 0x00000002,
 };
 
+using VertexIDSlice = folly::StringPiece;
 using VertexIntID = int64_t;
 using IndexID = int32_t;
 
@@ -38,7 +39,8 @@ readInt(const char* data, int32_t len) {
 static constexpr int32_t kVertexLen = sizeof(PartitionID) + sizeof(TagID) + sizeof(TagVersion);
 
 // size of vertex key except srcId and dstId
-static constexpr int32_t kEdgeLen = sizeof(PartitionID) + sizeof(EdgeType) + sizeof(EdgeRanking) + sizeof(EdgeVersion);
+static constexpr int32_t kEdgeLen = sizeof(PartitionID) + sizeof(EdgeType) +
+                                    sizeof(EdgeRanking) + sizeof(EdgeVersion);
 
 static constexpr int32_t kSystemLen = sizeof(PartitionID) + sizeof(NebulaSystemKeyType);
 
@@ -59,13 +61,10 @@ static constexpr uint32_t kEdgeMaskSet      = kTagEdgeMask;
 // Write Tag by |=
 static constexpr uint32_t kTagMaskSet       = ~kTagEdgeMask;
 
-static constexpr int32_t kVertexIndexLen = sizeof(PartitionID) + sizeof(IndexID)
-                                           + sizeof(VertexID);
+static constexpr int32_t kVertexIndexLen = sizeof(PartitionID) + sizeof(IndexID);
 
-static constexpr int32_t kEdgeIndexLen = sizeof(PartitionID) + sizeof(IndexID)
-                                         + sizeof(VertexID) * 2 + sizeof(EdgeRanking);
-
-static constexpr int32_t kIndexLen = std::min(kVertexIndexLen, kEdgeIndexLen);
+static constexpr int32_t kEdgeIndexLen = sizeof(PartitionID) + sizeof(IndexID) + 
+                                         sizeof(EdgeRanking);
 
 }  // namespace nebula
 #endif  // COMMON_TYPES_H_
diff --git a/src/common/test/CMakeLists.txt b/src/common/test/CMakeLists.txt
index 89fed2e22..60274d354 100644
--- a/src/common/test/CMakeLists.txt
+++ b/src/common/test/CMakeLists.txt
@@ -1,6 +1,23 @@
 nebula_add_test(
-    NAME nebula_key_utils_test
-    SOURCES NebulaKeyUtilsTest.cpp
-    OBJECTS $<TARGET_OBJECTS:keyutils_obj>
-    LIBRARIES gtest
-)
\ No newline at end of file
+    NAME
+        nebula_key_utils_test
+    SOURCES
+        NebulaKeyUtilsTest.cpp
+    OBJECTS
+        $<TARGET_OBJECTS:keyutils_obj>
+    LIBRARIES
+        gtest
+)
+
+nebula_add_test(
+    NAME
+        index_key_utils_test
+    SOURCES
+        IndexKeyUtilsTest.cpp
+    OBJECTS
+        $<TARGET_OBJECTS:keyutils_obj>
+        $<TARGET_OBJECTS:common_base_obj>
+        $<TARGET_OBJECTS:common_datatypes_obj>
+    LIBRARIES
+        gtest
+)
diff --git a/src/common/test/IndexKeyUtilsTest.cpp b/src/common/test/IndexKeyUtilsTest.cpp
new file mode 100644
index 000000000..609d4cb0f
--- /dev/null
+++ b/src/common/test/IndexKeyUtilsTest.cpp
@@ -0,0 +1,208 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "base/Base.h"
+#include "common/IndexKeyUtils.h"
+#include <gtest/gtest.h>
+
+namespace nebula {
+
+VertexID getStringId(int64_t vId) {
+    std::string id;
+    id.append(reinterpret_cast<const char*>(&vId), sizeof(int64_t));
+    return id;
+}
+
+IndexValues getIndexValues() {
+    IndexValues values;
+    values.emplace_back(Value::Type::STRING, IndexKeyUtils::encodeValue(Value("str")));
+    values.emplace_back(Value::Type::BOOL, IndexKeyUtils::encodeValue(Value(true)));
+    values.emplace_back(Value::Type::INT, IndexKeyUtils::encodeValue(Value(12L)));
+    values.emplace_back(Value::Type::FLOAT, IndexKeyUtils::encodeValue(Value(12.12f)));
+    nebula::Date d;
+    d.year = 2020;
+    d.month = 4;
+    d.day = 11;
+    values.emplace_back(Value::Type::DATE, IndexKeyUtils::encodeValue(Value(std::move(d))));
+
+    nebula::DateTime dt;
+    dt.year = 2020;
+    dt.month = 4;
+    dt.day = 11;
+    dt.hour = 12;
+    dt.minute = 30;
+    dt.sec = 22;
+    dt.microsec = 1111;
+    dt.timezone = 2222;
+    values.emplace_back(Value::Type::DATE, IndexKeyUtils::encodeValue(Value(std::move(dt))));
+    return values;
+}
+
+bool evalInt64(int64_t val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::INT);
+    if (!res.ok()) {
+        return false;
+    }
+
+    EXPECT_EQ(Value::Type::INT, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getInt();
+}
+
+bool evalDouble(double val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::FLOAT);
+    if (!res.ok()) {
+        return false;
+    }
+    EXPECT_EQ(Value::Type::FLOAT, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getFloat();
+}
+
+bool evalBool(bool val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::BOOL);
+    if (!res.ok()) {
+        return false;
+    }
+    EXPECT_EQ(Value::Type::BOOL, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getBool();
+}
+
+bool evalString(std::string val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::STRING);
+    if (!res.ok()) {
+        return false;
+    }
+    EXPECT_EQ(Value::Type::STRING, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getStr();
+}
+
+bool evalDate(nebula::Date val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::DATE);
+    if (!res.ok()) {
+        return false;
+    }
+    EXPECT_EQ(Value::Type::DATE, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getDate();
+}
+
+bool evalDateTime(nebula::DateTime val) {
+    Value v(val);
+    auto str = IndexKeyUtils::encodeValue(v);
+    auto res = IndexKeyUtils::decodeValue(str, Value::Type::DATETIME);
+    if (!res.ok()) {
+        return false;
+    }
+    EXPECT_EQ(Value::Type::DATETIME, res.value().type());
+    EXPECT_EQ(v, res.value());
+    return val == res.value().getDateTime();
+}
+
+TEST(IndexKeyUtilsTest, encodeValue) {
+    EXPECT_TRUE(evalInt64(1));
+    EXPECT_TRUE(evalInt64(0));
+    EXPECT_TRUE(evalInt64(std::numeric_limits<int64_t>::max()));
+    EXPECT_TRUE(evalInt64(std::numeric_limits<int64_t>::min()));
+    EXPECT_TRUE(evalDouble(1.1));
+    EXPECT_TRUE(evalDouble(0.0));
+    EXPECT_TRUE(evalDouble(std::numeric_limits<double>::max()));
+    EXPECT_TRUE(evalDouble(std::numeric_limits<double>::min()));
+    EXPECT_TRUE(evalDouble(-std::numeric_limits<double>::max()));
+    EXPECT_TRUE(evalDouble(-(0.000000001 - std::numeric_limits<double>::min())));
+    EXPECT_TRUE(evalBool(true));
+    EXPECT_TRUE(evalBool(false));
+    EXPECT_TRUE(evalString("test"));
+    nebula::Date d;
+    d.year = 2020;
+    d.month = 4;
+    d.day = 11;
+    EXPECT_TRUE(evalDate(d));
+
+    nebula::DateTime dt;
+    dt.year = 2020;
+    dt.month = 4;
+    dt.day = 11;
+    dt.hour = 12;
+    dt.minute = 30;
+    dt.sec = 22;
+    dt.microsec = 1111;
+    dt.timezone = 2222;
+    EXPECT_TRUE(evalDateTime(dt));
+}
+
+TEST(IndexKeyUtilsTest, encodeDouble) {
+    EXPECT_TRUE(evalDouble(100.5));
+    EXPECT_TRUE(evalDouble(200.5));
+    EXPECT_TRUE(evalDouble(300.5));
+    EXPECT_TRUE(evalDouble(400.5));
+    EXPECT_TRUE(evalDouble(500.5));
+    EXPECT_TRUE(evalDouble(600.5));
+}
+
+TEST(IndexKeyUtilsTest, vertexIndexKeyV1) {
+    auto values = getIndexValues();
+    auto key = IndexKeyUtils::vertexIndexKey(8, 1, 1, getStringId(1), values);
+    ASSERT_EQ(1, IndexKeyUtils::getIndexId(key));
+    ASSERT_EQ(getStringId(1), IndexKeyUtils::getIndexVertexID(8, key));
+    ASSERT_EQ(true, IndexKeyUtils::isIndexKey(key));
+}
+
+TEST(IndexKeyUtilsTest, vertexIndexKeyV2) {
+    auto values = getIndexValues();
+    auto key = IndexKeyUtils::vertexIndexKey(100, 1, 1, "vertex_1_1_1_1", values);
+    ASSERT_EQ(1, IndexKeyUtils::getIndexId(key));
+
+    VertexID vid = "vertex_1_1_1_1";
+    vid.append(100 - vid.size(), '\0');
+    ASSERT_EQ(vid, IndexKeyUtils::getIndexVertexID(100, key));
+    ASSERT_EQ(true, IndexKeyUtils::isIndexKey(key));
+}
+
+TEST(IndexKeyUtilsTest, edgeIndexKeyV1) {
+    auto values = getIndexValues();
+    auto key = IndexKeyUtils::edgeIndexKey(8, 1, 1, getStringId(1), 1, getStringId(2), values);
+    ASSERT_EQ(1, IndexKeyUtils::getIndexId(key));
+    ASSERT_EQ(getStringId(1), IndexKeyUtils::getIndexSrcId(8, key));
+    ASSERT_EQ(1, IndexKeyUtils::getIndexRank(8, key));
+    ASSERT_EQ(getStringId(2), IndexKeyUtils::getIndexDstId(8, key));
+    ASSERT_EQ(true, IndexKeyUtils::isIndexKey(key));
+}
+
+TEST(IndexKeyUtilsTest, edgeIndexKeyV2) {
+    VertexID vid = "vertex_1_1_1_1";
+    auto values = getIndexValues();
+    auto key = IndexKeyUtils::edgeIndexKey(100, 1, 1, vid, 1, vid, values);
+    ASSERT_EQ(1, IndexKeyUtils::getIndexId(key));
+    vid.append(100 - vid.size(), '\0');
+    ASSERT_EQ(vid, IndexKeyUtils::getIndexSrcId(100, key));
+    ASSERT_EQ(1, IndexKeyUtils::getIndexRank(100, key));
+    ASSERT_EQ(vid, IndexKeyUtils::getIndexDstId(100, key));
+    ASSERT_EQ(true, IndexKeyUtils::isIndexKey(key));
+}
+
+}  // namespace nebula
+
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+    folly::init(&argc, &argv, true);
+    google::SetStderrLogging(google::INFO);
+
+    return RUN_ALL_TESTS();
+}
diff --git a/src/storage/CompactionFilter.h b/src/storage/CompactionFilter.h
index a2a2b8bc2..16e40fd3e 100644
--- a/src/storage/CompactionFilter.h
+++ b/src/storage/CompactionFilter.h
@@ -9,6 +9,7 @@
 
 #include "base/Base.h"
 #include "common/NebulaKeyUtils.h"
+#include "common/IndexKeyUtils.h"
 #include "codec/RowReader.h"
 #include "meta/NebulaSchemaProvider.h"
 #include "kvstore/CompactionFilter.h"
@@ -51,7 +52,7 @@ class StorageCompactionFilter final : public kvstore::KVFilter {
                 VLOG(3) << "Extra versions has been filtered!";
                 return true;
             }
-        } else if (NebulaKeyUtils::isIndexKey(key)) {
+        } else if (IndexKeyUtils::isIndexKey(key)) {
             if (!indexValid(spaceId, key)) {
                 VLOG(3) << "Index invalid for the key " << key;
                 return true;
@@ -172,7 +173,7 @@ class StorageCompactionFilter final : public kvstore::KVFilter {
     }
 
     bool indexValid(GraphSpaceID spaceId, const folly::StringPiece& key) const {
-        auto indexId = NebulaKeyUtils::getIndexId(key);
+        auto indexId = IndexKeyUtils::getIndexId(key);
         auto eRet = this->indexMan_->getEdgeIndex(spaceId, indexId);
         if (eRet.ok()) {
             return true;