diff --git a/docs/flex/interactive/data_model.md b/docs/flex/interactive/data_model.md
index 01f9dcd25c2b..18a5eca0b5be 100644
--- a/docs/flex/interactive/data_model.md
+++ b/docs/flex/interactive/data_model.md
@@ -67,8 +67,6 @@ property_type:
   primitive_type: DT_STRING
 ```
 
-Please note that we currently do not support the use of string data type for properties on edges.
-
 ### Array Types
 
 Array types are currently not supported, but are planned to be supported in the near future.
diff --git a/flex/bin/interactive_server.cc b/flex/bin/interactive_server.cc
index 52e210a7ada6..800077cd4555 100644
--- a/flex/bin/interactive_server.cc
+++ b/flex/bin/interactive_server.cc
@@ -314,6 +314,7 @@ int main(int argc, char** argv) {
     data_path = vm["data-path"].as<std::string>();
 
     auto schema = gs::Schema::LoadFromYaml(graph_schema_path);
+
     // Ths schema is loaded just to get the plugin dir and plugin list
     gs::init_codegen_proxy(vm, graph_schema_path, engine_config_file);
     db.Close();
diff --git a/flex/engines/graph_db/database/graph_db.cc b/flex/engines/graph_db/database/graph_db.cc
index 500f3412906f..9c80a11b782e 100644
--- a/flex/engines/graph_db/database/graph_db.cc
+++ b/flex/engines/graph_db/database/graph_db.cc
@@ -76,6 +76,7 @@ Result<bool> GraphDB::Open(const Schema& schema, const std::string& data_dir,
   }
 
   if ((!create_empty_graph) && (!graph_.schema().Equals(schema))) {
+    LOG(ERROR) << "Schema inconsistent..\n";
     return Result<bool>(StatusCode::InternalError,
                         "Schema of work directory is not compatible with the "
                         "graph schema",
diff --git a/flex/engines/graph_db/database/transaction_utils.h b/flex/engines/graph_db/database/transaction_utils.h
index c8d6b039046e..d338af989ed6 100644
--- a/flex/engines/graph_db/database/transaction_utils.h
+++ b/flex/engines/graph_db/database/transaction_utils.h
@@ -26,74 +26,53 @@
 namespace gs {
 
 inline void serialize_field(grape::InArchive& arc, const Any& prop) {
-  switch (prop.type) {
-  case PropertyType::kBool:
+  if (prop.type == PropertyType::Bool()) {
     arc << prop.value.b;
-    break;
-  case PropertyType::kInt32:
+  } else if (prop.type == PropertyType::Int32()) {
     arc << prop.value.i;
-    break;
-  case PropertyType::kUInt32:
+  } else if (prop.type == PropertyType::UInt32()) {
     arc << prop.value.ui;
-    break;
-  case PropertyType::kDate:
+  } else if (prop.type == PropertyType::Date()) {
     arc << prop.value.d.milli_second;
-    break;
-  case PropertyType::kString:
+  } else if (prop.type == PropertyType::String()) {
     arc << prop.value.s;
-    break;
-  case PropertyType::kEmpty:
-    break;
-  case PropertyType::kInt64:
+  } else if (prop.type == PropertyType::Int64()) {
     arc << prop.value.l;
-    break;
-  case PropertyType::kUInt64:
+  } else if (prop.type == PropertyType::UInt64()) {
     arc << prop.value.ul;
-    break;
-  case PropertyType::kDouble:
+  } else if (prop.type == PropertyType::Double()) {
     arc << prop.value.db;
-    break;
-  case PropertyType::kFloat:
+  } else if (prop.type == PropertyType::Float()) {
     arc << prop.value.f;
-    break;
-  default:
-    LOG(FATAL) << "Unexpected property type";
+  } else if (prop.type == PropertyType::Empty()) {
+  } else {
+    LOG(FATAL) << "Unexpected property type" << int(prop.type.type_enum);
   }
 }
 
 inline void deserialize_field(grape::OutArchive& arc, Any& prop) {
-  switch (prop.type) {
-  case PropertyType::kBool:
+  if (prop.type == PropertyType::Bool()) {
     arc >> prop.value.b;
-    break;
-  case PropertyType::kInt32:
+  } else if (prop.type == PropertyType::Int32()) {
     arc >> prop.value.i;
-    break;
-  case PropertyType::kUInt32:
+  } else if (prop.type == PropertyType::UInt32()) {
     arc >> prop.value.ui;
-    break;
-  case PropertyType::kDate:
+  } else if (prop.type == PropertyType::Date()) {
     arc >> prop.value.d.milli_second;
-    break;
-  case PropertyType::kString:
+  } else if (prop.type == PropertyType::String()) {
     arc >> prop.value.s;
-    break;
-  case PropertyType::kEmpty:
-    break;
-  case PropertyType::kInt64:
+  } else if (prop.type == PropertyType::Int64()) {
     arc >> prop.value.l;
-    break;
-  case PropertyType::kUInt64:
+  } else if (prop.type == PropertyType::UInt64()) {
     arc >> prop.value.ul;
-    break;
-  case PropertyType::kDouble:
+  } else if (prop.type == PropertyType::Double()) {
     arc >> prop.value.db;
-    break;
-  case PropertyType::kFloat:
+  } else if (prop.type == PropertyType::Float()) {
     arc >> prop.value.f;
-    break;
-  default:
-    LOG(FATAL) << "Unexpected property type";
+  } else if (prop.type == PropertyType::Empty()) {
+  } else {
+    LOG(FATAL) << "Unexpected property type: "
+               << static_cast<int>(prop.type.type_enum);
   }
 }
 
diff --git a/flex/engines/hqps_db/core/operator/sink.h b/flex/engines/hqps_db/core/operator/sink.h
index 037efb05d398..3321819318b9 100644
--- a/flex/engines/hqps_db/core/operator/sink.h
+++ b/flex/engines/hqps_db/core/operator/sink.h
@@ -179,41 +179,31 @@ void template_set_tuple_value(results::Collection* collection,
 /////Sink Any
 
 void set_any_to_common_value(const Any& any, common::Value* value) {
-  switch (any.type) {
-  case gs::PropertyType::kBool:
+  if (any.type == PropertyType::Bool()) {
     value->set_boolean(any.value.b);
-    break;
-  case gs::PropertyType::kInt32:
+  } else if (any.type == PropertyType::Int32()) {
     value->set_i32(any.value.i);
-    break;
-  case gs::PropertyType::kInt64:
-    value->set_i64(any.value.l);
-    break;
-  case gs::PropertyType::kUInt32:
-    // FIXME(zhanglei): temporarily use i64, fix this after common.proto is
+  } else if (any.type == PropertyType::UInt32()) {
+    // FIXME(zhanglei): temporarily use i32, fix this after common.proto is
     // changed
     value->set_i32(any.value.ui);
-    break;
-  case gs::PropertyType::kUInt64:
+  } else if (any.type == PropertyType::Int64()) {
+    value->set_i64(any.value.l);
+  } else if (any.type == PropertyType::UInt64()) {
     // FIXME(zhanglei): temporarily use i64, fix this after common.proto is
     // changed
     value->set_i64(any.value.ul);
-    break;
-  case gs::PropertyType::kDate:
-    value->set_i64(any.value.d.milli_second);
-    break;
-  case gs::PropertyType::kString:
-    value->mutable_str()->assign(any.value.s.data(), any.value.s.size());
-    break;
-  case gs::PropertyType::kDouble:
+  } else if (any.type == PropertyType::Double()) {
     value->set_f64(any.value.db);
-    break;
-  case gs::PropertyType::kFloat:
+  } else if (any.type == PropertyType::Float()) {
     value->set_f64(any.value.f);
-    break;
-  default:
-    LOG(WARNING) << "Unsupported type: " << any.type;
-    break;
+  } else if (any.type == PropertyType::Date()) {
+    value->set_i64(any.value.d.milli_second);
+  } else if (any.type == PropertyType::String()) {
+    value->mutable_str()->assign(any.value.s.data(), any.value.s.size());
+  } else {
+    LOG(WARNING) << "Unexpected property type: "
+                 << static_cast<int>(any.type.type_enum);
   }
 }
 
diff --git a/flex/engines/hqps_db/database/mutable_csr_interface.h b/flex/engines/hqps_db/database/mutable_csr_interface.h
index 782d306d3169..d2d6999b895a 100644
--- a/flex/engines/hqps_db/database/mutable_csr_interface.h
+++ b/flex/engines/hqps_db/database/mutable_csr_interface.h
@@ -852,7 +852,7 @@ class MutableCSRInterface {
           *std::dynamic_pointer_cast<TypedColumn<float>>(column));
     } else {
       LOG(FATAL) << "unexpected type to create column, "
-                 << static_cast<int>(type);
+                 << static_cast<int>(type.type_enum);
       return nullptr;
     }
   }
diff --git a/flex/engines/hqps_db/structures/multi_edge_set/untyped_edge_set.h b/flex/engines/hqps_db/structures/multi_edge_set/untyped_edge_set.h
index ee4abad40c84..3e948edf93a0 100644
--- a/flex/engines/hqps_db/structures/multi_edge_set/untyped_edge_set.h
+++ b/flex/engines/hqps_db/structures/multi_edge_set/untyped_edge_set.h
@@ -248,7 +248,7 @@ class UnTypedEdgeSet {
     while (iter != end_iter) {
       auto edata = iter.GetData();
       PropT prop;
-      if (edata.type == TypeConverter<PropT>::property_type) {
+      if (edata.type == TypeConverter<PropT>::property_type()) {
         ConvertAny<PropT>::to(edata, prop);
       }
       CHECK(cur_ind < sum) << "cur: " << cur_ind << ", sum: " << sum;
diff --git a/flex/storages/rt_mutable_graph/dual_csr.h b/flex/storages/rt_mutable_graph/dual_csr.h
index 6468f1e688ef..35e951614d4a 100644
--- a/flex/storages/rt_mutable_graph/dual_csr.h
+++ b/flex/storages/rt_mutable_graph/dual_csr.h
@@ -164,8 +164,10 @@ class DualCsr : public DualCsrBase {
 template <>
 class DualCsr<std::string_view> : public DualCsrBase {
  public:
-  DualCsr(EdgeStrategy oe_strategy, EdgeStrategy ie_strategy)
-      : in_csr_(nullptr), out_csr_(nullptr), column_(StorageStrategy::kMem) {
+  DualCsr(EdgeStrategy oe_strategy, EdgeStrategy ie_strategy, uint16_t width)
+      : in_csr_(nullptr),
+        out_csr_(nullptr),
+        column_(StorageStrategy::kMem, width) {
     if (ie_strategy == EdgeStrategy::kNone) {
       in_csr_ = new EmptyCsr<std::string_view>(column_, column_idx_);
     } else if (ie_strategy == EdgeStrategy::kMultiple) {
diff --git a/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.cc b/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.cc
index 5450376250e5..8bfe4f2fe710 100644
--- a/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.cc
+++ b/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.cc
@@ -29,142 +29,86 @@ bool check_primary_key_type(std::shared_ptr<arrow::DataType> data_type) {
   }
 }
 
-void set_vertex_properties(gs::ColumnBase* col,
-                           std::shared_ptr<arrow::ChunkedArray> array,
-                           const std::vector<vid_t>& vids) {
+void set_vertex_column_from_string_array(
+    gs::ColumnBase* col, std::shared_ptr<arrow::ChunkedArray> array,
+    const std::vector<vid_t>& vids) {
   auto type = array->type();
-  auto col_type = col->type();
+  CHECK(type->Equals(arrow::large_utf8()) || type->Equals(arrow::utf8()))
+      << "Inconsistent data type, expect string, but got " << type->ToString();
   size_t cur_ind = 0;
-  if (col_type == PropertyType::kBool) {
-    CHECK(type->Equals(arrow::boolean()))
-        << "Inconsistent data type, expect bool, but got " << type->ToString();
+  if (type->Equals(arrow::large_utf8())) {
     for (auto j = 0; j < array->num_chunks(); ++j) {
       auto casted =
-          std::static_pointer_cast<arrow::BooleanArray>(array->chunk(j));
+          std::static_pointer_cast<arrow::LargeStringArray>(array->chunk(j));
       for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(vids[cur_ind++],
-                     std::move(AnyConverter<bool>::to_any(casted->Value(k))));
+        auto str = casted->GetView(k);
+        std::string_view sw(str.data(), str.size());
+        col->set_any(vids[cur_ind++], std::move(sw));
       }
     }
-  } else if (col_type == PropertyType::kInt64) {
-    CHECK(type->Equals(arrow::int64()))
-        << "Inconsistent data type, expect int64, but got " << type->ToString();
+  } else {
     for (auto j = 0; j < array->num_chunks(); ++j) {
       auto casted =
-          std::static_pointer_cast<arrow::Int64Array>(array->chunk(j));
+          std::static_pointer_cast<arrow::StringArray>(array->chunk(j));
       for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(
-            vids[cur_ind++],
-            std::move(AnyConverter<int64_t>::to_any(casted->Value(k))));
+        auto str = casted->GetView(k);
+        std::string_view sw(str.data(), str.size());
+        col->set_any(vids[cur_ind++], std::move(sw));
       }
     }
+  }
+}
+
+void set_vertex_properties(gs::ColumnBase* col,
+                           std::shared_ptr<arrow::ChunkedArray> array,
+                           const std::vector<vid_t>& vids) {
+  auto type = array->type();
+  auto col_type = col->type();
+
+  // TODO(zhanglei): reduce the dummy code here with a template function.
+  if (col_type == PropertyType::kBool) {
+    set_single_vertex_column<bool>(col, array, vids);
+  } else if (col_type == PropertyType::kInt64) {
+    set_single_vertex_column<int64_t>(col, array, vids);
   } else if (col_type == PropertyType::kInt32) {
-    CHECK(type->Equals(arrow::int32()))
-        << "Inconsistent data type, expect int32, but got " << type->ToString();
-    for (auto j = 0; j < array->num_chunks(); ++j) {
-      auto casted =
-          std::static_pointer_cast<arrow::Int32Array>(array->chunk(j));
-      for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(
-            vids[cur_ind++],
-            std::move(AnyConverter<int32_t>::to_any(casted->Value(k))));
-      }
-    }
+    set_single_vertex_column<int32_t>(col, array, vids);
   } else if (col_type == PropertyType::kUInt64) {
-    CHECK(type->Equals(arrow::uint64()))
-        << "Inconsistent data type, expect uint64, but got "
-        << type->ToString();
-    for (auto j = 0; j < array->num_chunks(); ++j) {
-      auto casted =
-          std::static_pointer_cast<arrow::UInt64Array>(array->chunk(j));
-      for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(
-            vids[cur_ind++],
-            std::move(AnyConverter<uint64_t>::to_any(casted->Value(k))));
-      }
-    }
+    set_single_vertex_column<uint64_t>(col, array, vids);
   } else if (col_type == PropertyType::kUInt32) {
-    CHECK(type->Equals(arrow::uint32()))
-        << "Inconsistent data type, expect uint32, but got "
-        << type->ToString();
-    for (auto j = 0; j < array->num_chunks(); ++j) {
-      auto casted =
-          std::static_pointer_cast<arrow::UInt32Array>(array->chunk(j));
-      for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(
-            vids[cur_ind++],
-            std::move(AnyConverter<uint32_t>::to_any(casted->Value(k))));
-      }
-    }
+    set_single_vertex_column<uint32_t>(col, array, vids);
   } else if (col_type == PropertyType::kDouble) {
-    CHECK(type->Equals(arrow::float64()))
-        << "Inconsistent data type, expect double, but got "
-        << type->ToString();
-    for (auto j = 0; j < array->num_chunks(); ++j) {
-      auto casted =
-          std::static_pointer_cast<arrow::DoubleArray>(array->chunk(j));
-      for (auto k = 0; k < casted->length(); ++k) {
-        col->set_any(vids[cur_ind++],
-                     std::move(AnyConverter<double>::to_any(casted->Value(k))));
-      }
-    }
+    set_single_vertex_column<double>(col, array, vids);
   } else if (col_type == PropertyType::kFloat) {
-    CHECK(type->Equals(arrow::float32()))
-        << "Inconsistent data type, expect float, but got " << type->ToString();
+    set_single_vertex_column<float>(col, array, vids);
+  } else if (col_type == PropertyType::kStringMap) {
+    set_vertex_column_from_string_array(col, array, vids);
+  } else if (col_type == PropertyType::kDate) {
+    set_vertex_column_from_timestamp_array(col, array, vids);
+  } else if (col_type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    set_vertex_column_from_string_array(col, array, vids);
+  } else {
+    LOG(FATAL) << "Not support type: " << type->ToString();
+  }
+}
+
+void set_vertex_column_from_timestamp_array(
+    gs::ColumnBase* col, std::shared_ptr<arrow::ChunkedArray> array,
+    const std::vector<vid_t>& vids) {
+  auto type = array->type();
+  auto col_type = col->type();
+  size_t cur_ind = 0;
+  if (type->Equals(arrow::timestamp(arrow::TimeUnit::type::MILLI))) {
     for (auto j = 0; j < array->num_chunks(); ++j) {
       auto casted =
-          std::static_pointer_cast<arrow::FloatArray>(array->chunk(j));
+          std::static_pointer_cast<arrow::TimestampArray>(array->chunk(j));
       for (auto k = 0; k < casted->length(); ++k) {
         col->set_any(vids[cur_ind++],
-                     std::move(AnyConverter<float>::to_any(casted->Value(k))));
-      }
-    }
-  } else if (col_type == PropertyType::kString ||
-             col_type == PropertyType::kStringMap) {
-    CHECK(type->Equals(arrow::large_utf8()) || type->Equals(arrow::utf8()))
-        << "Inconsistent data type, expect string, but got "
-        << type->ToString();
-    if (type->Equals(arrow::large_utf8())) {
-      for (auto j = 0; j < array->num_chunks(); ++j) {
-        auto casted =
-            std::static_pointer_cast<arrow::LargeStringArray>(array->chunk(j));
-        for (auto k = 0; k < casted->length(); ++k) {
-          auto str = casted->GetView(k);
-          std::string_view str_view(str.data(), str.size());
-          col->set_any(
-              vids[cur_ind++],
-              std::move(AnyConverter<std::string_view>::to_any(str_view)));
-        }
-      }
-    } else {
-      for (auto j = 0; j < array->num_chunks(); ++j) {
-        auto casted =
-            std::static_pointer_cast<arrow::StringArray>(array->chunk(j));
-        for (auto k = 0; k < casted->length(); ++k) {
-          auto str = casted->GetView(k);
-          std::string_view str_view(str.data(), str.size());
-          col->set_any(
-              vids[cur_ind++],
-              std::move(AnyConverter<std::string_view>::to_any(str_view)));
-        }
+                     std::move(AnyConverter<Date>::to_any(casted->Value(k))));
       }
     }
-  } else if (col_type == PropertyType::kDate) {
-    if (type->Equals(arrow::timestamp(arrow::TimeUnit::type::MILLI))) {
-      for (auto j = 0; j < array->num_chunks(); ++j) {
-        auto casted =
-            std::static_pointer_cast<arrow::TimestampArray>(array->chunk(j));
-        for (auto k = 0; k < casted->length(); ++k) {
-          col->set_any(vids[cur_ind++],
-                       std::move(AnyConverter<Date>::to_any(casted->Value(k))));
-        }
-      }
-    } else {
-      LOG(FATAL) << "Not implemented: converting " << type->ToString() << " to "
-                 << col_type;
-    }
   } else {
-    LOG(FATAL) << "Not support type: " << type->ToString();
+    LOG(FATAL) << "Not implemented: converting " << type->ToString() << " to "
+               << col_type;
   }
 }
 
@@ -223,15 +167,15 @@ void AbstractArrowFragmentLoader::AddVerticesRecordBatch(
 
   if (type == PropertyType::kInt64) {
     addVertexRecordBatchImpl<int64_t>(v_label_id, v_files, supplier_creator);
-  } else if (type == PropertyType::kString) {
-    addVertexRecordBatchImpl<std::string_view>(v_label_id, v_files,
-                                               supplier_creator);
   } else if (type == PropertyType::kInt32) {
     addVertexRecordBatchImpl<int32_t>(v_label_id, v_files, supplier_creator);
   } else if (type == PropertyType::kUInt32) {
     addVertexRecordBatchImpl<uint32_t>(v_label_id, v_files, supplier_creator);
   } else if (type == PropertyType::kUInt64) {
     addVertexRecordBatchImpl<uint64_t>(v_label_id, v_files, supplier_creator);
+  } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    addVertexRecordBatchImpl<std::string_view>(v_label_id, v_files,
+                                               supplier_creator);
   }
   VLOG(10) << "Finish init vertices for label " << v_label_name;
 }
@@ -320,14 +264,6 @@ void AbstractArrowFragmentLoader::AddEdgesRecordBatch(
       addEdgesRecordBatchImpl<uint64_t>(src_label_i, dst_label_i, edge_label_i,
                                         filenames, supplier_creator);
     }
-  } else if (property_types[0] == PropertyType::kString) {
-    if (filenames.empty()) {
-      basic_fragment_loader_.AddNoPropEdgeBatch<std::string_view>(
-          src_label_i, dst_label_i, edge_label_i);
-    } else {
-      addEdgesRecordBatchImpl<std::string_view>(
-          src_label_i, dst_label_i, edge_label_i, filenames, supplier_creator);
-    }
   } else if (property_types[0] == PropertyType::kDouble) {
     if (filenames.empty()) {
       basic_fragment_loader_.AddNoPropEdgeBatch<double>(
@@ -344,6 +280,14 @@ void AbstractArrowFragmentLoader::AddEdgesRecordBatch(
       addEdgesRecordBatchImpl<float>(src_label_i, dst_label_i, edge_label_i,
                                      filenames, supplier_creator);
     }
+  } else if (property_types[0].type_enum == impl::PropertyTypeImpl::kVarChar) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<std::string_view>(
+          src_label_i, dst_label_i, edge_label_i);
+    } else {
+      addEdgesRecordBatchImpl<std::string_view>(
+          src_label_i, dst_label_i, edge_label_i, filenames, supplier_creator);
+    }
   } else {
     LOG(FATAL) << "Unsupported edge property type." << property_types[0];
   }
diff --git a/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.h b/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.h
index 22f1b95bef9a..43e46a9539db 100644
--- a/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.h
+++ b/flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.h
@@ -38,6 +38,36 @@ class IRecordBatchSupplier {
 
 bool check_primary_key_type(std::shared_ptr<arrow::DataType> data_type);
 
+// For Primitive types.
+template <typename COL_T>
+void set_single_vertex_column(gs::ColumnBase* col,
+                              std::shared_ptr<arrow::ChunkedArray> array,
+                              const std::vector<vid_t>& vids) {
+  using arrow_array_type = typename gs::TypeConverter<COL_T>::ArrowArrayType;
+  auto array_type = array->type();
+  auto arrow_type = gs::TypeConverter<COL_T>::ArrowTypeValue();
+  CHECK(array_type->Equals(arrow_type))
+      << "Inconsistent data type, expect " << arrow_type->ToString()
+      << ", but got " << array_type->ToString();
+  size_t cur_ind = 0;
+  for (auto j = 0; j < array->num_chunks(); ++j) {
+    auto casted = std::static_pointer_cast<arrow_array_type>(array->chunk(j));
+    for (auto k = 0; k < casted->length(); ++k) {
+      col->set_any(vids[cur_ind++],
+                   std::move(AnyConverter<COL_T>::to_any(casted->Value(k))));
+    }
+  }
+}
+
+// For String types.
+void set_vertex_column_from_string_array(
+    gs::ColumnBase* col, std::shared_ptr<arrow::ChunkedArray> array,
+    const std::vector<vid_t>& vids);
+
+void set_vertex_column_from_timestamp_array(
+    gs::ColumnBase* col, std::shared_ptr<arrow::ChunkedArray> array,
+    const std::vector<vid_t>& vids);
+
 void set_vertex_properties(gs::ColumnBase* col,
                            std::shared_ptr<arrow::ChunkedArray> array,
                            const std::vector<vid_t>& vids);
@@ -107,6 +137,7 @@ static void append_edges(
     std::shared_ptr<arrow::Array> dst_col, const LFIndexer<vid_t>& src_indexer,
     const LFIndexer<vid_t>& dst_indexer,
     std::vector<std::shared_ptr<arrow::Array>>& edata_cols,
+    const PropertyType& edge_prop,
     std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& parsed_edges,
     std::vector<int32_t>& ie_degree, std::vector<int32_t>& oe_degree) {
   CHECK(src_col->length() == dst_col->length());
@@ -415,30 +446,31 @@ class AbstractArrowFragmentLoader : public IFragmentLoader {
         }
         CHECK(property_cols.size() <= 1)
             << "Currently only support at most one property on edge";
-
+        auto edge_property =
+            schema_.get_edge_property(src_label_id, dst_label_id, e_label_id);
         // add edges to vector
         CHECK(src_col->length() == dst_col->length());
         if (src_col_type->Equals(arrow::int64())) {
-          append_edges<int64_t, EDATA_T>(src_col, dst_col, src_indexer,
-                                         dst_indexer, property_cols,
-                                         parsed_edges, ie_degree, oe_degree);
+          append_edges<int64_t, EDATA_T>(
+              src_col, dst_col, src_indexer, dst_indexer, property_cols,
+              edge_property, parsed_edges, ie_degree, oe_degree);
         } else if (src_col_type->Equals(arrow::uint64())) {
-          append_edges<uint64_t, EDATA_T>(src_col, dst_col, src_indexer,
-                                          dst_indexer, property_cols,
-                                          parsed_edges, ie_degree, oe_degree);
+          append_edges<uint64_t, EDATA_T>(
+              src_col, dst_col, src_indexer, dst_indexer, property_cols,
+              edge_property, parsed_edges, ie_degree, oe_degree);
         } else if (src_col_type->Equals(arrow::int32())) {
-          append_edges<int32_t, EDATA_T>(src_col, dst_col, src_indexer,
-                                         dst_indexer, property_cols,
-                                         parsed_edges, ie_degree, oe_degree);
+          append_edges<int32_t, EDATA_T>(
+              src_col, dst_col, src_indexer, dst_indexer, property_cols,
+              edge_property, parsed_edges, ie_degree, oe_degree);
         } else if (src_col_type->Equals(arrow::uint32())) {
-          append_edges<uint32_t, EDATA_T>(src_col, dst_col, src_indexer,
-                                          dst_indexer, property_cols,
-                                          parsed_edges, ie_degree, oe_degree);
+          append_edges<uint32_t, EDATA_T>(
+              src_col, dst_col, src_indexer, dst_indexer, property_cols,
+              edge_property, parsed_edges, ie_degree, oe_degree);
         } else {
           // must be string
           append_edges<std::string_view, EDATA_T>(
               src_col, dst_col, src_indexer, dst_indexer, property_cols,
-              parsed_edges, ie_degree, oe_degree);
+              edge_property, parsed_edges, ie_degree, oe_degree);
         }
         first_batch = false;
       }
diff --git a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
index 7765dab52b67..1d0d630f8da1 100644
--- a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
+++ b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
@@ -23,11 +23,12 @@
 namespace gs {
 
 template <typename EDATA_T>
-TypedMutableCsrBase<EDATA_T>* create_typed_csr(EdgeStrategy es) {
+TypedMutableCsrBase<EDATA_T>* create_typed_csr(EdgeStrategy es,
+                                               PropertyType edge_property) {
   if (es == EdgeStrategy::kSingle) {
-    return new SingleMutableCsr<EDATA_T>();
+    return new SingleMutableCsr<EDATA_T>(edge_property);
   } else if (es == EdgeStrategy::kMultiple) {
-    return new MutableCsr<EDATA_T>();
+    return new MutableCsr<EDATA_T>(edge_property);
   } else if (es == EdgeStrategy::kNone) {
     return new EmptyCsr<EDATA_T>();
   }
@@ -61,9 +62,12 @@ class BasicFragmentLoader {
     CHECK(v_label < vertex_label_num_);
     std::string filename =
         vertex_map_prefix(schema_.get_vertex_label_name(v_label));
+    auto primary_keys = schema_.get_vertex_primary_key(v_label);
+    auto type = std::get<0>(primary_keys[0]);
+
     build_lf_indexer<KEY_T, vid_t>(indexer, filename, lf_indexers_[v_label],
                                    snapshot_dir(work_dir_, 0),
-                                   tmp_dir(work_dir_));
+                                   tmp_dir(work_dir_), type);
   }
 
   template <typename EDATA_T>
@@ -80,7 +84,14 @@ class BasicFragmentLoader {
         src_label_name, dst_label_name, edge_label_name);
     EdgeStrategy ie_strategy = schema_.get_incoming_edge_strategy(
         src_label_name, dst_label_name, edge_label_name);
-    dual_csr_list_[index] = new DualCsr<EDATA_T>(oe_strategy, ie_strategy);
+    if constexpr (std::is_same_v<EDATA_T, std::string_view>) {
+      const auto& prop = schema_.get_edge_properties(src_label_id, dst_label_id,
+                                                     edge_label_id);
+      dual_csr_list_[index] = new DualCsr<std::string_view>(
+          oe_strategy, ie_strategy, prop[0].additional_type_info.max_length);
+    } else {
+      dual_csr_list_[index] = new DualCsr<EDATA_T>(oe_strategy, ie_strategy);
+    }
     ie_[index] = dual_csr_list_[index]->GetInCsr();
     oe_[index] = dual_csr_list_[index]->GetOutCsr();
     dual_csr_list_[index]->BatchInit(
@@ -110,20 +121,43 @@ class BasicFragmentLoader {
     EdgeStrategy ie_strategy = schema_.get_incoming_edge_strategy(
         src_label_name, dst_label_name, edge_label_name);
 
-    auto dual_csr = new DualCsr<EDATA_T>(oe_strategy, ie_strategy);
-    dual_csr_list_[index] = dual_csr;
-    ie_[index] = dual_csr_list_[index]->GetInCsr();
-    oe_[index] = dual_csr_list_[index]->GetOutCsr();
-    CHECK(ie_degree.size() == dst_indexer.size());
-    CHECK(oe_degree.size() == src_indexer.size());
-    dual_csr->BatchInit(
-        oe_prefix(src_label_name, dst_label_name, edge_label_name),
-        ie_prefix(src_label_name, dst_label_name, edge_label_name),
-        edata_prefix(src_label_name, dst_label_name, edge_label_name),
-        tmp_dir(work_dir_), oe_degree, ie_degree);
-    for (auto& edge : edges) {
-      dual_csr->BatchPutEdge(std::get<0>(edge), std::get<1>(edge),
-                             std::get<2>(edge));
+    if constexpr (std::is_same_v<EDATA_T, std::string_view>) {
+      const auto& prop = schema_.get_edge_properties(src_label_id, dst_label_id,
+                                                     edge_label_id);
+      auto dual_csr = new DualCsr<std::string_view>(
+          oe_strategy, ie_strategy, prop[0].additional_type_info.max_length);
+      dual_csr_list_[index] = dual_csr;
+      ie_[index] = dual_csr_list_[index]->GetInCsr();
+      oe_[index] = dual_csr_list_[index]->GetOutCsr();
+      CHECK(ie_degree.size() == dst_indexer.size());
+      CHECK(oe_degree.size() == src_indexer.size());
+      dual_csr->BatchInit(
+          oe_prefix(src_label_name, dst_label_name, edge_label_name),
+          ie_prefix(src_label_name, dst_label_name, edge_label_name),
+          edata_prefix(src_label_name, dst_label_name, edge_label_name),
+          tmp_dir(work_dir_), oe_degree, ie_degree);
+      for (auto& edge : edges) {
+        dual_csr->BatchPutEdge(std::get<0>(edge), std::get<1>(edge),
+                               std::get<2>(edge));
+      }
+
+    } else {
+      auto dual_csr = new DualCsr<EDATA_T>(oe_strategy, ie_strategy);
+
+      dual_csr_list_[index] = dual_csr;
+      ie_[index] = dual_csr_list_[index]->GetInCsr();
+      oe_[index] = dual_csr_list_[index]->GetOutCsr();
+      CHECK(ie_degree.size() == dst_indexer.size());
+      CHECK(oe_degree.size() == src_indexer.size());
+      dual_csr->BatchInit(
+          oe_prefix(src_label_name, dst_label_name, edge_label_name),
+          ie_prefix(src_label_name, dst_label_name, edge_label_name),
+          edata_prefix(src_label_name, dst_label_name, edge_label_name),
+          tmp_dir(work_dir_), oe_degree, ie_degree);
+      for (auto& edge : edges) {
+        dual_csr->BatchPutEdge(std::get<0>(edge), std::get<1>(edge),
+                               std::get<2>(edge));
+      }
     }
     VLOG(10) << "Finish adding edge batch of size: " << edges.size();
   }
diff --git a/flex/storages/rt_mutable_graph/mutable_csr.h b/flex/storages/rt_mutable_graph/mutable_csr.h
index c0e96bebdab5..7100dca2298b 100644
--- a/flex/storages/rt_mutable_graph/mutable_csr.h
+++ b/flex/storages/rt_mutable_graph/mutable_csr.h
@@ -495,7 +495,7 @@ class TypedMutableCsrConstEdgeIter : public MutableCsrConstEdgeIterBase {
 
  public:
   explicit TypedMutableCsrConstEdgeIter(const MutableNbrSlice<EDATA_T>& slice)
-      : cur_(slice.begin()), end_(slice.end()) {}
+      : cur_(slice.begin()), end_(slice.end()){}
   ~TypedMutableCsrConstEdgeIter() = default;
 
   vid_t get_neighbor() const override { return (*cur_).get_neighbor(); }
@@ -625,7 +625,8 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
   using slice_t = MutableNbrSlice<EDATA_T>;
   using mut_slice_t = MutableNbrSliceMut<EDATA_T>;
 
-  MutableCsr() : locks_(nullptr) {}
+  MutableCsr()
+      : locks_(nullptr) {}
   ~MutableCsr() {
     if (locks_ != nullptr) {
       delete[] locks_;
@@ -865,8 +866,7 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
 
   std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
       vid_t v) const override {
-    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(
-        get_edges(v));
+    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(get_edges(v));
   }
 
   MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
@@ -1277,8 +1277,7 @@ class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
 
   std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
       vid_t v) const override {
-    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(
-        get_edges(v));
+    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(get_edges(v));
   }
 
   MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
diff --git a/flex/storages/rt_mutable_graph/mutable_property_fragment.cc b/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
index 06afe4ef4921..690964f375aa 100644
--- a/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
+++ b/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
@@ -100,8 +100,9 @@ inline DualCsrBase* create_csr(EdgeStrategy oes, EdgeStrategy ies,
       return new DualCsr<double>(oes, ies);
     } else if (properties[0] == PropertyType::kFloat) {
       return new DualCsr<float>(oes, ies);
-    } else if (properties[0] == PropertyType::kString) {
-      return new DualCsr<std::string_view>(oes, ies);
+    } else if (properties[0].type_enum == impl::PropertyTypeImpl::kVarChar) {
+      return new DualCsr<std::string_view>(
+          oes, ies, properties[0].additional_type_info.max_length);
     }
   }
   LOG(FATAL) << "not support edge strategy or edge data type";
diff --git a/flex/storages/rt_mutable_graph/schema.cc b/flex/storages/rt_mutable_graph/schema.cc
index 12edcc30e656..9143b46d1472 100644
--- a/flex/storages/rt_mutable_graph/schema.cc
+++ b/flex/storages/rt_mutable_graph/schema.cc
@@ -395,7 +395,8 @@ static PropertyType StringToPropertyType(const std::string& str) {
   } else if (str == "Date" || str == DT_DATE) {
     return PropertyType::kDate;
   } else if (str == "String" || str == DT_STRING) {
-    return PropertyType::kString;
+    // DT_STRING is a alias for VARCHAR(STRING_DEFAULT_MAX_LENGTH);
+    return PropertyType::Varchar(PropertyType::STRING_DEFAULT_MAX_LENGTH);
   } else if (str == DT_STRINGMAP) {
     return PropertyType::kStringMap;
   } else if (str == "Empty") {
@@ -435,6 +436,30 @@ StorageStrategy StringToStorageStrategy(const std::string& str) {
   }
 }
 
+static bool parse_property_type(YAML::Node node, PropertyType& type) {
+  std::string prop_type_str{};
+  if (node["primitive_type"]) {
+    if (!get_scalar(node, "primitive_type", prop_type_str)) {
+      return false;
+    }
+  } else if (node["varchar"]) {
+    auto varchar_node = node["varchar"];
+    int length{};
+    if (!varchar_node["max_length"] ||
+        !get_scalar(varchar_node, "max_length", length)) {
+      return false;
+    }
+    type = PropertyType::Varchar(length);
+    return true;
+  } else if (node["date"]) {
+    auto format = node["date"].as<std::string>();
+    prop_type_str = DT_DATE;
+  } else {
+    return false;
+  }
+  type = StringToPropertyType(prop_type_str);
+  return true;
+}
 static bool parse_vertex_properties(YAML::Node node,
                                     const std::string& label_name,
                                     std::vector<PropertyType>& types,
@@ -452,9 +477,9 @@ static bool parse_vertex_properties(YAML::Node node,
   }
 
   for (int i = 0; i < prop_num; ++i) {
-    std::string prop_type_str, strategy_str, prop_name_str;
+    std::string strategy_str, prop_name_str;
     if (!get_scalar(node[i], "property_name", prop_name_str)) {
-      LOG(ERROR) << "name of vertex-" << label_name << " prop-" << i - 1
+      LOG(ERROR) << "Name of vertex-" << label_name << " prop-" << i - 1
                  << " is not specified...";
       return false;
     }
@@ -464,17 +489,9 @@ static bool parse_vertex_properties(YAML::Node node,
       return false;
     }
     auto prop_type_node = node[i]["property_type"];
-    if (prop_type_node["primitive_type"]) {
-      if (!get_scalar(prop_type_node, "primitive_type", prop_type_str)) {
-        LOG(ERROR) << "type of vertex-" << label_name << " prop-" << i - 1
-                   << " is not specified...";
-        return false;
-      }
-    } else if (prop_type_node["date"]) {
-      auto format = prop_type_node["date"].as<std::string>();
-      prop_type_str = DT_DATE;
-    } else {
-      LOG(ERROR) << "Unknown type of vertex-" << label_name << " prop-" << i - 1
+    PropertyType prop_type;
+    if (!parse_property_type(prop_type_node, prop_type)) {
+      LOG(ERROR) << "type of vertex-" << label_name << " prop-" << i - 1
                  << " is not specified...";
       return false;
     }
@@ -483,10 +500,10 @@ static bool parse_vertex_properties(YAML::Node node,
         get_scalar(node[i]["x_csr_params"], "storage_strategy", strategy_str);
       }
     }
-    types.push_back(StringToPropertyType(prop_type_str));
+    types.push_back(prop_type);
     strategies.push_back(StringToStorageStrategy(strategy_str));
     VLOG(10) << "prop-" << i - 1 << " name: " << prop_name_str
-             << " type: " << prop_type_str << " strategy: " << strategy_str;
+             << " type: " << prop_type << " strategy: " << strategy_str;
     names.push_back(prop_name_str);
   }
 
@@ -510,19 +527,15 @@ static bool parse_edge_properties(YAML::Node node,
   int prop_num = node.size();
 
   for (int i = 0; i < prop_num; ++i) {
-    std::string prop_type_str, strategy_str, prop_name_str;
-    if (node[i]["property_type"]) {
-      if (!get_scalar(node[i]["property_type"], "primitive_type",
-                      prop_type_str)) {
-        if (!get_scalar(node[i]["property_type"], "date", prop_type_str)) {
-          LOG(ERROR) << "Fail to parse property type of edge-" << label_name
-                     << " prop-" << i << " ...";
-          return false;
-        } else {
-          prop_type_str = DT_DATE;
-        }
-      }
-    } else {
+    std::string strategy_str, prop_name_str;
+    if (!node[i]["property_type"]) {
+      LOG(ERROR) << "type of edge-" << label_name << " prop-" << i - 1
+                 << " is not specified...";
+      return false;
+    }
+    auto prop_type_node = node[i]["property_type"];
+    PropertyType prop_type;
+    if (!parse_property_type(prop_type_node, prop_type)) {
       LOG(ERROR) << "type of edge-" << label_name << " prop-" << i - 1
                  << " is not specified...";
       return false;
@@ -533,7 +546,7 @@ static bool parse_edge_properties(YAML::Node node,
       return false;
     }
 
-    types.push_back(StringToPropertyType(prop_type_str));
+    types.push_back(prop_type);
     names.push_back(prop_name_str);
   }
 
diff --git a/flex/storages/rt_mutable_graph/schema.h b/flex/storages/rt_mutable_graph/schema.h
index 55cdee2f79c9..1b41311a28eb 100644
--- a/flex/storages/rt_mutable_graph/schema.h
+++ b/flex/storages/rt_mutable_graph/schema.h
@@ -30,6 +30,11 @@ class Schema {
   // How many built-in plugins are there.
   // Currently only one builtin plugin, SERVER_APP is supported.
   static constexpr uint8_t RESERVED_PLUGIN_NUM = 1;
+  static constexpr const char* PRIMITIVE_TYPE_KEY = "primitive_type";
+  static constexpr const char* VARCHAR_KEY = "varchar";
+  static constexpr const char* MAX_LENGTH_KEY = "max_length";
+  static constexpr const uint16_t STRING_DEFAULT_MAX_LENGTH = 256;
+
   using label_type = label_t;
   Schema();
   ~Schema();
diff --git a/flex/storages/rt_mutable_graph/types.h b/flex/storages/rt_mutable_graph/types.h
index f6c9dca2f3b2..6066fe69253d 100644
--- a/flex/storages/rt_mutable_graph/types.h
+++ b/flex/storages/rt_mutable_graph/types.h
@@ -30,6 +30,7 @@ using timestamp_t = uint32_t;
 using vid_t = uint32_t;
 using label_t = uint8_t;
 
+// primitive types
 static constexpr const char* DT_SIGNED_INT32 = "DT_SIGNED_INT32";
 static constexpr const char* DT_UNSIGNED_INT32 = "DT_UNSIGNED_INT32";
 static constexpr const char* DT_SIGNED_INT64 = "DT_SIGNED_INT64";
diff --git a/flex/tests/hqps/CMakeLists.txt b/flex/tests/hqps/CMakeLists.txt
index 31e072f2bb65..003e4e2f44f9 100644
--- a/flex/tests/hqps/CMakeLists.txt
+++ b/flex/tests/hqps/CMakeLists.txt
@@ -7,6 +7,6 @@ foreach(f ${GS_TEST_FILES})
         string(REGEX MATCH "^(.*)\\.[^.]*$" dummy ${f})
         set(T_NAME ${CMAKE_MATCH_1})
         message(STATUS "Found graphscope test - " ${T_NAME})
-        add_executable(${T_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/${T_NAME}.cc ${CMAKE_CURRENT_SOURCE_DIR}/../../utils/app_utils.cc)
-        target_link_libraries(${T_NAME}  hqps_plan_proto flex_rt_mutable_graph flex_graph_db  ${GLOG_LIBRARIES} ${LIBGRAPELITE_LIBRARIES})
+        add_executable(${T_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/${T_NAME}.cc)
+        target_link_libraries(${T_NAME}  hqps_plan_proto flex_rt_mutable_graph flex_graph_db  flex_utils ${GLOG_LIBRARIES} ${LIBGRAPELITE_LIBRARIES})
 endforeach()
diff --git a/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml b/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml
index 58576c0d309d..2e46a4ae64ed 100644
--- a/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml
+++ b/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml
@@ -14,7 +14,8 @@ schema:
         - property_id: 1
           property_name: name
           property_type:
-            primitive_type: DT_STRING
+            varchar:
+              max_length: 64
         - property_id: 2
           property_name: age
           property_type:
@@ -33,11 +34,13 @@ schema:
         - property_id: 1
           property_name: name
           property_type:
-            primitive_type: DT_STRING
+            varchar:
+              max_length: 64
         - property_id: 2
           property_name: lang
           property_type:
-            primitive_type: DT_STRING
+            varchar:
+              max_length: 64
       primary_keys:
         - id
   edge_types:
@@ -68,4 +71,5 @@ schema:
         - property_id: 0
           property_name: weight
           property_type:
-            primitive_type: DT_STRING
\ No newline at end of file
+            varchar:
+              max_length: 64
\ No newline at end of file
diff --git a/flex/tests/rt_mutable_graph/test_acid.cc b/flex/tests/rt_mutable_graph/test_acid.cc
index d3a9bc275e77..a076384e07ed 100644
--- a/flex/tests/rt_mutable_graph/test_acid.cc
+++ b/flex/tests/rt_mutable_graph/test_acid.cc
@@ -129,9 +129,9 @@ void AtomicityInit(GraphDB& db, const std::string& work_dir, int thread_num) {
   Schema schema;
   schema.add_vertex_label("PERSON",
                           {
-                              PropertyType::kInt64,
-                              PropertyType::kString,  // name
-                              PropertyType::kString,  // emails
+                              PropertyType::kInt64,        // id
+                              PropertyType::Varchar(256),  // name
+                              PropertyType::Varchar(256),  // emails
                           },
                           {"id", "name", "emails"},
                           {std::tuple<gs::PropertyType, std::string, size_t>(
@@ -341,8 +341,8 @@ void G0Init(GraphDB& db, const std::string& work_dir, int thread_num) {
   Schema schema;
   schema.add_vertex_label("PERSON",
                           {
-                              PropertyType::kInt64,   // id
-                              PropertyType::kString,  // version history
+                              PropertyType::kInt64,        // id
+                              PropertyType::Varchar(256),  // version history
                           },
                           {
                               "id",
@@ -359,7 +359,7 @@ void G0Init(GraphDB& db, const std::string& work_dir, int thread_num) {
                           4096);
   schema.add_edge_label("PERSON", "PERSON", "KNOWS",
                         {
-                            PropertyType::kString,  // version history
+                            PropertyType::Varchar(256),  // version history
                         },
                         {}, EdgeStrategy::kMultiple, EdgeStrategy::kMultiple);
   db.Open(schema, work_dir, thread_num);
@@ -470,7 +470,7 @@ std::tuple<std::string, std::string, std::string> G0Check(GraphDB& db,
   }
   CHECK(oeit.IsValid());
   Any k_version_history_field = oeit.GetData();
-  CHECK(k_version_history_field.type == PropertyType::kString);
+  CHECK(k_version_history_field.type == PropertyType::Varchar(256));
   std::string k_version_history(k_version_history_field.value.s);
 
   return std::make_tuple(p1_version_history, p2_version_history,
@@ -594,7 +594,7 @@ void G1BInit(GraphDB& db, const std::string& work_dir, int thread_num) {
 
   schema.add_vertex_label("PERSON",
                           {
-                              PropertyType::kInt64,
+                              PropertyType::kInt64,  // id
                               PropertyType::kInt64,  // version
                           },
                           {"id", "version"},
@@ -667,7 +667,7 @@ void G1CInit(GraphDB& db, const std::string& work_dir, int thread_num) {
   Schema schema;
   schema.add_vertex_label("PERSON",
                           {
-                              PropertyType::kInt64,
+                              PropertyType::kInt64,  // id
                               PropertyType::kInt64,  // version
                           },
                           {"id", "version"},
@@ -1004,9 +1004,9 @@ void OTVInit(GraphDB& db, const std::string& work_dir, int thread_num) {
   Schema schema;
   schema.add_vertex_label("PERSON",
                           {
-                              PropertyType::kInt64,   // id
-                              PropertyType::kString,  // name
-                              PropertyType::kInt64,   // version
+                              PropertyType::kInt64,        // id
+                              PropertyType::Varchar(256),  // name
+                              PropertyType::kInt64,        // version
                           },
                           {"id", "name", "version"},
                           {std::tuple<gs::PropertyType, std::string, size_t>(
diff --git a/flex/tests/rt_mutable_graph/test_empty_graph.cc b/flex/tests/rt_mutable_graph/test_empty_graph.cc
index 5de453aa2e02..6736a309ec95 100644
--- a/flex/tests/rt_mutable_graph/test_empty_graph.cc
+++ b/flex/tests/rt_mutable_graph/test_empty_graph.cc
@@ -32,8 +32,8 @@ int main(int argc, char** argv) {
   schema.add_vertex_label(
       "PERSON",
       {
-          gs::PropertyType::kString,  // name
-          gs::PropertyType::kString,  // emails
+          gs::PropertyType::Varchar(16),  // name
+          gs::PropertyType::Varchar(32),  // emails
       },
       {"name", "emails"},
       {std::tuple<gs::PropertyType, std::string, size_t>(
diff --git a/flex/utils/CMakeLists.txt b/flex/utils/CMakeLists.txt
index a54e7c69f617..d073671bcbed 100644
--- a/flex/utils/CMakeLists.txt
+++ b/flex/utils/CMakeLists.txt
@@ -1,6 +1,12 @@
 
 file(GLOB_RECURSE UTILS_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
 add_library(flex_utils SHARED ${UTILS_SRC_FILES})
+if (ARROW_SHARED_LIB)
+        target_link_libraries(flex_utils ${ARROW_SHARED_LIB})
+else()
+        target_link_libraries(flex_utils ${ARROW_STATIC_LIB})
+endif()
+target_link_libraries(flex_utils ${YAML_CPP_LIBRARIES})
 
 install(TARGETS flex_utils
         RUNTIME DESTINATION bin
diff --git a/flex/utils/arrow_utils.cc b/flex/utils/arrow_utils.cc
index 3563f28c059d..5cb9749386e3 100644
--- a/flex/utils/arrow_utils.cc
+++ b/flex/utils/arrow_utils.cc
@@ -16,31 +16,33 @@
 
 namespace gs {
 std::shared_ptr<arrow::DataType> PropertyTypeToArrowType(PropertyType type) {
-  switch (type) {
-  case PropertyType::kBool:
+  if (type == PropertyType::Bool()) {
     return arrow::boolean();
-  case PropertyType::kInt32:
+  } else if (type == PropertyType::Int32()) {
     return arrow::int32();
-  case PropertyType::kInt64:
+  } else if (type == PropertyType::Int64()) {
     return arrow::int64();
-  case PropertyType::kUInt32:
+  } else if (type == PropertyType::UInt32()) {
     return arrow::uint32();
-  case PropertyType::kUInt64:
+  } else if (type == PropertyType::UInt64()) {
     return arrow::uint64();
-  case PropertyType::kDouble:
+  } else if (type == PropertyType::Double()) {
     return arrow::float64();
-  case PropertyType::kFloat:
+  } else if (type == PropertyType::Float()) {
     return arrow::float32();
-  case PropertyType::kDate:
+  } else if (type == PropertyType::Date()) {
     return arrow::timestamp(arrow::TimeUnit::MILLI);
-  case PropertyType::kString:
+  } else if (type == PropertyType::String()) {
     return arrow::large_utf8();
-  case PropertyType::kStringMap:
+  } else if (type == PropertyType::StringMap()) {
     return arrow::large_utf8();
-  case PropertyType::kEmpty:
+  } else if (type == PropertyType::Empty()) {
     return arrow::null();
-  default:
-    LOG(FATAL) << "Unexpected property type: " << static_cast<int>(type);
+  } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    return arrow::large_utf8();
+  } else {
+    LOG(FATAL) << "Unexpected property type: "
+               << static_cast<int>(type.type_enum);
     return nullptr;
   }
 }
diff --git a/flex/utils/arrow_utils.h b/flex/utils/arrow_utils.h
index 75a3afec3778..edce99d78e37 100644
--- a/flex/utils/arrow_utils.h
+++ b/flex/utils/arrow_utils.h
@@ -184,7 +184,7 @@ struct TypeConverter;
 
 template <>
 struct TypeConverter<bool> {
-  static constexpr PropertyType property_type = PropertyType::kBool;
+  static PropertyType property_type() { return PropertyType::kBool; }
   using ArrowType = arrow::BooleanType;
   using ArrowArrayType = arrow::BooleanArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -194,7 +194,7 @@ struct TypeConverter<bool> {
 
 template <>
 struct TypeConverter<int32_t> {
-  static constexpr PropertyType property_type = PropertyType::kInt32;
+  static PropertyType property_type() { return PropertyType::kInt32; }
   using ArrowType = arrow::Int32Type;
   using ArrowArrayType = arrow::Int32Array;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -204,7 +204,7 @@ struct TypeConverter<int32_t> {
 
 template <>
 struct TypeConverter<uint32_t> {
-  static constexpr PropertyType property_type = PropertyType::kUInt32;
+  static PropertyType property_type() { return PropertyType::kUInt32; }
   using ArrowType = arrow::UInt32Type;
   using ArrowArrayType = arrow::UInt32Array;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -214,7 +214,7 @@ struct TypeConverter<uint32_t> {
 
 template <>
 struct TypeConverter<int64_t> {
-  static constexpr PropertyType property_type = PropertyType::kInt64;
+  static PropertyType property_type() { return PropertyType::kInt64; }
   using ArrowType = arrow::Int64Type;
   using ArrowArrayType = arrow::Int64Array;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -224,7 +224,7 @@ struct TypeConverter<int64_t> {
 
 template <>
 struct TypeConverter<uint64_t> {
-  static constexpr PropertyType property_type = PropertyType::kUInt64;
+  static PropertyType property_type() { return PropertyType::kUInt64; }
   using ArrowType = arrow::UInt64Type;
   using ArrowArrayType = arrow::UInt64Array;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -234,7 +234,7 @@ struct TypeConverter<uint64_t> {
 
 template <>
 struct TypeConverter<double> {
-  static constexpr PropertyType property_type = PropertyType::kDouble;
+  static PropertyType property_type() { return PropertyType::kDouble; }
   using ArrowType = arrow::DoubleType;
   using ArrowArrayType = arrow::DoubleArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -244,7 +244,7 @@ struct TypeConverter<double> {
 
 template <>
 struct TypeConverter<float> {
-  static constexpr PropertyType property_type = PropertyType::kFloat;
+  static PropertyType property_type() { return PropertyType::kFloat; }
   using ArrowType = arrow::FloatType;
   using ArrowArrayType = arrow::FloatArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -253,7 +253,7 @@ struct TypeConverter<float> {
 };
 template <>
 struct TypeConverter<std::string> {
-  static constexpr PropertyType property_type = PropertyType::kString;
+  static PropertyType property_type() { return PropertyType::kString; }
   using ArrowType = arrow::LargeStringType;
   using ArrowArrayType = arrow::LargeStringArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -263,7 +263,7 @@ struct TypeConverter<std::string> {
 
 template <>
 struct TypeConverter<std::string_view> {
-  static constexpr PropertyType property_type = PropertyType::kString;
+  static PropertyType property_type() { return PropertyType::kString; }
   using ArrowType = arrow::LargeStringType;
   using ArrowArrayType = arrow::LargeStringArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
@@ -273,7 +273,7 @@ struct TypeConverter<std::string_view> {
 
 template <>
 struct TypeConverter<Date> {
-  static constexpr PropertyType property_type = PropertyType::kDate;
+  static PropertyType property_type() { return PropertyType::kDate; }
   using ArrowType = arrow::TimestampType;
   using ArrowArrayType = arrow::TimestampArray;
   static std::shared_ptr<arrow::DataType> ArrowTypeValue() {
diff --git a/flex/utils/id_indexer.h b/flex/utils/id_indexer.h
index 9cc0bfae1bec..fd7d4ee17924 100644
--- a/flex/utils/id_indexer.h
+++ b/flex/utils/id_indexer.h
@@ -187,7 +187,7 @@ template <typename KEY_T, typename INDEX_T>
 void build_lf_indexer(const IdIndexer<KEY_T, INDEX_T>& input,
                       const std::string& filename, LFIndexer<INDEX_T>& lf,
                       const std::string& snapshot_dir,
-                      const std::string& work_dir);
+                      const std::string& work_dir, PropertyType type);
 
 template <typename INDEX_T>
 class LFIndexer {
@@ -221,10 +221,11 @@ class LFIndexer {
       keys_ = new TypedColumn<uint64_t>(StorageStrategy::kMem);
     } else if (type == PropertyType::kUInt32) {
       keys_ = new TypedColumn<uint32_t>(StorageStrategy::kMem);
-    } else if (type == PropertyType::kString) {
-      keys_ = new StringColumn(StorageStrategy::kMem);
     } else if (type == PropertyType::kStringMap) {
       keys_ = new StringMapColumn<uint8_t>(StorageStrategy::kMem);
+    } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+      keys_ = new StringColumn(StorageStrategy::kMem,
+                               type.additional_type_info.max_length);
     } else {
       LOG(FATAL) << "Not support type [" << type << "] as pk type ..";
     }
@@ -232,12 +233,15 @@ class LFIndexer {
 
   void build_empty_LFIndexer(const std::string& filename,
                              const std::string& snapshot_dir,
-                             const std::string& work_dir, size_t size) {
+                             const std::string& work_dir) {
     keys_->open(filename + ".keys", "", work_dir);
     indices_.open(work_dir + "/" + filename + ".indices", false);
 
     num_elements_.store(0);
     indices_size_ = 0;
+    dump_meta(work_dir + "/" + filename + ".meta");
+    indices_.reset();
+    keys_->close();
   }
 
   void reserve(size_t size) { rehash(std::max(size, num_elements_.load())); }
@@ -344,20 +348,12 @@ class LFIndexer {
 
   void open(const std::string& name, const std::string& snapshot_dir,
             const std::string& work_dir) {
-    if (!std::filesystem::exists(work_dir + "/" + name + ".meta")) {
-      if (std::filesystem::exists(snapshot_dir + "/" + name + ".meta")) {
-        copy_to_tmp(snapshot_dir + "/" + name, work_dir + "/" + name);
-
-      } else {
-        build_empty_LFIndexer(name, "", work_dir, 0);
-        num_elements_.store(0);
-        indices_.open(work_dir + "/" + name + ".indices", false);
-        indices_size_ = indices_.size();
-        dump_meta(work_dir + "/" + name + ".meta");
-        indices_.reset();
-        keys_->close();
-      }
+    if (std::filesystem::exists(snapshot_dir + "/" + name + ".meta")) {
+      copy_to_tmp(snapshot_dir + "/" + name, work_dir + "/" + name);
+    } else {
+      build_empty_LFIndexer(name, "", work_dir);
     }
+
     load_meta(work_dir + "/" + name + ".meta");
     keys_->open(name + ".keys", "", work_dir);
     indices_.open(work_dir + "/" + name + ".indices", false);
@@ -471,7 +467,7 @@ class LFIndexer {
                                const std::string& filename,
                                LFIndexer<_INDEX_T>& output,
                                const std::string& snapshot_dir,
-                               const std::string& work_dir);
+                               const std::string& work_dir, PropertyType type);
 };
 
 template <typename INDEX_T>
@@ -496,7 +492,7 @@ class IdIndexer : public IdIndexerBase<INDEX_T> {
   IdIndexer() : hasher_() { reset_to_empty_state(); }
   ~IdIndexer() {}
 
-  PropertyType get_type() const override { return AnyConverter<KEY_T>::type; }
+  PropertyType get_type() const override { return AnyConverter<KEY_T>::type(); }
 
   void _add(const Any& oid) override {
     assert(get_type() == oid.type);
@@ -897,7 +893,7 @@ class IdIndexer : public IdIndexerBase<INDEX_T> {
                                const std::string& filename,
                                LFIndexer<_INDEX_T>& output,
                                const std::string& snapshot_dir,
-                               const std::string& work_dir);
+                               const std::string& work_dir, PropertyType type);
 };
 
 template <typename KEY_T, typename INDEX_T>
@@ -927,9 +923,9 @@ template <typename KEY_T, typename INDEX_T>
 void build_lf_indexer(const IdIndexer<KEY_T, INDEX_T>& input,
                       const std::string& filename, LFIndexer<INDEX_T>& lf,
                       const std::string& snapshot_dir,
-                      const std::string& work_dir) {
+                      const std::string& work_dir, PropertyType type) {
   size_t size = input.keys_.size();
-  lf.init(AnyConverter<KEY_T>::type);
+  lf.init(type);
   lf.keys_->open(filename + ".keys", "", work_dir);
   lf.keys_->resize(size);
   _move_data<KEY_T, INDEX_T>()(input.keys_, *lf.keys_, size);
diff --git a/flex/utils/property/column.cc b/flex/utils/property/column.cc
index 88ae73a4e27d..8829c5f9cb71 100644
--- a/flex/utils/property/column.cc
+++ b/flex/utils/property/column.cc
@@ -37,7 +37,7 @@ class TypedEmptyColumn : public ColumnBase {
   size_t size() const override { return 0; }
   void resize(size_t size) override {}
 
-  PropertyType type() const override { return AnyConverter<T>::type; }
+  PropertyType type() const override { return AnyConverter<T>::type(); }
 
   void set_value(size_t index, const T& val) {}
 
@@ -57,6 +57,46 @@ class TypedEmptyColumn : public ColumnBase {
   }
 };
 
+template <>
+class TypedEmptyColumn<std::string_view> : public ColumnBase {
+ public:
+  TypedEmptyColumn(int32_t max_length = PropertyType::STRING_DEFAULT_MAX_LENGTH)
+      : max_length_(max_length) {}
+  ~TypedEmptyColumn() {}
+
+  void open(const std::string& name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {}
+  void touch(const std::string& filename) override {}
+  void dump(const std::string& filename) override {}
+  void copy_to_tmp(const std::string& cur_path,
+                   const std::string& tmp_path) override {}
+  void close() override {}
+  size_t size() const override { return 0; }
+  void resize(size_t size) override {}
+
+  PropertyType type() const override { return PropertyType::kString; }
+
+  void set_value(size_t index, const std::string_view& val) {}
+
+  void set_any(size_t index, const Any& value) override {}
+
+  std::string_view get_view(size_t index) const { return std::string_view{}; }
+
+  Any get(size_t index) const override { return Any(); }
+
+  void ingest(uint32_t index, grape::OutArchive& arc) override {
+    std::string_view val;
+    arc >> val;
+  }
+
+  StorageStrategy storage_strategy() const override {
+    return StorageStrategy::kNone;
+  }
+
+ private:
+  int32_t max_length_;
+};
+
 using IntEmptyColumn = TypedEmptyColumn<int32_t>;
 using UIntEmptyColumn = TypedEmptyColumn<uint32_t>;
 using LongEmptyColumn = TypedEmptyColumn<int64_t>;
@@ -86,13 +126,14 @@ std::shared_ptr<ColumnBase> CreateColumn(PropertyType type,
       return std::make_shared<FloatEmptyColumn>();
     } else if (type == PropertyType::kDate) {
       return std::make_shared<DateEmptyColumn>();
-    } else if (type == PropertyType::kString) {
-      return std::make_shared<StringEmptyColumn>();
     } else if (type == PropertyType::kStringMap) {
       return std::make_shared<StringEmptyColumn>();
+    } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+      return std::make_shared<StringEmptyColumn>(
+          type.additional_type_info.max_length);
     } else {
       LOG(FATAL) << "unexpected type to create column, "
-                 << static_cast<int>(type);
+                 << static_cast<int>(type.type_enum);
       return nullptr;
     }
   } else {
@@ -112,13 +153,14 @@ std::shared_ptr<ColumnBase> CreateColumn(PropertyType type,
       return std::make_shared<FloatColumn>(strategy);
     } else if (type == PropertyType::kDate) {
       return std::make_shared<DateColumn>(strategy);
-    } else if (type == PropertyType::kString) {
-      return std::make_shared<StringColumn>(strategy);
     } else if (type == PropertyType::kStringMap) {
       return std::make_shared<StringMapColumn<uint8_t>>(strategy);
+    } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+      return std::make_shared<StringColumn>(
+          strategy, type.additional_type_info.max_length);
     } else {
       LOG(FATAL) << "unexpected type to create column, "
-                 << static_cast<int>(type);
+                 << static_cast<int>(type.type_enum);
       return nullptr;
     }
   }
diff --git a/flex/utils/property/column.h b/flex/utils/property/column.h
index c32f5c9134e8..1810289ed1cf 100644
--- a/flex/utils/property/column.h
+++ b/flex/utils/property/column.h
@@ -144,7 +144,7 @@ class TypedColumn : public ColumnBase {
     }
   }
 
-  PropertyType type() const override { return AnyConverter<T>::type; }
+  PropertyType type() const override { return AnyConverter<T>::type(); }
 
   void set_value(size_t index, const T& val) {
     assert(index >= basic_size_ && index < basic_size_ + extra_size_);
@@ -197,7 +197,9 @@ using FloatColumn = TypedColumn<float>;
 template <>
 class TypedColumn<std::string_view> : public ColumnBase {
  public:
-  TypedColumn(StorageStrategy strategy, size_t width = 64) : width_(width) {}
+  TypedColumn(StorageStrategy strategy,
+              uint16_t width = PropertyType::STRING_DEFAULT_MAX_LENGTH)
+      : width_(width) {}
   ~TypedColumn() { close(); }
 
   void open(const std::string& name, const std::string& snapshot_dir,
@@ -206,6 +208,7 @@ class TypedColumn<std::string_view> : public ColumnBase {
     if (std::filesystem::exists(basic_path + ".items")) {
       basic_buffer_.open(basic_path, true);
       basic_size_ = basic_buffer_.size();
+
     } else {
       basic_size_ = 0;
     }
@@ -215,7 +218,6 @@ class TypedColumn<std::string_view> : public ColumnBase {
     } else {
       extra_buffer_.open(work_dir + "/" + name, false);
       extra_size_ = extra_buffer_.size();
-
       pos_.store(extra_buffer_.data_size());
     }
   }
@@ -308,9 +310,7 @@ class TypedColumn<std::string_view> : public ColumnBase {
     }
   }
 
-  PropertyType type() const override {
-    return AnyConverter<std::string_view>::type;
-  }
+  PropertyType type() const override { return PropertyType::Varchar(width_); }
 
   void set_value(size_t idx, const std::string_view& val) {
     assert(idx >= basic_size_ && idx < basic_size_ + extra_size_);
@@ -357,7 +357,7 @@ class TypedColumn<std::string_view> : public ColumnBase {
   size_t extra_size_;
   std::atomic<size_t> pos_;
   StorageStrategy strategy_;
-  size_t width_;
+  uint16_t width_;
 };
 
 using StringColumn = TypedColumn<std::string_view>;
diff --git a/flex/utils/property/types.cc b/flex/utils/property/types.cc
index 3223740c61f1..b8cf1e9266f5 100644
--- a/flex/utils/property/types.cc
+++ b/flex/utils/property/types.cc
@@ -20,38 +20,145 @@
 
 namespace gs {
 
+const PropertyType PropertyType::kEmpty =
+    PropertyType(impl::PropertyTypeImpl::kEmpty);
+const PropertyType PropertyType::kBool =
+    PropertyType(impl::PropertyTypeImpl::kBool);
+const PropertyType PropertyType::kUInt8 =
+    PropertyType(impl::PropertyTypeImpl::kUInt8);
+const PropertyType PropertyType::kUInt16 =
+    PropertyType(impl::PropertyTypeImpl::kUInt16);
+const PropertyType PropertyType::kInt32 =
+    PropertyType(impl::PropertyTypeImpl::kInt32);
+const PropertyType PropertyType::kUInt32 =
+    PropertyType(impl::PropertyTypeImpl::kUInt32);
+const PropertyType PropertyType::kFloat =
+    PropertyType(impl::PropertyTypeImpl::kFloat);
+const PropertyType PropertyType::kInt64 =
+    PropertyType(impl::PropertyTypeImpl::kInt64);
+const PropertyType PropertyType::kUInt64 =
+    PropertyType(impl::PropertyTypeImpl::kUInt64);
+const PropertyType PropertyType::kDouble =
+    PropertyType(impl::PropertyTypeImpl::kDouble);
+const PropertyType PropertyType::kDate =
+    PropertyType(impl::PropertyTypeImpl::kDate);
+const PropertyType PropertyType::kString =
+    PropertyType(impl::PropertyTypeImpl::kString);
+const PropertyType PropertyType::kStringMap =
+    PropertyType(impl::PropertyTypeImpl::kStringMap);
+
+bool PropertyType::operator==(const PropertyType& other) const {
+  if (type_enum == impl::PropertyTypeImpl::kVarChar &&
+      other.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    return additional_type_info.max_length ==
+           other.additional_type_info.max_length;
+  }
+  if ((type_enum == impl::PropertyTypeImpl::kString &&
+       other.type_enum == impl::PropertyTypeImpl::kVarChar) ||
+      (type_enum == impl::PropertyTypeImpl::kVarChar &&
+       other.type_enum == impl::PropertyTypeImpl::kString)) {
+    return true;
+  }
+  return type_enum == other.type_enum;
+}
+
+bool PropertyType::operator!=(const PropertyType& other) const {
+  return !(*this == other);
+}
+
+bool PropertyType::IsVarchar() const {
+  return type_enum == impl::PropertyTypeImpl::kVarChar;
+}
+
+/////////////////////////////// Get Type Instance
+//////////////////////////////////
+PropertyType PropertyType::Empty() {
+  return PropertyType(impl::PropertyTypeImpl::kEmpty);
+}
+PropertyType PropertyType::Bool() {
+  return PropertyType(impl::PropertyTypeImpl::kBool);
+}
+PropertyType PropertyType::UInt8() {
+  return PropertyType(impl::PropertyTypeImpl::kUInt8);
+}
+PropertyType PropertyType::UInt16() {
+  return PropertyType(impl::PropertyTypeImpl::kUInt16);
+}
+PropertyType PropertyType::Int32() {
+  return PropertyType(impl::PropertyTypeImpl::kInt32);
+}
+PropertyType PropertyType::UInt32() {
+  return PropertyType(impl::PropertyTypeImpl::kUInt32);
+}
+PropertyType PropertyType::Float() {
+  return PropertyType(impl::PropertyTypeImpl::kFloat);
+}
+PropertyType PropertyType::Int64() {
+  return PropertyType(impl::PropertyTypeImpl::kInt64);
+}
+PropertyType PropertyType::UInt64() {
+  return PropertyType(impl::PropertyTypeImpl::kUInt64);
+}
+PropertyType PropertyType::Double() {
+  return PropertyType(impl::PropertyTypeImpl::kDouble);
+}
+PropertyType PropertyType::Date() {
+  return PropertyType(impl::PropertyTypeImpl::kDate);
+}
+PropertyType PropertyType::String() {
+  return PropertyType(impl::PropertyTypeImpl::kString);
+}
+PropertyType PropertyType::StringMap() {
+  return PropertyType(impl::PropertyTypeImpl::kStringMap);
+}
+PropertyType PropertyType::Varchar(uint16_t max_length) {
+  return PropertyType(impl::PropertyTypeImpl::kVarChar, max_length);
+}
+
+grape::InArchive& operator<<(grape::InArchive& in_archive,
+                             const PropertyType& value) {
+  in_archive << value.type_enum;
+  if (value.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    in_archive << value.additional_type_info.max_length;
+  }
+  return in_archive;
+}
+grape::OutArchive& operator>>(grape::OutArchive& out_archive,
+                              PropertyType& value) {
+  out_archive >> value.type_enum;
+  if (value.type_enum == impl::PropertyTypeImpl::kVarChar) {
+    out_archive >> value.additional_type_info.max_length;
+  }
+  return out_archive;
+}
+
 grape::InArchive& operator<<(grape::InArchive& in_archive, const Any& value) {
-  switch (value.type) {
-  case PropertyType::kBool:
+  if (value.type == PropertyType::Empty()) {
+    in_archive << value.type;
+  } else if (value.type == PropertyType::Bool()) {
     in_archive << value.type << value.value.b;
-    break;
-  case PropertyType::kInt32:
+  } else if (value.type == PropertyType::UInt8()) {
+    in_archive << value.type << value.value.u8;
+  } else if (value.type == PropertyType::UInt16()) {
+    in_archive << value.type << value.value.u16;
+  } else if (value.type == PropertyType::Int32()) {
     in_archive << value.type << value.value.i;
-    break;
-  case PropertyType::kInt64:
-    in_archive << value.type << value.value.l;
-    break;
-  case PropertyType::kUInt32:
+  } else if (value.type == PropertyType::UInt32()) {
     in_archive << value.type << value.value.ui;
-    break;
-  case PropertyType::kUInt64:
+  } else if (value.type == PropertyType::Float()) {
+    in_archive << value.type << value.value.f;
+  } else if (value.type == PropertyType::Int64()) {
+    in_archive << value.type << value.value.l;
+  } else if (value.type == PropertyType::UInt64()) {
     in_archive << value.type << value.value.ul;
-    break;
-  case PropertyType::kDate:
+  } else if (value.type == PropertyType::Double()) {
+    in_archive << value.type << value.value.db;
+  } else if (value.type == PropertyType::Date()) {
     in_archive << value.type << value.value.d.milli_second;
-    break;
-  case PropertyType::kString:
+  } else if (value.type == PropertyType::String()) {
     in_archive << value.type << value.value.s;
-    break;
-  case PropertyType::kDouble:
-    in_archive << value.type << value.value.db;
-    break;
-  case PropertyType::kFloat:
-    in_archive << value.type << value.value.f;
-    break;
-  default:
+  } else {
     in_archive << PropertyType::kEmpty;
-    break;
   }
 
   return in_archive;
@@ -59,36 +166,31 @@ grape::InArchive& operator<<(grape::InArchive& in_archive, const Any& value) {
 
 grape::OutArchive& operator>>(grape::OutArchive& out_archive, Any& value) {
   out_archive >> value.type;
-  switch (value.type) {
-  case PropertyType::kBool:
+  if (value.type == PropertyType::Empty()) {
+  } else if (value.type == PropertyType::Bool()) {
     out_archive >> value.value.b;
-    break;
-  case PropertyType::kInt32:
+  } else if (value.type == PropertyType::UInt8()) {
+    out_archive >> value.value.u8;
+  } else if (value.type == PropertyType::UInt16()) {
+    out_archive >> value.value.u16;
+  } else if (value.type == PropertyType::Int32()) {
     out_archive >> value.value.i;
-    break;
-  case PropertyType::kInt64:
-    out_archive >> value.value.l;
-    break;
-  case PropertyType::kUInt32:
+  } else if (value.type == PropertyType::UInt32()) {
     out_archive >> value.value.ui;
-    break;
-  case PropertyType::kUInt64:
+  } else if (value.type == PropertyType::Float()) {
+    out_archive >> value.value.f;
+  } else if (value.type == PropertyType::Int64()) {
+    out_archive >> value.value.l;
+  } else if (value.type == PropertyType::UInt64()) {
     out_archive >> value.value.ul;
-    break;
-  case PropertyType::kDate:
+  } else if (value.type == PropertyType::Double()) {
+    out_archive >> value.value.db;
+  } else if (value.type == PropertyType::Date()) {
     out_archive >> value.value.d.milli_second;
-    break;
-  case PropertyType::kString:
+  } else if (value.type == PropertyType::String()) {
     out_archive >> value.value.s;
-    break;
-  case PropertyType::kDouble:
-    out_archive >> value.value.db;
-    break;
-  case PropertyType::kFloat:
-    out_archive >> value.value.f;
-    break;
-  default:
-    break;
+  } else {
+    value.type = PropertyType::kEmpty;
   }
 
   return out_archive;
diff --git a/flex/utils/property/types.h b/flex/utils/property/types.h
index d352bb0eaa6b..8ac36222ff98 100644
--- a/flex/utils/property/types.h
+++ b/flex/utils/property/types.h
@@ -32,7 +32,9 @@ enum class StorageStrategy {
   kMem,
 };
 
-enum class PropertyType {
+namespace impl {
+
+enum class PropertyTypeImpl {
   kInt32,
   kDate,
   kString,
@@ -46,6 +48,62 @@ enum class PropertyType {
   kUInt8,
   kUInt16,
   kStringMap,
+  kVarChar,
+};
+
+// Stores additional type information for PropertyTypeImpl
+union AdditionalTypeInfo {
+  uint16_t max_length;  // for varchar
+};
+}  // namespace impl
+
+struct PropertyType {
+  static constexpr const uint16_t STRING_DEFAULT_MAX_LENGTH = 256;
+  impl::PropertyTypeImpl type_enum;
+  impl::AdditionalTypeInfo additional_type_info;
+
+  constexpr PropertyType()
+      : type_enum(impl::PropertyTypeImpl::kEmpty), additional_type_info() {}
+  constexpr PropertyType(impl::PropertyTypeImpl type)
+      : type_enum(type), additional_type_info() {}
+  constexpr PropertyType(impl::PropertyTypeImpl type, uint16_t max_length)
+      : type_enum(type), additional_type_info({.max_length = max_length}) {
+    assert(type == impl::PropertyTypeImpl::kVarChar);
+  }
+
+  bool IsVarchar() const;
+
+  static PropertyType Empty();
+  static PropertyType Bool();
+  static PropertyType UInt8();
+  static PropertyType UInt16();
+  static PropertyType Int32();
+  static PropertyType UInt32();
+  static PropertyType Float();
+  static PropertyType Int64();
+  static PropertyType UInt64();
+  static PropertyType Double();
+  static PropertyType Date();
+  static PropertyType String();
+  static PropertyType StringMap();
+  static PropertyType Varchar(uint16_t max_length);
+
+  static const PropertyType kEmpty;
+  static const PropertyType kBool;
+  static const PropertyType kUInt8;
+  static const PropertyType kUInt16;
+  static const PropertyType kInt32;
+  static const PropertyType kUInt32;
+  static const PropertyType kFloat;
+  static const PropertyType kInt64;
+  static const PropertyType kUInt64;
+  static const PropertyType kDouble;
+  static const PropertyType kDate;
+  static const PropertyType kString;
+  static const PropertyType kStringMap;
+
+  bool operator==(const PropertyType& other) const;
+  bool operator!=(const PropertyType& other) const;
 };
 
 struct Date {
@@ -180,7 +238,8 @@ struct Any {
     } else if (type == PropertyType::kFloat) {
       return std::to_string(value.f);
     } else {
-      LOG(FATAL) << "Unexpected property type: " << static_cast<int>(type);
+      LOG(FATAL) << "Unexpected property type: "
+                 << static_cast<int>(type.type_enum);
       return "";
     }
   }
@@ -262,6 +321,11 @@ struct Any {
         return value.b == other.value.b;
       } else if (type == PropertyType::kFloat) {
         return value.f == other.value.f;
+      } else if (type.type_enum == impl::PropertyTypeImpl::kVarChar) {
+        if (other.type.type_enum != impl::PropertyTypeImpl::kVarChar) {
+          return false;
+        }
+        return value.s == other.value.s;
       } else {
         return false;
       }
@@ -296,8 +360,9 @@ struct Any {
         return false;
       }
     } else {
-      LOG(FATAL) << "Type [" << static_cast<int>(type) << "] and ["
-                 << static_cast<int>(other.type) << "] cannot be compared..";
+      LOG(FATAL) << "Type [" << static_cast<int>(type.type_enum) << "] and ["
+                 << static_cast<int>(other.type.type_enum)
+                 << "] cannot be compared..";
     }
   }
 
@@ -405,7 +470,7 @@ struct AnyConverter {};
 // specialization for bool
 template <>
 struct AnyConverter<bool> {
-  static constexpr PropertyType type = PropertyType::kBool;
+  static PropertyType type() { return PropertyType::kBool; }
 
   static Any to_any(const bool& value) {
     Any ret;
@@ -429,7 +494,7 @@ struct AnyConverter<bool> {
 
 template <>
 struct AnyConverter<uint8_t> {
-  static constexpr PropertyType type = PropertyType::kUInt8;
+  static PropertyType type() { return PropertyType::kUInt8; }
   static Any to_any(const uint8_t& value) {
     Any ret;
     ret.set_u8(value);
@@ -443,7 +508,7 @@ struct AnyConverter<uint8_t> {
 
 template <>
 struct AnyConverter<uint16_t> {
-  static constexpr PropertyType type = PropertyType::kUInt16;
+  static PropertyType type() { return PropertyType::kUInt16; }
   static Any to_any(const uint16_t& value) {
     Any ret;
     ret.set_u16(value);
@@ -457,7 +522,7 @@ struct AnyConverter<uint16_t> {
 
 template <>
 struct AnyConverter<int32_t> {
-  static constexpr PropertyType type = PropertyType::kInt32;
+  static PropertyType type() { return PropertyType::kInt32; }
 
   static Any to_any(const int32_t& value) {
     Any ret;
@@ -483,7 +548,7 @@ struct AnyConverter<int32_t> {
 
 template <>
 struct AnyConverter<uint32_t> {
-  static constexpr PropertyType type = PropertyType::kUInt32;
+  static PropertyType type() { return PropertyType::kUInt32; }
 
   static Any to_any(const uint32_t& value) {
     Any ret;
@@ -508,7 +573,7 @@ struct AnyConverter<uint32_t> {
 };
 template <>
 struct AnyConverter<int64_t> {
-  static constexpr PropertyType type = PropertyType::kInt64;
+  static PropertyType type() { return PropertyType::kInt64; }
 
   static Any to_any(const int64_t& value) {
     Any ret;
@@ -534,7 +599,7 @@ struct AnyConverter<int64_t> {
 
 template <>
 struct AnyConverter<uint64_t> {
-  static constexpr PropertyType type = PropertyType::kUInt64;
+  static PropertyType type() { return PropertyType::kUInt64; }
 
   static Any to_any(const uint64_t& value) {
     Any ret;
@@ -560,7 +625,7 @@ struct AnyConverter<uint64_t> {
 
 template <>
 struct AnyConverter<Date> {
-  static constexpr PropertyType type = PropertyType::kDate;
+  static PropertyType type() { return PropertyType::kDate; }
 
   static Any to_any(const Date& value) {
     Any ret;
@@ -590,7 +655,7 @@ struct AnyConverter<Date> {
 
 template <>
 struct AnyConverter<std::string_view> {
-  static constexpr PropertyType type = PropertyType::kString;
+  static PropertyType type() { return PropertyType::kString; }
 
   static Any to_any(const std::string_view& value) {
     Any ret;
@@ -616,7 +681,7 @@ struct AnyConverter<std::string_view> {
 
 template <>
 struct AnyConverter<std::string> {
-  static constexpr PropertyType type = PropertyType::kString;
+  static PropertyType type() { return PropertyType::kString; }
 
   static Any to_any(const std::string& value) {
     Any ret;
@@ -642,7 +707,7 @@ struct AnyConverter<std::string> {
 
 template <>
 struct AnyConverter<grape::EmptyType> {
-  static constexpr PropertyType type = PropertyType::kEmpty;
+  static PropertyType type() { return PropertyType::kEmpty; }
 
   static Any to_any(const grape::EmptyType& value) {
     Any ret;
@@ -666,7 +731,7 @@ struct AnyConverter<grape::EmptyType> {
 
 template <>
 struct AnyConverter<double> {
-  static constexpr PropertyType type = PropertyType::kDouble;
+  static PropertyType type() { return PropertyType::kDouble; }
 
   static Any to_any(const double& value) {
     Any ret;
@@ -693,7 +758,7 @@ struct AnyConverter<double> {
 // specilization for float
 template <>
 struct AnyConverter<float> {
-  static constexpr PropertyType type = PropertyType::kFloat;
+  static PropertyType type() { return PropertyType::kFloat; }
 
   static Any to_any(const float& value) {
     Any ret;
@@ -715,6 +780,11 @@ struct AnyConverter<float> {
   static const float& from_any_value(const AnyValue& value) { return value.f; }
 };
 
+grape::InArchive& operator<<(grape::InArchive& in_archive,
+                             const PropertyType& value);
+grape::OutArchive& operator>>(grape::OutArchive& out_archive,
+                              PropertyType& value);
+
 grape::InArchive& operator<<(grape::InArchive& in_archive, const Any& value);
 grape::OutArchive& operator>>(grape::OutArchive& out_archive, Any& value);
 
@@ -733,40 +803,36 @@ inline ostream& operator<<(ostream& os, const gs::Date& dt) {
 }
 
 inline ostream& operator<<(ostream& os, gs::PropertyType pt) {
-  switch (pt) {
-  case gs::PropertyType::kBool:
+  if (pt == gs::PropertyType::Bool()) {
     os << "bool";
-    break;
-  case gs::PropertyType::kInt32:
+  } else if (pt == gs::PropertyType::Empty()) {
+    os << "empty";
+  } else if (pt == gs::PropertyType::UInt8()) {
+    os << "uint8";
+  } else if (pt == gs::PropertyType::UInt16()) {
+    os << "uint16";
+  } else if (pt == gs::PropertyType::Int32()) {
     os << "int32";
-    break;
-  case gs::PropertyType::kUInt32:
+  } else if (pt == gs::PropertyType::UInt32()) {
     os << "uint32";
-    break;
-  case gs::PropertyType::kInt64:
+  } else if (pt == gs::PropertyType::Float()) {
+    os << "float";
+  } else if (pt == gs::PropertyType::Int64()) {
     os << "int64";
-    break;
-  case gs::PropertyType::kUInt64:
+  } else if (pt == gs::PropertyType::UInt64()) {
     os << "uint64";
-    break;
-  case gs::PropertyType::kDate:
-    os << "Date";
-    break;
-  case gs::PropertyType::kString:
-    os << "String";
-    break;
-  case gs::PropertyType::kEmpty:
-    os << "Empty";
-    break;
-  case gs::PropertyType::kDouble:
+  } else if (pt == gs::PropertyType::Double()) {
     os << "double";
-    break;
-  case gs::PropertyType::kFloat:
-    os << "float";
-    break;
-  default:
-    os << "Unknown";
-    break;
+  } else if (pt == gs::PropertyType::Date()) {
+    os << "date";
+  } else if (pt == gs::PropertyType::String()) {
+    os << "string";
+  } else if (pt == gs::PropertyType::StringMap()) {
+    os << "string_map";
+  } else if (pt.type_enum == gs::impl::PropertyTypeImpl::kVarChar) {
+    os << "varchar(" << pt.additional_type_info.max_length << ")";
+  } else {
+    os << "unknown";
   }
   return os;
 }