Repository: alibaba/GraphScope (language: python)

Issue:
feat(interactive): Support string property on edge
Support using string type on edge property.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/flex.yml b/.github/workflows/flex.yml
index 9e9d40192b39..c11fbe350d2c 100644
--- a/.github/workflows/flex.yml
+++ b/.github/workflows/flex.yml
@@ -68,6 +68,17 @@ jobs:
       run: |
         git clone -b master --single-branch --depth=1 https://github.com/GraphScope/gstest.git ${GS_TEST_DIR}
 
+    - name: Test String edge property on modern graph
+      env:
+        FLEX_DATA_DIR: ${{ github.workspace }}/flex/interactive/examples/modern_graph/
+      run: |
+        rm -rf /tmp/csr-data-dir/
+        cd ${GITHUB_WORKSPACE}/flex/build/
+        SCHEMA_FILE=../tests/rt_mutable_graph/modern_graph_string_edge.yaml 
+        BULK_LOAD_FILE=../interactive/examples/modern_graph/bulk_load.yaml
+        GLOG_v=10 ./bin/bulk_loader  -g ${SCHEMA_FILE} -l ${BULK_LOAD_FILE} -d /tmp/csr-data-dir/
+        GLOG_v=10 ./tests/rt_mutable_graph/string_edge_property_test ${SCHEMA_FILE} /tmp/csr-data-dir/
+
     - name: Test Graph Loading on modern graph
       env:
         FLEX_DATA_DIR: ${{ github.workspace }}/flex/interactive/examples/modern_graph/
diff --git a/flex/engines/graph_db/database/graph_db_session.cc b/flex/engines/graph_db/database/graph_db_session.cc
index 1cf657bd7539..e90be0308188 100644
--- a/flex/engines/graph_db/database/graph_db_session.cc
+++ b/flex/engines/graph_db/database/graph_db_session.cc
@@ -157,7 +157,9 @@ Result<std::vector<char>> GraphDBSession::Eval(const std::string& input) {
 
     LOG(INFO) << "[Query-" << (int) type << "][Thread-" << thread_id_
               << "] retry - " << i << " / " << MAX_RETRY;
-    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    if (i + 1 < MAX_RETRY) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
 
     decoder.reset(str_data, str_len);
     result_buffer.clear();
diff --git a/flex/engines/graph_db/database/graph_db_session.h b/flex/engines/graph_db/database/graph_db_session.h
index 9aefd4cc8029..cd875a1b8de5 100644
--- a/flex/engines/graph_db/database/graph_db_session.h
+++ b/flex/engines/graph_db/database/graph_db_session.h
@@ -37,7 +37,7 @@ void put_argment(gs::Encoder& encoder, const query::Argument& argment);
 
 class GraphDBSession {
  public:
-  static constexpr int32_t MAX_RETRY = 4;
+  static constexpr int32_t MAX_RETRY = 3;
   GraphDBSession(GraphDB& db, Allocator& alloc, WalWriter& logger,
                  const std::string& work_dir, int thread_id)
       : db_(db),
diff --git a/flex/engines/graph_db/database/read_transaction.h b/flex/engines/graph_db/database/read_transaction.h
index c6f1d9ed4606..0ae10dbc4c8c 100644
--- a/flex/engines/graph_db/database/read_transaction.h
+++ b/flex/engines/graph_db/database/read_transaction.h
@@ -27,27 +27,28 @@ namespace gs {
 
 class MutablePropertyFragment;
 class VersionManager;
-
 template <typename EDATA_T>
 class AdjListView {
   class nbr_iterator {
-    using nbr_t = MutableNbr<EDATA_T>;
+    using const_nbr_t = typename MutableNbrSlice<EDATA_T>::const_nbr_t;
+    using const_nbr_ptr_t = typename MutableNbrSlice<EDATA_T>::const_nbr_ptr_t;
 
    public:
-    nbr_iterator(const nbr_t* ptr, const nbr_t* end, timestamp_t timestamp)
+    nbr_iterator(const_nbr_ptr_t ptr, const_nbr_ptr_t end,
+                 timestamp_t timestamp)
         : ptr_(ptr), end_(end), timestamp_(timestamp) {
-      while (ptr_ != end_ && ptr_->timestamp > timestamp_) {
+      while (ptr_ != end_ && ptr_->get_timestamp() > timestamp_) {
         ++ptr_;
       }
     }
 
-    const nbr_t& operator*() const { return *ptr_; }
+    const_nbr_t& operator*() const { return *ptr_; }
 
-    const nbr_t* operator->() const { return ptr_; }
+    const_nbr_ptr_t operator->() const { return ptr_; }
 
     nbr_iterator& operator++() {
       ++ptr_;
-      while (ptr_ != end_ && ptr_->timestamp > timestamp_) {
+      while (ptr_ != end_ && ptr_->get_timestamp() > timestamp_) {
         ++ptr_;
       }
       return *this;
@@ -62,8 +63,8 @@ class AdjListView {
     }
 
    private:
-    const nbr_t* ptr_;
-    const nbr_t* end_;
+    const_nbr_ptr_t ptr_;
+    const_nbr_ptr_t end_;
     timestamp_t timestamp_;
   };
 
diff --git a/flex/engines/graph_db/database/update_transaction.cc b/flex/engines/graph_db/database/update_transaction.cc
index 89c2c18b2cb4..5e158494ed4a 100644
--- a/flex/engines/graph_db/database/update_transaction.cc
+++ b/flex/engines/graph_db/database/update_transaction.cc
@@ -708,35 +708,8 @@ void UpdateTransaction::batch_commit(UpdateBatch& batch) {
     bool dst_flag = graph_.get_lid(dst_label, dst, dst_lid);
 
     if (src_flag && dst_flag) {
-      auto oe = graph_.get_outgoing_edges_mut(src_label, src_lid, dst_label,
-                                              edge_label);
-      while (oe != nullptr && oe->is_valid()) {
-        if (oe->get_neighbor() == dst_lid) {
-          oe->set_data(prop, timestamp_);
-          src_flag = false;
-          break;
-        }
-        oe->next();
-      }
-      auto ie = graph_.get_incoming_edges_mut(dst_label, dst_lid, src_label,
-                                              edge_label);
-      while (ie != nullptr && ie->is_valid()) {
-        if (ie->get_neighbor() == src_lid) {
-          dst_flag = false;
-          ie->set_data(prop, timestamp_);
-          break;
-        }
-        ie->next();
-      }
-      if ((!src_flag) || (!dst_flag)) {
-      } else {
-        grape::InArchive arc;
-        arc << prop;
-
-        grape::OutArchive out_arc(std::move(arc));
-        graph_.IngestEdge(src_label, src_lid, dst_label, dst_lid, edge_label,
-                          timestamp_, out_arc, alloc_);
-      }
+      graph_.UpdateEdge(src_label, src_lid, dst_label, dst_lid, edge_label,
+                        timestamp_, prop, alloc_);
     }
   }
   auto& arc = batch.GetArc();
@@ -822,9 +795,6 @@ void UpdateTransaction::applyEdgesUpdates() {
           }
         }
 
-        MutableCsrBase* csr =
-            graph_.get_oe_csr(src_label, dst_label, edge_label);
-
         for (auto& pair : added_edges_[oe_csr_index]) {
           vid_t v = pair.first;
           auto& add_list = pair.second;
@@ -839,7 +809,11 @@ void UpdateTransaction::applyEdgesUpdates() {
               continue;
             auto u = add_list[idx];
             auto value = edge_data.at(u).first;
-            csr->put_generic_edge(v, u, value, timestamp_, alloc_);
+            grape::InArchive iarc;
+            serialize_field(iarc, value);
+            grape::OutArchive oarc(std::move(iarc));
+            graph_.IngestEdge(src_label, v, dst_label, u, edge_label,
+                              timestamp_, oarc, alloc_);
           }
         }
       }
@@ -875,22 +849,6 @@ void UpdateTransaction::applyEdgesUpdates() {
             }
           }
         }
-
-        MutableCsrBase* csr =
-            graph_.get_ie_csr(dst_label, src_label, edge_label);
-
-        for (auto& pair : added_edges_[ie_csr_index]) {
-          vid_t v = pair.first;
-          auto& add_list = pair.second;
-          if (add_list.empty()) {
-            continue;
-          }
-          auto& edge_data = updated_edge_data_[ie_csr_index].at(v);
-          for (auto u : add_list) {
-            auto value = edge_data.at(u).first;
-            csr->put_generic_edge(v, u, value, timestamp_, alloc_);
-          }
-        }
       }
     }
   }
diff --git a/flex/storages/rt_mutable_graph/dual_csr.h b/flex/storages/rt_mutable_graph/dual_csr.h
new file mode 100644
index 000000000000..d90f509b4390
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/dual_csr.h
@@ -0,0 +1,303 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef GRAPHSCOPE_FRAGMENT_DUAL_CSR_H_
+#define GRAPHSCOPE_FRAGMENT_DUAL_CSR_H_
+
+#include <stdio.h>
+
+#include <grape/serialization/in_archive.h>
+#include "flex/storages/rt_mutable_graph/mutable_csr.h"
+#include "flex/utils/allocators.h"
+
+namespace gs {
+
+class DualCsrBase {
+ public:
+  DualCsrBase() = default;
+  virtual ~DualCsrBase() = default;
+  virtual void BatchInit(const std::string& oe_name, const std::string& ie_name,
+                         const std::string& edata_name,
+                         const std::string& work_dir,
+                         const std::vector<int>& oe_degree,
+                         const std::vector<int>& ie_degree) = 0;
+  virtual void Open(const std::string& oe_name, const std::string& ie_name,
+                    const std::string& edata_name,
+                    const std::string& snapshot_dir,
+                    const std::string& work_dir) = 0;
+  virtual void Dump(const std::string& oe_name, const std::string& ie_name,
+                    const std::string& edata_name,
+                    const std::string& new_snapshot_dir) = 0;
+
+  virtual void PutEdge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                       Allocator& alloc) = 0;
+
+  virtual void IngestEdge(vid_t src, vid_t dst, grape::OutArchive& oarc,
+                          timestamp_t timestamp, Allocator& alloc) = 0;
+
+  virtual void UpdateEdge(vid_t src, vid_t dst, const Any& oarc,
+                          timestamp_t timestamp, Allocator& alloc) = 0;
+
+  virtual MutableCsrBase* GetInCsr() = 0;
+  virtual MutableCsrBase* GetOutCsr() = 0;
+};
+
+template <typename EDATA_T>
+class DualCsr : public DualCsrBase {
+ public:
+  DualCsr(EdgeStrategy oe_strategy, EdgeStrategy ie_strategy)
+      : in_csr_(nullptr), out_csr_(nullptr) {
+    if (ie_strategy == EdgeStrategy::kNone) {
+      in_csr_ = new EmptyCsr<EDATA_T>();
+    } else if (ie_strategy == EdgeStrategy::kMultiple) {
+      in_csr_ = new MutableCsr<EDATA_T>();
+    } else if (ie_strategy == EdgeStrategy::kSingle) {
+      in_csr_ = new SingleMutableCsr<EDATA_T>();
+    }
+    if (oe_strategy == EdgeStrategy::kNone) {
+      out_csr_ = new EmptyCsr<EDATA_T>();
+    } else if (oe_strategy == EdgeStrategy::kMultiple) {
+      out_csr_ = new MutableCsr<EDATA_T>();
+    } else if (oe_strategy == EdgeStrategy::kSingle) {
+      out_csr_ = new SingleMutableCsr<EDATA_T>();
+    }
+  }
+  ~DualCsr() {
+    if (in_csr_ != nullptr) {
+      delete in_csr_;
+    }
+    if (out_csr_ != nullptr) {
+      delete out_csr_;
+    }
+  }
+  void BatchInit(const std::string& oe_name, const std::string& ie_name,
+                 const std::string& edata_name, const std::string& work_dir,
+                 const std::vector<int>& oe_degree,
+                 const std::vector<int>& ie_degree) override {
+    in_csr_->batch_init(ie_name, work_dir, ie_degree);
+    out_csr_->batch_init(oe_name, work_dir, oe_degree);
+  }
+
+  void Open(const std::string& oe_name, const std::string& ie_name,
+            const std::string& edata_name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {
+    in_csr_->open(ie_name, snapshot_dir, work_dir);
+    out_csr_->open(oe_name, snapshot_dir, work_dir);
+  }
+
+  void Dump(const std::string& oe_name, const std::string& ie_name,
+            const std::string& edata_name,
+            const std::string& new_snapshot_dir) override {
+    in_csr_->dump(ie_name, new_snapshot_dir);
+    out_csr_->dump(oe_name, new_snapshot_dir);
+  }
+
+  MutableCsrBase* GetInCsr() override { return in_csr_; }
+  MutableCsrBase* GetOutCsr() override { return out_csr_; }
+  void PutEdge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+               Allocator& alloc) override {
+    EDATA_T prop;
+    ConvertAny<EDATA_T>::to(data, prop);
+    in_csr_->put_edge(dst, src, prop, ts, alloc);
+    out_csr_->put_edge(src, dst, prop, ts, alloc);
+  }
+
+  void UpdateEdge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                  Allocator& alloc) override {
+    auto oe = out_csr_->edge_iter_mut(src);
+    EDATA_T prop;
+    ConvertAny<EDATA_T>::to(data, prop);
+    bool src_flag = false, dst_flag = false;
+    while (oe != nullptr && oe->is_valid()) {
+      if (oe->get_neighbor() == dst) {
+        oe->set_data(prop, ts);
+        src_flag = true;
+        break;
+      }
+      oe->next();
+    }
+    auto ie = in_csr_->edge_iter_mut(dst);
+    while (ie != nullptr && ie->is_valid()) {
+      if (ie->get_neighbor() == src) {
+        dst_flag = true;
+        ie->set_data(prop, ts);
+        break;
+      }
+      ie->next();
+    }
+    if (!(src_flag || dst_flag)) {
+      in_csr_->put_edge(dst, src, prop, ts, alloc);
+      out_csr_->put_edge(src, dst, prop, ts, alloc);
+    }
+  }
+
+  void IngestEdge(vid_t src, vid_t dst, grape::OutArchive& oarc, timestamp_t ts,
+                  Allocator& alloc) override {
+    EDATA_T data;
+    oarc >> data;
+    in_csr_->put_edge(dst, src, data, ts, alloc);
+    out_csr_->put_edge(src, dst, data, ts, alloc);
+  }
+
+  void BatchPutEdge(vid_t src, vid_t dst, const EDATA_T& data) {
+    in_csr_->batch_put_edge(dst, src, data);
+    out_csr_->batch_put_edge(src, dst, data);
+  }
+
+ private:
+  TypedMutableCsrBase<EDATA_T>* in_csr_;
+  TypedMutableCsrBase<EDATA_T>* out_csr_;
+};
+
+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) {
+    if (ie_strategy == EdgeStrategy::kNone) {
+      in_csr_ = new EmptyCsr<std::string_view>(column_, column_idx_);
+    } else if (ie_strategy == EdgeStrategy::kMultiple) {
+      in_csr_ = new MutableCsr<std::string_view>(column_, column_idx_);
+    } else if (ie_strategy == EdgeStrategy::kSingle) {
+      in_csr_ = new SingleMutableCsr<std::string_view>(column_, column_idx_);
+    }
+    if (oe_strategy == EdgeStrategy::kNone) {
+      out_csr_ = new EmptyCsr<std::string_view>(column_, column_idx_);
+    } else if (oe_strategy == EdgeStrategy::kMultiple) {
+      out_csr_ = new MutableCsr<std::string_view>(column_, column_idx_);
+    } else if (oe_strategy == EdgeStrategy::kSingle) {
+      out_csr_ = new SingleMutableCsr<std::string_view>(column_, column_idx_);
+    }
+  }
+  ~DualCsr() {
+    if (in_csr_ != nullptr) {
+      delete in_csr_;
+    }
+    if (out_csr_ != nullptr) {
+      delete out_csr_;
+    }
+  }
+  void BatchInit(const std::string& oe_name, const std::string& ie_name,
+                 const std::string& edata_name, const std::string& work_dir,
+                 const std::vector<int>& oe_degree,
+                 const std::vector<int>& ie_degree) override {
+    size_t ie_num = in_csr_->batch_init(ie_name, work_dir, ie_degree);
+    size_t oe_num = out_csr_->batch_init(oe_name, work_dir, oe_degree);
+    column_.open(edata_name, "", work_dir);
+    column_.resize(std::max(ie_num, oe_num));
+    column_idx_.store(0);
+  }
+
+  void Open(const std::string& oe_name, const std::string& ie_name,
+            const std::string& edata_name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {
+    in_csr_->open(ie_name, snapshot_dir, work_dir);
+    out_csr_->open(oe_name, snapshot_dir, work_dir);
+    column_.open(edata_name, snapshot_dir, work_dir);
+    column_idx_.store(column_.size());
+    column_.resize(column_.size() + (column_.size() + 4) / 5);
+  }
+
+  void Dump(const std::string& oe_name, const std::string& ie_name,
+            const std::string& edata_name,
+            const std::string& new_snapshot_dir) override {
+    in_csr_->dump(ie_name, new_snapshot_dir);
+    out_csr_->dump(oe_name, new_snapshot_dir);
+    column_.resize(column_idx_.load());
+    column_.dump(new_snapshot_dir + "/" + edata_name);
+  }
+
+  MutableCsrBase* GetInCsr() override { return in_csr_; }
+  MutableCsrBase* GetOutCsr() override { return out_csr_; }
+  void PutEdge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+               Allocator& alloc) override {
+    std::string_view val = data.AsStringView();
+    size_t row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, val);
+    in_csr_->put_edge_with_index(dst, src, row_id, ts, alloc);
+    out_csr_->put_edge_with_index(src, dst, row_id, ts, alloc);
+  }
+
+  void UpdateEdge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                  Allocator& alloc) override {
+    auto oe_ptr = out_csr_->edge_iter_mut(src);
+    std::string_view prop = data.AsStringView();
+    auto oe =
+        dynamic_cast<TypedMutableCsrEdgeIter<std::string_view>*>(oe_ptr.get());
+    size_t index = std::numeric_limits<size_t>::max();
+    while (oe != nullptr && oe->is_valid()) {
+      if (oe->get_neighbor() == dst) {
+        oe->set_timestamp(ts);
+        index = oe->get_index();
+        break;
+      }
+      oe->next();
+    }
+    auto ie_ptr = in_csr_->edge_iter_mut(dst);
+    auto ie =
+        dynamic_cast<TypedMutableCsrEdgeIter<std::string_view>*>(ie_ptr.get());
+    while (ie != nullptr && ie->is_valid()) {
+      if (ie->get_neighbor() == src) {
+        ie->set_timestamp(ts);
+        index = ie->get_index();
+        break;
+      }
+      ie->next();
+    }
+    if (index != std::numeric_limits<size_t>::max()) {
+      column_.set_value(index, prop);
+    } else {
+      size_t row_id = column_idx_.fetch_add(1);
+      column_.set_value(row_id, prop);
+      in_csr_->put_edge_with_index(dst, src, row_id, ts, alloc);
+      out_csr_->put_edge_with_index(src, dst, row_id, ts, alloc);
+    }
+  }
+
+  void IngestEdge(vid_t src, vid_t dst, grape::OutArchive& oarc, timestamp_t ts,
+                  Allocator& alloc) override {
+    std::string_view prop;
+    oarc >> prop;
+    size_t row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, prop);
+    in_csr_->put_edge_with_index(dst, src, row_id, ts, alloc);
+    out_csr_->put_edge_with_index(src, dst, row_id, ts, alloc);
+  }
+
+  void BatchPutEdge(vid_t src, vid_t dst, const std::string_view& data) {
+    size_t row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, data);
+    in_csr_->batch_put_edge_with_index(dst, src, row_id);
+    out_csr_->batch_put_edge_with_index(src, dst, row_id);
+  }
+
+  void BatchPutEdge(vid_t src, vid_t dst, const std::string& data) {
+    size_t row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, data);
+
+    in_csr_->batch_put_edge_with_index(dst, src, row_id);
+    out_csr_->batch_put_edge_with_index(src, dst, row_id);
+  }
+
+ private:
+  TypedMutableCsrBase<std::string_view>* in_csr_;
+  TypedMutableCsrBase<std::string_view>* out_csr_;
+  std::atomic<size_t> column_idx_;
+  StringColumn column_;
+};
+
+}  // namespace gs
+
+#endif  // GRAPHSCOPE_FRAGMENT_DUAL_CSR_H_
diff --git a/flex/storages/rt_mutable_graph/file_names.h b/flex/storages/rt_mutable_graph/file_names.h
index 6a836edff76c..a5b855b7fe4d 100644
--- a/flex/storages/rt_mutable_graph/file_names.h
+++ b/flex/storages/rt_mutable_graph/file_names.h
@@ -192,6 +192,11 @@ inline std::string oe_prefix(const std::string& src_label,
   return "oe_" + src_label + "_" + edge_label + "_" + dst_label;
 }
 
+inline std::string edata_prefix(const std::string& src_label,
+                                const std::string& dst_label,
+                                const std::string& edge_label) {
+  return "e_" + src_label + "_" + edge_label + "_" + dst_label + "_data";
+}
 inline std::string vertex_table_prefix(const std::string& label) {
   return "vertex_table_" + label;
 }
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 77bd87c0247e..5450376250e5 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
@@ -322,10 +322,11 @@ void AbstractArrowFragmentLoader::AddEdgesRecordBatch(
     }
   } else if (property_types[0] == PropertyType::kString) {
     if (filenames.empty()) {
-      basic_fragment_loader_.AddNoPropEdgeBatch<std::string>(
+      basic_fragment_loader_.AddNoPropEdgeBatch<std::string_view>(
           src_label_i, dst_label_i, edge_label_i);
     } else {
-      LOG(FATAL) << "Unsupported edge property type.";
+      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()) {
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 9280ec9c5fe1..22f1b95bef9a 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
@@ -204,7 +204,9 @@ static void append_edges(
                                    arrow::StringArray>::value ||
                       std::is_same<arrow_array_type,
                                    arrow::LargeStringArray>::value) {
-          std::get<2>(parsed_edges[cur_ind++]) = data->GetView(j);
+          auto str = data->GetView(j);
+          std::string_view str_view(str.data(), str.size());
+          std::get<2>(parsed_edges[cur_ind++]) = str_view;
         } else {
           std::get<2>(parsed_edges[cur_ind++]) = data->Value(j);
         }
diff --git a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc
index d9c15fbedb05..d6535ac21553 100644
--- a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc
+++ b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc
@@ -28,6 +28,8 @@ BasicFragmentLoader::BasicFragmentLoader(const Schema& schema,
   vertex_data_.resize(vertex_label_num_);
   ie_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_, NULL);
   oe_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_, NULL);
+  dual_csr_list_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_,
+                        NULL);
   lf_indexers_.resize(vertex_label_num_);
   std::filesystem::create_directories(runtime_dir(prefix));
   std::filesystem::create_directories(snapshot_dir(prefix, 0));
@@ -76,14 +78,11 @@ void BasicFragmentLoader::LoadFragment() {
         size_t index = src_label * vertex_label_num_ * edge_label_num_ +
                        dst_label * edge_label_num_ + edge_label;
         if (schema_.exist(src_label_name, dst_label_name, edge_label_name)) {
-          if (ie_[index] != NULL) {
-            ie_[index]->dump(
-                ie_prefix(src_label_name, dst_label_name, edge_label_name),
-                snapshot_dir(work_dir_, 0));
-          }
-          if (oe_[index] != NULL) {
-            oe_[index]->dump(
+          if (dual_csr_list_[index] != NULL) {
+            dual_csr_list_[index]->Dump(
                 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),
                 snapshot_dir(work_dir_, 0));
           }
         }
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 b3896de3a17f..7765dab52b67 100644
--- a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
+++ b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
@@ -80,14 +80,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);
-    ie_[index] = create_typed_csr<EDATA_T>(ie_strategy);
-    oe_[index] = create_typed_csr<EDATA_T>(oe_strategy);
-    ie_[index]->batch_init(
-        ie_prefix(src_label_name, dst_label_name, edge_label_name),
-        tmp_dir(work_dir_), {});
-    oe_[index]->batch_init(
+    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(
         oe_prefix(src_label_name, dst_label_name, edge_label_name),
-        tmp_dir(work_dir_), {});
+        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_), {}, {});
   }
 
   template <typename EDATA_T>
@@ -109,26 +109,22 @@ 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);
-    auto ie_csr = create_typed_csr<EDATA_T>(ie_strategy);
-    auto oe_csr = create_typed_csr<EDATA_T>(oe_strategy);
+
+    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());
-
-    ie_csr->batch_init(
-        ie_prefix(src_label_name, dst_label_name, edge_label_name),
-        tmp_dir(work_dir_), ie_degree);
-    oe_csr->batch_init(
+    dual_csr->BatchInit(
         oe_prefix(src_label_name, dst_label_name, edge_label_name),
-        tmp_dir(work_dir_), oe_degree);
-
+        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) {
-      ie_csr->batch_put_edge(std::get<1>(edge), std::get<0>(edge),
-                             std::get<2>(edge));
-      oe_csr->batch_put_edge(std::get<0>(edge), std::get<1>(edge),
+      dual_csr->BatchPutEdge(std::get<0>(edge), std::get<1>(edge),
                              std::get<2>(edge));
     }
-    ie_[index] = ie_csr;
-    oe_[index] = oe_csr;
     VLOG(10) << "Finish adding edge batch of size: " << edges.size();
   }
 
@@ -147,6 +143,7 @@ class BasicFragmentLoader {
   size_t vertex_label_num_, edge_label_num_;
   std::vector<LFIndexer<vid_t>> lf_indexers_;
   std::vector<MutableCsrBase*> ie_, oe_;
+  std::vector<DualCsrBase*> dual_csr_list_;
   std::vector<Table> vertex_data_;
 };
 }  // namespace gs
diff --git a/flex/storages/rt_mutable_graph/mutable_csr.cc b/flex/storages/rt_mutable_graph/mutable_csr.cc
index 32a52e869968..3467741dd421 100644
--- a/flex/storages/rt_mutable_graph/mutable_csr.cc
+++ b/flex/storages/rt_mutable_graph/mutable_csr.cc
@@ -35,9 +35,6 @@ template class MutableCsr<uint32_t>;
 template class SingleMutableCsr<Date>;
 template class MutableCsr<Date>;
 
-template class SingleMutableCsr<std::string>;
-template class MutableCsr<std::string>;
-
 template class SingleMutableCsr<int64_t>;
 template class MutableCsr<int64_t>;
 
diff --git a/flex/storages/rt_mutable_graph/mutable_csr.h b/flex/storages/rt_mutable_graph/mutable_csr.h
index e2a1a8693aa8..cf23adf7ef6c 100644
--- a/flex/storages/rt_mutable_graph/mutable_csr.h
+++ b/flex/storages/rt_mutable_graph/mutable_csr.h
@@ -25,6 +25,7 @@
 #include "flex/storages/rt_mutable_graph/types.h"
 #include "flex/utils/allocators.h"
 #include "flex/utils/mmap_array.h"
+#include "flex/utils/property/column.h"
 #include "flex/utils/property/types.h"
 #include "grape/serialization/in_archive.h"
 #include "grape/serialization/out_archive.h"
@@ -41,6 +42,17 @@ struct MutableNbr {
         data(rhs.data) {}
   ~MutableNbr() = default;
 
+  const EDATA_T& get_data() const { return data; }
+  vid_t get_neighbor() const { return neighbor; }
+  timestamp_t get_timestamp() const { return timestamp.load(); }
+
+  void set_data(const EDATA_T& val, timestamp_t ts) {
+    data = val;
+    timestamp.store(ts);
+  }
+  void set_neighbor(vid_t neighbor) { neighbor = neighbor; }
+  void set_timestamp(timestamp_t ts) { timestamp.store(ts); }
+
   vid_t neighbor;
   std::atomic<timestamp_t> timestamp;
   EDATA_T data;
@@ -52,7 +64,14 @@ struct MutableNbr<grape::EmptyType> {
   MutableNbr(const MutableNbr& rhs)
       : neighbor(rhs.neighbor), timestamp(rhs.timestamp.load()) {}
   ~MutableNbr() = default;
-
+  void set_data(const grape::EmptyType&, timestamp_t ts) {
+    timestamp.store(ts);
+  }
+  void set_neighbor(vid_t neighbor) { neighbor = neighbor; }
+  void set_timestamp(timestamp_t ts) { timestamp.store(ts); }
+  const grape::EmptyType& get_data() const { return data; }
+  vid_t get_neighbor() const { return neighbor; }
+  timestamp_t get_timestamp() const { return timestamp.load(); }
   vid_t neighbor;
   union {
     std::atomic<timestamp_t> timestamp;
@@ -63,17 +82,18 @@ struct MutableNbr<grape::EmptyType> {
 template <typename EDATA_T>
 class MutableNbrSlice {
  public:
-  using nbr_t = MutableNbr<EDATA_T>;
+  using const_nbr_t = MutableNbr<EDATA_T>;
+  using const_nbr_ptr_t = const MutableNbr<EDATA_T>*;
   MutableNbrSlice() = default;
   ~MutableNbrSlice() = default;
 
   void set_size(int size) { size_ = size; }
   int size() const { return size_; }
 
-  void set_begin(const nbr_t* ptr) { ptr_ = ptr; }
+  void set_begin(const_nbr_ptr_t ptr) { ptr_ = ptr; }
 
-  const nbr_t* begin() const { return ptr_; }
-  const nbr_t* end() const { return ptr_ + size_; }
+  const_nbr_ptr_t begin() const { return ptr_; }
+  const_nbr_ptr_t end() const { return ptr_ + size_; }
 
   static MutableNbrSlice empty() {
     MutableNbrSlice ret;
@@ -83,14 +103,89 @@ class MutableNbrSlice {
   }
 
  private:
-  const nbr_t* ptr_;
+  const_nbr_ptr_t ptr_;
   int size_;
 };
 
+template <>
+class MutableNbrSlice<std::string_view> {
+ public:
+  struct MutableColumnNbr {
+    using const_nbr_t = MutableNbr<size_t>;
+    using const_nbr_ptr_t = const MutableNbr<size_t>*;
+
+    MutableColumnNbr(const_nbr_ptr_t ptr, const StringColumn& column)
+        : ptr_(ptr), column_(column) {}
+    vid_t get_neighbor() const { return ptr_->neighbor; }
+    std::string_view get_data() const { return column_.get_view(ptr_->data); }
+    timestamp_t get_timestamp() const { return ptr_->timestamp.load(); }
+
+    const MutableColumnNbr& operator*() const { return *this; }
+    const MutableColumnNbr* operator->() const { return this; }
+    const MutableColumnNbr& operator=(const MutableColumnNbr& nbr) const {
+      ptr_ = nbr.ptr_;
+      return *this;
+    }
+    bool operator==(const MutableColumnNbr& nbr) const {
+      return ptr_ == nbr.ptr_;
+    }
+    bool operator!=(const MutableColumnNbr& nbr) const {
+      return ptr_ != nbr.ptr_;
+    }
+    const MutableColumnNbr& operator++() const {
+      ++ptr_;
+      return *this;
+    }
+
+    const MutableColumnNbr& operator+=(size_t n) const {
+      ptr_ += n;
+      return *this;
+    }
+
+    size_t operator-(const MutableColumnNbr& nbr) const {
+      return ptr_ - nbr.ptr_;
+    }
+
+    bool operator<(const MutableColumnNbr& nbr) const {
+      return ptr_ < nbr.ptr_;
+    }
+
+    mutable const_nbr_ptr_t ptr_;
+    const StringColumn& column_;
+  };
+  using const_nbr_t = const MutableColumnNbr;
+  using const_nbr_ptr_t = const MutableColumnNbr;
+  MutableNbrSlice(const StringColumn& column) : column_(column) {}
+  ~MutableNbrSlice() = default;
+  void set_size(int size) { slice_.set_size(size); }
+  int size() const { return slice_.size(); }
+
+  void set_begin(const MutableNbr<size_t>* ptr) { slice_.set_begin(ptr); }
+
+  const MutableColumnNbr begin() const {
+    return MutableColumnNbr(slice_.begin(), column_);
+  }
+  const MutableColumnNbr end() const {
+    return MutableColumnNbr(slice_.end(), column_);
+  }
+
+  static MutableNbrSlice empty(const StringColumn& column) {
+    MutableNbrSlice ret(column);
+    ret.set_begin(nullptr);
+    ret.set_size(0);
+    return ret;
+  }
+
+ private:
+  MutableNbrSlice<size_t> slice_;
+  const StringColumn& column_;
+};
+
 template <typename EDATA_T>
 class MutableNbrSliceMut {
  public:
   using nbr_t = MutableNbr<EDATA_T>;
+  using nbr_ptr_t = MutableNbr<EDATA_T>*;
   MutableNbrSliceMut() = default;
   ~MutableNbrSliceMut() = default;
 
@@ -114,6 +209,80 @@ class MutableNbrSliceMut {
   int size_;
 };
 
+template <>
+class MutableNbrSliceMut<std::string_view> {
+ public:
+  struct MutableColumnNbr {
+    using nbr_t = MutableNbr<size_t>;
+
+    MutableColumnNbr(nbr_t* ptr, StringColumn& column)
+        : ptr_(ptr), column_(column) {}
+    vid_t neighbor() const { return ptr_->neighbor; }
+    std::string_view data() { return column_.get_view(ptr_->data); }
+    vid_t get_neighbor() const { return ptr_->neighbor; }
+    const std::string_view get_data() const {
+      return column_.get_view(ptr_->data);
+    }
+    timestamp_t get_timestamp() const { return ptr_->timestamp.load(); }
+    size_t get_index() const { return ptr_->data; }
+    void set_data(const std::string_view& sw, timestamp_t ts) {
+      column_.set_value(ptr_->data, sw);
+      ptr_->timestamp.store(ts);
+    }
+    void set_neighbor(vid_t neighbor) { ptr_->neighbor = neighbor; }
+
+    void set_timestamp(timestamp_t ts) { ptr_->timestamp.store(ts); }
+
+    const MutableColumnNbr& operator*() const { return *this; }
+    MutableColumnNbr& operator*() { return *this; }
+    MutableColumnNbr& operator=(const MutableColumnNbr& nbr) {
+      ptr_ = nbr.ptr_;
+      return *this;
+    }
+    bool operator==(const MutableColumnNbr& nbr) const {
+      return ptr_ == nbr.ptr_;
+    }
+    bool operator!=(const MutableColumnNbr& nbr) const {
+      return ptr_ != nbr.ptr_;
+    }
+
+    MutableColumnNbr& operator++() {
+      ptr_++;
+      return *this;
+    }
+    MutableColumnNbr& operator+=(size_t n) {
+      ptr_ += n;
+      return *this;
+    }
+
+    bool operator<(const MutableColumnNbr& nbr) { return ptr_ < nbr.ptr_; }
+    nbr_t* ptr_;
+    StringColumn & column_;
+  };
+  using nbr_ptr_t = MutableColumnNbr;
+
+  MutableNbrSliceMut(StringColumn& column) : column_(column) {}
+  ~MutableNbrSliceMut() = default;
+  void set_size(int size) { slice_.set_size(size); }
+  int size() const { return slice_.size(); }
+
+  void set_begin(MutableNbr<size_t>* ptr) { slice_.set_begin(ptr); }
+
+  MutableColumnNbr begin() { return MutableColumnNbr(slice_.begin(), column_); }
+  MutableColumnNbr end() { return MutableColumnNbr(slice_.end(), column_); }
+
+  static MutableNbrSliceMut empty(StringColumn& column) {
+    MutableNbrSliceMut ret(column);
+    ret.set_begin(nullptr);
+    ret.set_size(0);
+    return ret;
+  }
+
+ private:
+  MutableNbrSliceMut<size_t> slice_;
+  StringColumn& column_;
+};
+
 template <typename T>
 struct UninitializedUtils {
   static void copy(T* new_buffer, T* old_buffer, size_t len) {
@@ -187,6 +356,73 @@ class MutableAdjlist {
   int capacity_;
 };
 
+template <>
+class MutableAdjlist<std::string_view> {
+ public:
+  using nbr_t = MutableNbr<size_t>;
+  using slice_t = MutableNbrSlice<std::string_view>;
+  using mut_slice_t = MutableNbrSliceMut<std::string_view>;
+  MutableAdjlist(StringColumn& column)
+      : buffer_(NULL), size_(0), capacity_(0) {}
+  ~MutableAdjlist() {}
+
+  void init(nbr_t* ptr, int cap, int size) {
+    buffer_ = ptr;
+    capacity_ = cap;
+    size_ = size;
+  }
+
+  void batch_put_edge(vid_t neighbor, const size_t& data, timestamp_t ts = 0) {
+    CHECK_LT(size_, capacity_);
+    auto& nbr = buffer_[size_++];
+    nbr.neighbor = neighbor;
+    nbr.data = data;
+    nbr.timestamp.store(ts);
+  }
+
+  void put_edge(vid_t neighbor, const size_t& data, timestamp_t ts,
+                Allocator& allocator) {
+    if (size_ == capacity_) {
+      capacity_ += ((capacity_) >> 1);
+      capacity_ = std::max(capacity_, 8);
+      nbr_t* new_buffer =
+          static_cast<nbr_t*>(allocator.allocate(capacity_ * sizeof(nbr_t)));
+      if (size_ > 0) {
+        UninitializedUtils<nbr_t>::copy(new_buffer, buffer_, size_);
+      }
+      buffer_ = new_buffer;
+    }
+    auto& nbr = buffer_[size_.fetch_add(1)];
+    nbr.neighbor = neighbor;
+    nbr.data = data;
+    nbr.timestamp.store(ts);
+  }
+
+  slice_t get_edges(const StringColumn& column) const {
+    slice_t ret(column);
+    ret.set_size(size_.load(std::memory_order_acquire));
+    ret.set_begin(buffer_);
+    return ret;
+  }
+
+  mut_slice_t get_edges_mut(StringColumn& column) {
+    mut_slice_t ret(column);
+    ret.set_size(size_.load());
+    ret.set_begin(buffer_);
+    return ret;
+  }
+
+  int capacity() const { return capacity_; }
+  int size() const { return size_; }
+  const nbr_t* data() const { return buffer_; }
+  nbr_t* data() { return buffer_; }
+
+ private:
+  nbr_t* buffer_;
+  std::atomic<int> size_;
+  int capacity_;
+};
+
 class MutableCsrConstEdgeIterBase {
  public:
   MutableCsrConstEdgeIterBase() = default;
@@ -222,8 +458,9 @@ class MutableCsrBase {
   MutableCsrBase() {}
   virtual ~MutableCsrBase() {}
 
-  virtual void batch_init(const std::string& name, const std::string& work_dir,
-                          const std::vector<int>& degree) = 0;
+  virtual size_t batch_init(const std::string& name,
+                            const std::string& work_dir,
+                            const std::vector<int>& degree) = 0;
 
   virtual void open(const std::string& name, const std::string& snapshot_dir,
                     const std::string& work_dir) = 0;
@@ -254,25 +491,24 @@ class MutableCsrBase {
 
 template <typename EDATA_T>
 class TypedMutableCsrConstEdgeIter : public MutableCsrConstEdgeIterBase {
-  using nbr_t = MutableNbr<EDATA_T>;
+  using const_nbr_ptr_t = typename MutableNbrSlice<EDATA_T>::const_nbr_ptr_t;
 
  public:
   explicit TypedMutableCsrConstEdgeIter(const MutableNbrSlice<EDATA_T>& slice)
       : cur_(slice.begin()), end_(slice.end()) {}
   ~TypedMutableCsrConstEdgeIter() = default;
 
-  vid_t get_neighbor() const override { return cur_->neighbor; }
+  vid_t get_neighbor() const override { return (*cur_).get_neighbor(); }
   Any get_data() const override {
-    return AnyConverter<EDATA_T>::to_any(cur_->data);
+    return AnyConverter<EDATA_T>::to_any((*cur_).get_data());
   }
-  timestamp_t get_timestamp() const override { return cur_->timestamp.load(); }
+  timestamp_t get_timestamp() const override { return (*cur_).get_timestamp(); }
 
   void next() override { ++cur_; }
   TypedMutableCsrConstEdgeIter& operator+=(size_t offset) override {
-    if (cur_ + offset >= end_) {
+    cur_ += offset;
+    if (!(cur_ < end_)) {
       cur_ = end_;
-    } else {
-      cur_ = cur_ + offset;
     }
     return *this;
   }
@@ -280,8 +516,8 @@ class TypedMutableCsrConstEdgeIter : public MutableCsrConstEdgeIterBase {
   size_t size() const override { return end_ - cur_; }
 
  private:
-  const nbr_t* cur_;
-  const nbr_t* end_;
+  const_nbr_ptr_t cur_;
+  const_nbr_ptr_t end_;
 };
 
 template <typename EDATA_T>
@@ -321,16 +557,66 @@ class TypedMutableCsrEdgeIter : public MutableCsrEdgeIterBase {
   nbr_t* end_;
 };
 
+template <>
+class TypedMutableCsrEdgeIter<std::string_view>
+    : public MutableCsrEdgeIterBase {
+  using nbr_ptr_t = typename MutableNbrSliceMut<std::string_view>::nbr_ptr_t;
+
+ public:
+  explicit TypedMutableCsrEdgeIter(MutableNbrSliceMut<std::string_view> slice)
+      : cur_(slice.begin()), end_(slice.end()) {}
+  ~TypedMutableCsrEdgeIter() = default;
+
+  vid_t get_neighbor() const override { return cur_.get_neighbor(); }
+  Any get_data() const override {
+    return AnyConverter<std::string_view>::to_any(cur_.get_data());
+  }
+  timestamp_t get_timestamp() const override { return cur_.get_timestamp(); }
+
+  void set_data(const Any& value, timestamp_t ts) override {
+    cur_.set_data(value.AsStringView(), ts);
+  }
+  size_t get_index() const { return cur_.get_index(); }
+  void set_timestamp(timestamp_t ts) { cur_.set_timestamp(ts); }
+
+  MutableCsrEdgeIterBase& operator+=(size_t offset) override {
+    cur_ += offset;
+    if (!(cur_ < end_)) {
+      cur_ = end_;
+    }
+    return *this;
+  }
+
+  void next() override { ++cur_; }
+  bool is_valid() const override { return cur_ != end_; }
+
+ private:
+  nbr_ptr_t cur_;
+  nbr_ptr_t end_;
+};
+
 template <typename EDATA_T>
 class TypedMutableCsrBase : public MutableCsrBase {
  public:
   using slice_t = MutableNbrSlice<EDATA_T>;
   virtual void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
                               timestamp_t ts = 0) = 0;
-
+  virtual void put_edge(vid_t src, vid_t dst, const EDATA_T& data,
+                        timestamp_t ts, Allocator& alloc) = 0;
   virtual slice_t get_edges(vid_t i) const = 0;
 };
 
+template <>
+class TypedMutableCsrBase<std::string_view> : public MutableCsrBase {
+ public:
+  using slice_t = MutableNbrSlice<std::string_view>;
+  virtual slice_t get_edges(vid_t i) const = 0;
+  virtual void batch_put_edge_with_index(vid_t src, vid_t dst,
+                                         const size_t& data,
+                                         timestamp_t ts = 0) = 0;
+  virtual void put_edge_with_index(vid_t src, vid_t dst, size_t index,
+                                   timestamp_t ts, Allocator& alloc) = 0;
+};
 template <typename EDATA_T>
 class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
  public:
@@ -346,8 +632,8 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
     }
   }
 
-  void batch_init(const std::string& name, const std::string& work_dir,
-                  const std::vector<int>& degree) override {
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
     size_t vnum = degree.size();
     adj_lists_.open(work_dir + "/" + name + ".adj", false);
     adj_lists_.resize(vnum);
@@ -367,6 +653,7 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
       adj_lists_[i].init(ptr, deg, 0);
       ptr += deg;
     }
+    return edge_num;
   }
 
   void open(const std::string& name, const std::string& snapshot_dir,
@@ -546,7 +833,7 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
   }
 
   void put_edge(vid_t src, vid_t dst, const EDATA_T& data, timestamp_t ts,
-                Allocator& alloc) {
+                Allocator& alloc) override {
     CHECK_LT(src, adj_lists_.size());
     locks_[src].lock();
     adj_lists_[src].put_edge(dst, data, ts, alloc);
@@ -593,59 +880,333 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
   mmap_array<nbr_t> nbr_list_;
 };
 
-template <typename EDATA_T>
-class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
+template <>
+class MutableCsr<std::string_view>
+    : public TypedMutableCsrBase<std::string_view> {
  public:
-  using nbr_t = MutableNbr<EDATA_T>;
-  using slice_t = MutableNbrSlice<EDATA_T>;
-  using mut_slice_t = MutableNbrSliceMut<EDATA_T>;
+  using nbr_t = MutableNbr<size_t>;
+  using adjlist_t = MutableAdjlist<std::string_view>;
+  using slice_t = MutableNbrSlice<std::string_view>;
+  using mut_slice_t = MutableNbrSliceMut<std::string_view>;
 
-  SingleMutableCsr() {}
-  ~SingleMutableCsr() {}
+  MutableCsr(StringColumn& column, std::atomic<size_t>& column_idx)
+      : column_(column), column_idx_(column_idx), locks_(nullptr) {}
+  ~MutableCsr() {
+    if (locks_ != nullptr) {
+      delete[] locks_;
+    }
+  }
 
-  void batch_init(const std::string& name, const std::string& work_dir,
-                  const std::vector<int>& degree) override {
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
     size_t vnum = degree.size();
-    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
-    nbr_list_.resize(vnum);
-    for (size_t k = 0; k != vnum; ++k) {
-      nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+    adj_lists_.open(work_dir + "/" + name + ".adj", false);
+    adj_lists_.resize(vnum);
+
+    locks_ = new grape::SpinLock[vnum];
+
+    size_t edge_num = 0;
+    for (auto d : degree) {
+      edge_num += d;
+    }
+    nbr_list_.open(work_dir + "/" + name + ".nbr", false);
+    nbr_list_.resize(edge_num);
+
+    nbr_t* ptr = nbr_list_.data();
+    for (vid_t i = 0; i < vnum; ++i) {
+      int deg = degree[i];
+      adj_lists_[i].init(ptr, deg, 0);
+      ptr += deg;
     }
+    return edge_num;
   }
 
   void open(const std::string& name, const std::string& snapshot_dir,
             const std::string& work_dir) override {
-    if (!std::filesystem::exists(work_dir + "/" + name + ".snbr")) {
-      copy_file(snapshot_dir + "/" + name + ".snbr",
-                work_dir + "/" + name + ".snbr");
-    }
-    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
-  }
+    mmap_array<int> degree_list;
+    degree_list.open(snapshot_dir + "/" + name + ".deg", true);
+    nbr_list_.open(snapshot_dir + "/" + name + ".nbr", true);
+    nbr_list_.touch(work_dir + "/" + name + ".nbr");
+    adj_lists_.open(work_dir + "/" + name + ".adj", false);
 
-  void dump(const std::string& name,
-            const std::string& new_snapshot_dir) override {
-    assert(!nbr_list_.filename().empty() &&
-           std::filesystem::exists(nbr_list_.filename()));
-    assert(!nbr_list_.read_only());
-    std::filesystem::create_hard_link(nbr_list_.filename(),
-                                      new_snapshot_dir + "/" + name + ".snbr");
-  }
+    adj_lists_.resize(degree_list.size());
+    locks_ = new grape::SpinLock[degree_list.size()];
 
-  void resize(vid_t vnum) override {
-    if (vnum > nbr_list_.size()) {
-      size_t old_size = nbr_list_.size();
-      nbr_list_.resize(vnum);
-      for (size_t k = old_size; k != vnum; ++k) {
-        nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+    std::string degree_list_idx_file = snapshot_dir + "/" + name + ".deg.idx";
+    if (std::filesystem::exists(degree_list_idx_file)) {
+      mmap_array<uint64_t> degree_list_idx;
+      degree_list_idx.open(degree_list_idx_file, true);
+      uint64_t chunk_size = degree_list_idx[0];
+      uint64_t chunk_num = degree_list_idx.size();
+      int concurrency = std::thread::hardware_concurrency();
+      std::vector<std::thread> threads;
+      std::atomic<uint64_t> chunk_i(0);
+      for (int i = 0; i < concurrency; ++i) {
+        threads.emplace_back([&]() {
+          while (true) {
+            uint64_t cur_chunk = chunk_i.fetch_add(1);
+            if (cur_chunk >= chunk_num) {
+              break;
+            }
+            uint64_t begin = cur_chunk * chunk_size;
+            uint64_t end = std::min(begin + chunk_size, degree_list.size());
+
+            uint64_t offset = cur_chunk == 0 ? 0 : degree_list_idx[cur_chunk];
+            nbr_t* ptr = nbr_list_.data() + offset;
+            while (begin < end) {
+              int degree = degree_list[begin];
+              adj_lists_[begin].init(ptr, degree, degree);
+              ptr += degree;
+              ++begin;
+            }
+          }
+        });
+      }
+      for (auto& thrd : threads) {
+        thrd.join();
       }
+
     } else {
-      nbr_list_.resize(vnum);
+      nbr_t* ptr = nbr_list_.data();
+      for (size_t i = 0; i < degree_list.size(); ++i) {
+        int degree = degree_list[i];
+        adj_lists_[i].init(ptr, degree, degree);
+        ptr += degree;
+      }
     }
   }
 
-  size_t size() const override { return nbr_list_.size(); }
-
-  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
+  void warmup(int thread_num) const override {
+    size_t vnum = adj_lists_.size();
+    std::vector<std::thread> threads;
+    std::atomic<size_t> v_i(0);
+    const size_t chunk = 4096;
+    std::atomic<size_t> output(0);
+    for (int i = 0; i < thread_num; ++i) {
+      threads.emplace_back([&]() {
+        size_t ret = 0;
+        while (true) {
+          size_t begin = std::min(v_i.fetch_add(chunk), vnum);
+          size_t end = std::min(begin + chunk, vnum);
+
+          if (begin == end) {
+            break;
+          }
+
+          while (begin < end) {
+            auto adj_list = get_edges(begin);
+            for (auto& nbr : adj_list) {
+              ret += nbr.get_neighbor();
+            }
+            ++begin;
+          }
+        }
+        output.fetch_add(ret);
+      });
+    }
+    for (auto& thrd : threads) {
+      thrd.join();
+    }
+    (void) output.load();
+  }
+
+  void dump(const std::string& name,
+            const std::string& new_spanshot_dir) override {
+    size_t vnum = adj_lists_.size();
+    bool reuse_nbr_list = true;
+    mmap_array<int> degree_list;
+    degree_list.open(new_spanshot_dir + "/" + name + ".deg", false);
+    degree_list.resize(vnum);
+    size_t offset = 0;
+    for (size_t i = 0; i < vnum; ++i) {
+      if (adj_lists_[i].size() != 0) {
+        if (!(adj_lists_[i].data() == nbr_list_.data() + offset &&
+              offset < nbr_list_.size())) {
+          reuse_nbr_list = false;
+        }
+      }
+      degree_list[i] = adj_lists_[i].size();
+      offset += degree_list[i];
+    }
+
+    {
+      size_t input_size = degree_list.size();
+      std::string degree_list_idx_file =
+          new_spanshot_dir + "/" + name + ".deg.idx";
+      if (input_size > 128 * 1024 &&
+          !std::filesystem::exists(degree_list_idx_file)) {
+        mmap_array<uint64_t> degree_list_idx;
+        degree_list_idx.open(degree_list_idx_file, false);
+
+        const uint64_t chunk_num = 128;
+        degree_list_idx.resize(chunk_num);
+
+        uint64_t chunk_size = (input_size + chunk_num - 1) / chunk_num;
+        uint64_t sum = 0;
+        for (size_t i = 0; i < input_size; ++i) {
+          // sum += degree_list[i];
+          if (i % chunk_size == 0) {
+            degree_list_idx[i / chunk_size] = sum;
+          }
+          sum += degree_list[i];
+        }
+        degree_list_idx[0] = chunk_size;
+      }
+    }
+
+    if (reuse_nbr_list && !nbr_list_.filename().empty() &&
+        std::filesystem::exists(nbr_list_.filename())) {
+      std::filesystem::create_hard_link(nbr_list_.filename(),
+                                        new_spanshot_dir + "/" + name + ".nbr");
+    } else {
+      FILE* fout =
+          fopen((new_spanshot_dir + "/" + name + ".nbr").c_str(), "wb");
+      for (size_t i = 0; i < vnum; ++i) {
+        fwrite(adj_lists_[i].data(), sizeof(nbr_t), adj_lists_[i].size(), fout);
+      }
+      fflush(fout);
+      fclose(fout);
+    }
+  }
+
+  void resize(vid_t vnum) override {
+    if (vnum > adj_lists_.size()) {
+      size_t old_size = adj_lists_.size();
+      adj_lists_.resize(vnum);
+      for (size_t k = old_size; k != vnum; ++k) {
+        adj_lists_[k].init(NULL, 0, 0);
+      }
+      delete[] locks_;
+      locks_ = new grape::SpinLock[vnum];
+    } else {
+      adj_lists_.resize(vnum);
+    }
+  }
+
+  size_t size() const override { return adj_lists_.size(); }
+
+  void put_generic_edge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                        Allocator& alloc) override {
+    LOG(FATAL) << "not implemented\n";
+  }
+
+  void put_edge(vid_t src, vid_t dst, size_t data, timestamp_t ts,
+                Allocator& alloc) {
+    CHECK_LT(src, adj_lists_.size());
+    locks_[src].lock();
+    adj_lists_[src].put_edge(dst, data, ts, alloc);
+    locks_[src].unlock();
+  }
+  void put_edge_with_index(vid_t src, vid_t dst, size_t index, timestamp_t ts,
+                           Allocator& alloc) override {
+    put_edge(src, dst, index, ts, alloc);
+  }
+
+  void batch_put_edge_with_index(vid_t src, vid_t dst, const size_t& data,
+                                 timestamp_t ts = 0) override {
+    adj_lists_[src].batch_put_edge(dst, data, ts);
+  }
+
+  int degree(vid_t i) const { return adj_lists_[i].size(); }
+
+  slice_t get_edges(vid_t i) const override {
+    return adj_lists_[i].get_edges(column_);
+  }
+  mut_slice_t get_edges_mut(vid_t i) {
+    return adj_lists_[i].get_edges_mut(column_);
+  }
+
+  void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
+                   Allocator& alloc) override {
+    auto row_id = column_idx_.load();
+    put_edge(src, dst, row_id - 1, ts, alloc);
+  }
+
+  void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
+                        timestamp_t ts, Allocator& alloc) override {
+    std::string_view sw;
+    arc >> sw;
+    auto row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, sw);
+    put_edge(src, dst, row_id, ts, alloc);
+  }
+
+  std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
+      vid_t v) const override {
+    return std::make_shared<TypedMutableCsrConstEdgeIter<std::string_view>>(
+        get_edges(v));
+  }
+
+  MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
+    return new TypedMutableCsrConstEdgeIter<std::string_view>(get_edges(v));
+  }
+  std::shared_ptr<MutableCsrEdgeIterBase> edge_iter_mut(vid_t v) override {
+    return std::make_shared<TypedMutableCsrEdgeIter<std::string_view>>(
+        get_edges_mut(v));
+  }
+
+ private:
+  grape::SpinLock* locks_;
+  mmap_array<adjlist_t> adj_lists_;
+  mmap_array<nbr_t> nbr_list_;
+  StringColumn& column_;
+  std::atomic<size_t>& column_idx_;
+};
+
+template <typename EDATA_T>
+class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
+ public:
+  using nbr_t = MutableNbr<EDATA_T>;
+  using slice_t = MutableNbrSlice<EDATA_T>;
+  using mut_slice_t = MutableNbrSliceMut<EDATA_T>;
+
+  SingleMutableCsr() {}
+  ~SingleMutableCsr() {}
+
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
+    size_t vnum = degree.size();
+    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
+    nbr_list_.resize(vnum);
+    for (size_t k = 0; k != vnum; ++k) {
+      nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+    }
+    return vnum;
+  }
+
+  void open(const std::string& name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {
+    if (!std::filesystem::exists(work_dir + "/" + name + ".snbr")) {
+      copy_file(snapshot_dir + "/" + name + ".snbr",
+                work_dir + "/" + name + ".snbr");
+    }
+    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
+  }
+
+  void dump(const std::string& name,
+            const std::string& new_snapshot_dir) override {
+    assert(!nbr_list_.filename().empty() &&
+           std::filesystem::exists(nbr_list_.filename()));
+    assert(!nbr_list_.read_only());
+    std::filesystem::create_hard_link(nbr_list_.filename(),
+                                      new_snapshot_dir + "/" + name + ".snbr");
+  }
+
+  void resize(vid_t vnum) override {
+    if (vnum > nbr_list_.size()) {
+      size_t old_size = nbr_list_.size();
+      nbr_list_.resize(vnum);
+      for (size_t k = old_size; k != vnum; ++k) {
+        nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+      }
+    } else {
+      nbr_list_.resize(vnum);
+    }
+  }
+
+  size_t size() const override { return nbr_list_.size(); }
+
+  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
                       timestamp_t ts = 0) override {
     nbr_list_[src].neighbor = dst;
     nbr_list_[src].data = data;
@@ -662,7 +1223,7 @@ class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
   }
 
   void put_edge(vid_t src, vid_t dst, const EDATA_T& data, timestamp_t ts,
-                Allocator&) {
+                Allocator&) override {
     CHECK_LT(src, nbr_list_.size());
     nbr_list_[src].neighbor = dst;
     nbr_list_[src].data = data;
@@ -758,6 +1319,186 @@ class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
   mmap_array<nbr_t> nbr_list_;
 };
 
+template <>
+class SingleMutableCsr<std::string_view>
+    : public TypedMutableCsrBase<std::string_view> {
+ public:
+  using nbr_t = MutableNbr<size_t>;
+  using slice_t = MutableNbrSlice<std::string_view>;
+  using mut_slice_t = MutableNbrSliceMut<std::string_view>;
+
+  SingleMutableCsr(StringColumn& column, std::atomic<size_t>& column_idx)
+      : column_(column), column_idx_(column_idx) {}
+  ~SingleMutableCsr() {}
+
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
+    size_t vnum = degree.size();
+    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
+    nbr_list_.resize(vnum);
+    for (size_t k = 0; k != vnum; ++k) {
+      nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+    }
+    return vnum;
+  }
+
+  void open(const std::string& name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {
+    if (!std::filesystem::exists(work_dir + "/" + name + ".snbr")) {
+      copy_file(snapshot_dir + "/" + name + ".snbr",
+                work_dir + "/" + name + ".snbr");
+    }
+    nbr_list_.open(work_dir + "/" + name + ".snbr", false);
+  }
+
+  void dump(const std::string& name,
+            const std::string& new_snapshot_dir) override {
+    assert(!nbr_list_.filename().empty() &&
+           std::filesystem::exists(nbr_list_.filename()));
+    assert(!nbr_list_.read_only());
+    std::filesystem::create_hard_link(nbr_list_.filename(),
+                                      new_snapshot_dir + "/" + name + ".snbr");
+  }
+
+  void resize(vid_t vnum) override {
+    if (vnum > nbr_list_.size()) {
+      size_t old_size = nbr_list_.size();
+      nbr_list_.resize(vnum);
+      for (size_t k = old_size; k != vnum; ++k) {
+        nbr_list_[k].timestamp.store(std::numeric_limits<timestamp_t>::max());
+      }
+    } else {
+      nbr_list_.resize(vnum);
+    }
+  }
+
+  size_t size() const override { return nbr_list_.size(); }
+
+  void put_generic_edge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                        Allocator& alloc) override {
+    LOG(FATAL) << "not implemented\n";
+  }
+
+  void put_edge(vid_t src, vid_t dst, size_t data, timestamp_t ts, Allocator&) {
+    CHECK_LT(src, nbr_list_.size());
+    nbr_list_[src].neighbor = dst;
+    nbr_list_[src].data = data;
+    CHECK_EQ(nbr_list_[src].timestamp, std::numeric_limits<timestamp_t>::max());
+    nbr_list_[src].timestamp.store(ts);
+  }
+
+  slice_t get_edges(vid_t i) const override {
+    slice_t ret(column_);
+    ret.set_size(nbr_list_[i].timestamp.load() ==
+                         std::numeric_limits<timestamp_t>::max()
+                     ? 0
+                     : 1);
+    if (ret.size() != 0) {
+      ret.set_begin(&nbr_list_[i]);
+    }
+    return ret;
+  }
+
+  mut_slice_t get_edges_mut(vid_t i) {
+    mut_slice_t ret(column_);
+    ret.set_size(nbr_list_[i].timestamp.load() ==
+                         std::numeric_limits<timestamp_t>::max()
+                     ? 0
+                     : 1);
+    if (ret.size() != 0) {
+      ret.set_begin(&nbr_list_[i]);
+    }
+    return ret;
+  }
+
+  const MutableNbr<std::string_view>& get_edge(vid_t i) const {
+    nbr_.neighbor = nbr_list_[i].neighbor;
+    nbr_.timestamp.store(nbr_list_[i].timestamp.load());
+    nbr_.data = column_.get_view(nbr_list_[i].data);
+    return nbr_;
+  }
+
+  void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
+                   Allocator& alloc) override {
+    auto row_id = column_idx_.load();
+    put_edge(src, dst, row_id - 1, ts, alloc);
+  }
+
+  void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
+                        timestamp_t ts, Allocator& alloc) override {
+    std::string_view sw;
+    arc >> sw;
+
+    auto row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, sw);
+    put_edge(src, dst, row_id, ts, alloc);
+  }
+  void put_edge_with_index(vid_t src, vid_t dst, size_t index, timestamp_t ts,
+                           Allocator& alloc) override {
+    put_edge(src, dst, index, ts, alloc);
+  }
+
+  void batch_put_edge_with_index(vid_t src, vid_t dst, const size_t& data,
+                                 timestamp_t ts = 0) override {
+    nbr_list_[src].neighbor = dst;
+    nbr_list_[src].data = data;
+    CHECK_EQ(nbr_list_[src].timestamp.load(),
+             std::numeric_limits<timestamp_t>::max());
+    nbr_list_[src].timestamp.store(ts);
+  }
+
+  std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
+      vid_t v) const override {
+    return std::make_shared<TypedMutableCsrConstEdgeIter<std::string_view>>(
+        get_edges(v));
+  }
+
+  MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
+    return new TypedMutableCsrConstEdgeIter<std::string_view>(get_edges(v));
+  }
+
+  std::shared_ptr<MutableCsrEdgeIterBase> edge_iter_mut(vid_t v) override {
+    return std::make_shared<TypedMutableCsrEdgeIter<std::string_view>>(
+        get_edges_mut(v));
+  }
+
+  void warmup(int thread_num) const override {
+    size_t vnum = nbr_list_.size();
+    std::vector<std::thread> threads;
+    std::atomic<size_t> v_i(0);
+    std::atomic<size_t> output(0);
+    const size_t chunk = 4096;
+    for (int i = 0; i < thread_num; ++i) {
+      threads.emplace_back([&]() {
+        size_t ret = 0;
+        while (true) {
+          size_t begin = std::min(v_i.fetch_add(chunk), vnum);
+          size_t end = std::min(begin + chunk, vnum);
+          if (begin == end) {
+            break;
+          }
+          while (begin < end) {
+            auto& nbr = nbr_list_[begin];
+            ret += nbr.neighbor;
+            ++begin;
+          }
+        }
+        output.fetch_add(ret);
+      });
+    }
+    for (auto& thrd : threads) {
+      thrd.join();
+    }
+    (void) output.load();
+  }
+
+ private:
+  mmap_array<nbr_t> nbr_list_;
+  mutable MutableNbr<std::string_view> nbr_;
+  std::atomic<size_t>& column_idx_;
+  StringColumn& column_;
+};
+
 template <typename EDATA_T>
 class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
   using slice_t = MutableNbrSlice<EDATA_T>;
@@ -766,8 +1507,10 @@ class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
   EmptyCsr() = default;
   ~EmptyCsr() = default;
 
-  void batch_init(const std::string& name, const std::string& work_dir,
-                  const std::vector<int>& degree) override {}
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
+    return 0;
+  }
 
   void open(const std::string& name, const std::string& snapshot_dir,
             const std::string& work_dir) override {}
@@ -788,7 +1531,8 @@ class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
 
   void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
                       timestamp_t ts = 0) override {}
-
+  void put_edge(vid_t src, vid_t dst, const EDATA_T& data, timestamp_t ts,
+                Allocator&) override {}
   void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
                    Allocator&) override {
     EDATA_T value;
@@ -797,6 +1541,7 @@ class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
 
   void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
                         const timestamp_t ts, Allocator&) override {}
+
   std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
       vid_t v) const override {
     return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(
@@ -811,6 +1556,71 @@ class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
         MutableNbrSliceMut<EDATA_T>::empty());
   }
 };
+
+template <>
+class EmptyCsr<std::string_view>
+    : public TypedMutableCsrBase<std::string_view> {
+  using slice_t = MutableNbrSlice<std::string_view>;
+
+ public:
+  EmptyCsr(StringColumn& column, std::atomic<size_t>& column_idx)
+      : column_(column), column_idx_(column_idx) {}
+  ~EmptyCsr() = default;
+
+  size_t batch_init(const std::string& name, const std::string& work_dir,
+                    const std::vector<int>& degree) override {
+    return 0;
+  }
+
+  void open(const std::string& name, const std::string& snapshot_dir,
+            const std::string& work_dir) override {}
+
+  void dump(const std::string& name,
+            const std::string& new_spanshot_dir) override {}
+
+  void warmup(int thread_num) const override {}
+
+  void resize(vid_t vnum) override {}
+
+  size_t size() const override { return 0; }
+
+  slice_t get_edges(vid_t i) const override { return slice_t::empty(column_); }
+
+  void put_generic_edge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                        Allocator&) override {
+    LOG(FATAL) << "not implemented\n";
+  }
+
+  void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
+                   Allocator&) override {}
+
+  void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
+                        const timestamp_t ts, Allocator&) override {
+    std::string_view sw;
+    arc >> sw;
+    auto row_id = column_idx_.fetch_add(1);
+    column_.set_value(row_id, sw);
+  }
+  void put_edge_with_index(vid_t src, vid_t dst, size_t index, timestamp_t ts,
+                           Allocator& alloc) override {}
+  void batch_put_edge_with_index(vid_t src, vid_t dst, const size_t& data,
+                                 timestamp_t ts = 0) override {}
+  std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
+      vid_t v) const override {
+    return std::make_shared<TypedMutableCsrConstEdgeIter<std::string_view>>(
+        MutableNbrSlice<std::string_view>::empty(column_));
+  }
+  MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
+    return new TypedMutableCsrConstEdgeIter<std::string_view>(
+        MutableNbrSlice<std::string_view>::empty(column_));
+  }
+  std::shared_ptr<MutableCsrEdgeIterBase> edge_iter_mut(vid_t v) override {
+    return std::make_shared<TypedMutableCsrEdgeIter<std::string_view>>(
+        MutableNbrSliceMut<std::string_view>::empty(column_));
+  }
+  std::atomic<size_t>& column_idx_;
+  StringColumn& column_;
+};
 }  // namespace gs
 
 #endif  // GRAPHSCOPE_GRAPH_MUTABLE_CSR_H_
diff --git a/flex/storages/rt_mutable_graph/mutable_property_fragment.cc b/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
index f8187923d4ad..72ba564c71bb 100644
--- a/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
+++ b/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
@@ -36,11 +36,12 @@ MutablePropertyFragment::~MutablePropertyFragment() {
                        dst_label * edge_label_num_ + e_label;
         if (ie_[index] != NULL) {
           ie_[index]->resize(degree_list[dst_label]);
-          delete ie_[index];
         }
         if (oe_[index] != NULL) {
           oe_[index]->resize(degree_list[src_label]);
-          delete oe_[index];
+        }
+        if (dual_csr_list_[index] != NULL) {
+          delete dual_csr_list_[index];
         }
       }
     }
@@ -55,12 +56,7 @@ void MutablePropertyFragment::loadSchema(const std::string& schema_path) {
 }
 
 void MutablePropertyFragment::Clear() {
-  for (auto ptr : ie_) {
-    if (ptr != NULL) {
-      delete ptr;
-    }
-  }
-  for (auto ptr : oe_) {
+  for (auto ptr : dual_csr_list_) {
     if (ptr != NULL) {
       delete ptr;
     }
@@ -69,6 +65,7 @@ void MutablePropertyFragment::Clear() {
   vertex_data_.clear();
   ie_.clear();
   oe_.clear();
+  dual_csr_list_.clear();
   vertex_label_num_ = 0;
   edge_label_num_ = 0;
   schema_.Clear();
@@ -82,79 +79,29 @@ void MutablePropertyFragment::DumpSchema(const std::string& schema_path) {
   io_adaptor->Close();
 }
 
-inline MutableCsrBase* create_csr(EdgeStrategy es,
-                                  const std::vector<PropertyType>& properties) {
+inline DualCsrBase* create_csr(EdgeStrategy oes, EdgeStrategy ies,
+                               const std::vector<PropertyType>& properties) {
   if (properties.empty()) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<grape::EmptyType>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<grape::EmptyType>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<grape::EmptyType>();
-    }
-  } else if (properties[0] == PropertyType::kBool) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<bool>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<bool>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<bool>();
-    }
-  } else if (properties[0] == PropertyType::kInt32) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<int>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<int>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<int>();
-    }
-  } else if (properties[0] == PropertyType::kUInt32) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<unsigned int>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<unsigned int>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<unsigned int>();
-    }
-  } else if (properties[0] == PropertyType::kDate) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<Date>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<Date>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<Date>();
-    }
-  } else if (properties[0] == PropertyType::kInt64) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<int64_t>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<int64_t>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<int64_t>();
-    }
-  } else if (properties[0] == PropertyType::kUInt64) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<uint64_t>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<uint64_t>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<uint64_t>();
-    }
-  } else if (properties[0] == PropertyType::kDouble) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<double>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<double>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<double>();
-    }
-  } else if (properties[0] == PropertyType::kFloat) {
-    if (es == EdgeStrategy::kSingle) {
-      return new SingleMutableCsr<float>();
-    } else if (es == EdgeStrategy::kMultiple) {
-      return new MutableCsr<float>();
-    } else if (es == EdgeStrategy::kNone) {
-      return new EmptyCsr<float>();
+    return new DualCsr<grape::EmptyType>(oes, ies);
+  } else if (properties.size() == 1) {
+    if (properties[0] == PropertyType::kBool) {
+      return new DualCsr<bool>(oes, ies);
+    } else if (properties[0] == PropertyType::kInt32) {
+      return new DualCsr<int32_t>(oes, ies);
+    } else if (properties[0] == PropertyType::kUInt32) {
+      return new DualCsr<uint32_t>(oes, ies);
+    } else if (properties[0] == PropertyType::kDate) {
+      return new DualCsr<Date>(oes, ies);
+    } else if (properties[0] == PropertyType::kInt64) {
+      return new DualCsr<int64_t>(oes, ies);
+    } else if (properties[0] == PropertyType::kUInt64) {
+      return new DualCsr<uint64_t>(oes, ies);
+    } else if (properties[0] == PropertyType::kDouble) {
+      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);
     }
   }
   LOG(FATAL) << "not support edge strategy or edge data type";
@@ -195,6 +142,9 @@ void MutablePropertyFragment::Open(const std::string& work_dir) {
   ie_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_, NULL);
   oe_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_, NULL);
 
+  dual_csr_list_.resize(vertex_label_num_ * vertex_label_num_ * edge_label_num_,
+                        NULL);
+
   for (size_t src_label_i = 0; src_label_i != vertex_label_num_;
        ++src_label_i) {
     std::string src_label =
@@ -217,13 +167,17 @@ void MutablePropertyFragment::Open(const std::string& work_dir) {
             src_label, dst_label, edge_label);
         EdgeStrategy ie_strategy = schema_.get_incoming_edge_strategy(
             src_label, dst_label, edge_label);
-        ie_[index] = create_csr(ie_strategy, properties);
-        oe_[index] = create_csr(oe_strategy, properties);
-        ie_[index]->open(ie_prefix(src_label, dst_label, edge_label),
-                         snapshot_dir, tmp_dir_path);
+        dual_csr_list_[index] =
+            create_csr(oe_strategy, ie_strategy, properties);
+        ie_[index] = dual_csr_list_[index]->GetInCsr();
+        oe_[index] = dual_csr_list_[index]->GetOutCsr();
+        dual_csr_list_[index]->Open(
+            oe_prefix(src_label, dst_label, edge_label),
+            ie_prefix(src_label, dst_label, edge_label),
+            edata_prefix(src_label, dst_label, edge_label), snapshot_dir,
+            tmp_dir_path);
         ie_[index]->resize(vertex_capacities[dst_label_i]);
-        oe_[index]->open(oe_prefix(src_label, dst_label, edge_label),
-                         snapshot_dir, tmp_dir_path);
+
         oe_[index]->resize(vertex_capacities[src_label_i]);
       }
     }
@@ -260,15 +214,14 @@ void MutablePropertyFragment::Dump(const std::string& work_dir,
         }
         size_t index = src_label_i * vertex_label_num_ * edge_label_num_ +
                        dst_label_i * edge_label_num_ + e_label_i;
-        if (ie_[index] != NULL) {
+        if (dual_csr_list_[index] != NULL) {
           ie_[index]->resize(vertex_num[dst_label_i]);
-          ie_[index]->dump(ie_prefix(src_label, dst_label, edge_label),
-                           snapshot_dir_path);
-        }
-        if (oe_[index] != NULL) {
           oe_[index]->resize(vertex_num[src_label_i]);
-          oe_[index]->dump(oe_prefix(src_label, dst_label, edge_label),
-                           snapshot_dir_path);
+          dual_csr_list_[index]->Dump(
+              oe_prefix(src_label, dst_label, edge_label),
+              ie_prefix(src_label, dst_label, edge_label),
+              edata_prefix(src_label, dst_label, edge_label),
+              snapshot_dir_path);
         }
       }
     }
@@ -305,6 +258,14 @@ void MutablePropertyFragment::IngestEdge(label_t src_label, vid_t src_lid,
   oe_[index]->ingest_edge(src_lid, dst_lid, arc, ts, alloc);
 }
 
+void MutablePropertyFragment::UpdateEdge(label_t src_label, vid_t src_lid,
+                                         label_t dst_label, vid_t dst_lid,
+                                         label_t edge_label, timestamp_t ts,
+                                         const Any& arc, Allocator& alloc) {
+  size_t index = src_label * vertex_label_num_ * edge_label_num_ +
+                 dst_label * edge_label_num_ + edge_label;
+  dual_csr_list_[index]->UpdateEdge(src_lid, dst_lid, arc, ts, alloc);
+}
 const Schema& MutablePropertyFragment::schema() const { return schema_; }
 
 Schema& MutablePropertyFragment::mutable_schema() { return schema_; }
diff --git a/flex/storages/rt_mutable_graph/mutable_property_fragment.h b/flex/storages/rt_mutable_graph/mutable_property_fragment.h
index eee57395a37a..b041ac46d199 100644
--- a/flex/storages/rt_mutable_graph/mutable_property_fragment.h
+++ b/flex/storages/rt_mutable_graph/mutable_property_fragment.h
@@ -22,6 +22,7 @@
 
 #include "flex/storages/rt_mutable_graph/schema.h"
 
+#include "flex/storages/rt_mutable_graph/dual_csr.h"
 #include "flex/storages/rt_mutable_graph/mutable_csr.h"
 #include "flex/storages/rt_mutable_graph/types.h"
 #include "flex/utils/arrow_utils.h"
@@ -43,6 +44,10 @@ class MutablePropertyFragment {
                   vid_t dst_lid, label_t edge_label, timestamp_t ts,
                   grape::OutArchive& arc, Allocator& alloc);
 
+  void UpdateEdge(label_t src_label, vid_t src_lid, label_t dst_label,
+                  vid_t dst_lid, label_t edge_label, timestamp_t ts,
+                  const Any& arc, Allocator& alloc);
+
   void Open(const std::string& work_dir);
 
   void Warmup(int thread_num);
@@ -105,6 +110,7 @@ class MutablePropertyFragment {
   Schema schema_;
   std::vector<LFIndexer<vid_t>> lf_indexers_;
   std::vector<MutableCsrBase*> ie_, oe_;
+  std::vector<DualCsrBase*> dual_csr_list_;
   std::vector<Table> vertex_data_;
 
   size_t vertex_label_num_, edge_label_num_;


Test patch:
diff --git a/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml b/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml
new file mode 100644
index 000000000000..58576c0d309d
--- /dev/null
+++ b/flex/tests/rt_mutable_graph/modern_graph_string_edge.yaml
@@ -0,0 +1,71 @@
+name: modern_graph # then must have a modern dir under ${data} directory
+store_type: mutable_csr  # v6d, groot, gart
+schema:
+  vertex_types:
+    - type_id: 0
+      type_name: person
+      x_csr_params:
+        max_vertex_num: 100
+      properties:
+        - property_id: 0
+          property_name: id
+          property_type:
+            primitive_type: DT_SIGNED_INT64
+        - property_id: 1
+          property_name: name
+          property_type:
+            primitive_type: DT_STRING
+        - property_id: 2
+          property_name: age
+          property_type:
+            primitive_type: DT_SIGNED_INT32
+      primary_keys:
+        - id
+    - type_id: 1
+      type_name: software
+      x_csr_params:
+        max_vertex_num: 100
+      properties:
+        - property_id: 0
+          property_name: id
+          property_type:
+            primitive_type: DT_SIGNED_INT64
+        - property_id: 1
+          property_name: name
+          property_type:
+            primitive_type: DT_STRING
+        - property_id: 2
+          property_name: lang
+          property_type:
+            primitive_type: DT_STRING
+      primary_keys:
+        - id
+  edge_types:
+    - type_id: 0
+      type_name: knows
+      vertex_type_pair_relations:
+        - source_vertex: person
+          destination_vertex: person
+          relation: MANY_TO_MANY
+          x_csr_params:
+            incoming_edge_strategy: Multiple
+            outgoing_edge_strategy: Multiple
+      properties:
+        - property_id: 0
+          property_name: weight
+          property_type:
+            primitive_type: DT_DOUBLE
+    - type_id: 1
+      type_name: created
+      vertex_type_pair_relations:
+        - source_vertex: person
+          destination_vertex: software
+          relation: ONE_TO_MANY
+          x_csr_params:
+            incoming_edge_strategy: Multiple
+            outgoing_edge_strategy: Single
+      properties:
+        - property_id: 0
+          property_name: weight
+          property_type:
+            primitive_type: DT_STRING
\ No newline at end of file
diff --git a/flex/tests/rt_mutable_graph/string_edge_property_test.cc b/flex/tests/rt_mutable_graph/string_edge_property_test.cc
new file mode 100644
index 000000000000..cff81b31c40a
--- /dev/null
+++ b/flex/tests/rt_mutable_graph/string_edge_property_test.cc
@@ -0,0 +1,144 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "grape/util.h"
+
+#include "flex/engines/graph_db/database/graph_db.h"
+
+#include <glog/logging.h>
+
+namespace gs {
+class TestStringEdgeProperty {
+ public:
+  TestStringEdgeProperty(GraphDB& db)
+      : db_(db),
+        src_label_(db.graph().schema().get_vertex_label_id("person")),
+        dst_label_(db.graph().schema().get_vertex_label_id("software")),
+        edge_label_(db.graph().schema().get_edge_label_id("created")) {}
+  void test() {
+    int64_t src, dst;
+    src = 1;
+    test_get_edge(src);
+    src = 1;
+    dst = 3;
+    test_get_graph_view(src, dst);
+    src = 3;
+    dst = 5;
+    test_add_edge(src, dst);
+  }
+  void test_get_edge(int64_t oid) {
+    vid_t src_lid;
+    CHECK(db_.graph().get_lid(src_label_, oid, src_lid));
+    auto oe = db_.graph().get_outgoing_edges(src_label_, src_lid, dst_label_,
+                                             edge_label_);
+    CHECK(oe != nullptr) << "Got nullptr oe\n";
+    CHECK(oe->get_data().type == PropertyType::kString)
+        << "Inconsistent type, Except: string, Got " << oe->get_data().type;
+    std::cout << oe->get_data().AsStringView() << "\n";
+    LOG(INFO) << "Finish test get edge\n";
+  }
+
+  void test_get_graph_view(int64_t src, int64_t dst) {
+    auto txn = db_.GetReadTransaction();
+    vid_t src_lid, dst_lid;
+
+    CHECK(db_.graph().get_lid(src_label_, src, src_lid));
+    CHECK(db_.graph().get_lid(dst_label_, dst, dst_lid));
+
+    {
+      auto graph_view = txn.GetOutgoingSingleGraphView<std::string_view>(
+          src_label_, dst_label_, edge_label_);
+      auto oe = graph_view.get_edge(src_lid);
+      LOG(INFO) << oe.get_data() << "\n";
+      CHECK(oe.get_data() == "0.4")
+          << "Inconsistent value, Excepted : 0.4, Got " << oe.get_data()
+          << "\n";
+    }
+    {
+      auto graph_view = txn.GetIncomingGraphView<std::string_view>(
+          dst_label_, src_label_, edge_label_);
+      auto ie = graph_view.get_edges(dst_lid);
+      for (auto& e : ie) {
+        LOG(INFO) << e.get_data() << "\n";
+      }
+    }
+    LOG(INFO) << "Finish test get GraphView\n";
+  }
+
+  void test_add_edge(int64_t src, int64_t dst) {
+    {
+      auto txn = db_.GetSingleVertexInsertTransaction();
+      std::string name = "test-3";
+      int age = 34;
+      CHECK(txn.AddVertex(src_label_, src, {Any::From(name), Any::From(age)}))
+          << "Add vertex failed";
+      txn.Commit();
+    }
+    vid_t src_lid, dst_lid;
+    CHECK(db_.graph().get_lid(src_label_, src, src_lid));
+    CHECK(db_.graph().get_lid(dst_label_, dst, dst_lid));
+
+    {
+      auto txn = db_.GetSingleEdgeInsertTransaction();
+      std::string str = "test";
+      CHECK(txn.AddEdge(src_label_, src, dst_label_, dst, edge_label_,
+                        Any::From(str)))
+          << "Add edge failed\n";
+      txn.Commit();
+    }
+    {
+      auto txn = db_.GetReadTransaction();
+      auto graph_view = txn.GetOutgoingSingleGraphView<std::string_view>(
+          src_label_, dst_label_, edge_label_);
+      auto oe = graph_view.get_edge(src_lid);
+      CHECK(oe.get_data() == "test")
+          << "Inconsistent value, Excepted: test, Got " << oe.get_data()
+          << "\n";
+    }
+    LOG(INFO) << "Finish test add edge\n";
+  }
+
+ private:
+  GraphDB& db_;
+  label_t src_label_;
+  label_t dst_label_;
+  label_t edge_label_;
+};
+
+}  // namespace gs
+// ./string_edge_property_test graph.yaml data_dir
+int main(int argc, char** argv) {
+  bool warmup = false;
+  uint32_t shard_num = 1;
+
+  std::string graph_schema_path = "";
+  std::string data_path = "";
+
+  graph_schema_path = argv[1];
+  data_path = argv[2];
+
+  double t0 = -grape::GetCurrentTime();
+  auto& db = gs::GraphDB::get();
+
+  auto schema = gs::Schema::LoadFromYaml(graph_schema_path);
+  db.Open(schema, data_path, shard_num, warmup);
+
+  t0 += grape::GetCurrentTime();
+
+  LOG(INFO) << "Finished loading graph, elapsed " << t0 << " s";
+  gs::TestStringEdgeProperty(db).test();
+
+  return 0;
+}