Repository: alibaba/GraphScope (language: python)

Issue:
refactor(Interactive): Support more configurable options for rt_mutable_graph bulk loading with csv file
Currently the supported options are very restricted. For example,
the delimiter can only be `|`, and user cannot configure whether to skip header, or some commented lines. 


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/flex.yml b/.github/workflows/flex.yml
index fb5381ef6fea..77267b8e6463 100644
--- a/.github/workflows/flex.yml
+++ b/.github/workflows/flex.yml
@@ -26,7 +26,7 @@ jobs:
     runs-on: ubuntu-20.04
     if: ${{ github.repository == 'alibaba/GraphScope' }}
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.4
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.6
     steps:
     - uses: actions/checkout@v3
 
diff --git a/.github/workflows/hqps-db-ci.yml b/.github/workflows/hqps-db-ci.yml
index 289add5288c9..8f875b64ae36 100644
--- a/.github/workflows/hqps-db-ci.yml
+++ b/.github/workflows/hqps-db-ci.yml
@@ -28,7 +28,7 @@ jobs:
     runs-on: ubuntu-20.04
     if: ${{ github.repository == 'alibaba/GraphScope' }}
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.4
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.6
     steps:
     - uses: actions/checkout@v3
 
diff --git a/flex/.devcontainer.json b/flex/.devcontainer.json
index 7bd8fe4dd2b2..6118d0fbdae8 100644
--- a/flex/.devcontainer.json
+++ b/flex/.devcontainer.json
@@ -3,7 +3,7 @@
 {
     "name": "GraphScope",
     // Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
-    "image": "registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.4",
+    "image": "registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.6",
     // Features to add to the dev container. More info: https://containers.dev/features.
     "features": {
         "ghcr.io/devcontainers/features/common-utils:2": {
diff --git a/flex/CMakeLists.txt b/flex/CMakeLists.txt
index 5169aa6257af..abe74ef335b3 100644
--- a/flex/CMakeLists.txt
+++ b/flex/CMakeLists.txt
@@ -19,6 +19,7 @@ include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../)
 set(DEFAULT_BUILD_TYPE "Release")
 
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++17 -mno-avx512f -fPIC")
+set(CMAKE_CXX_FLAGS_DEBUG "-g3 -O0")
 
 
 add_compile_definitions(FLEX_VERSION="${FLEX_VERSION}")
@@ -61,6 +62,20 @@ find_package(Boost REQUIRED COMPONENTS system filesystem
              # required by folly
              context program_options regex thread)
 
+#find arrow----------------------------------------------------------------------
+include("cmake/FindArrow.cmake")
+if (NOT ARROW_FOUND)
+    message(FATAL_ERROR "arrow not found, please install the arrow library")
+else ()
+    include_directories(SYSTEM ${ARROW_INCLUDE_DIRS})
+    if (TARGET arrow_shared)
+        set(ARROW_SHARED_LIB arrow_shared)
+    endif()
+    if (TARGET arrow_static)
+        set(ARROW_STATIC_LIB arrow_static)
+    endif()
+endif ()
+
 # Find Doxygen
 if (BUILD_DOC)
     find_package(Doxygen)
diff --git a/flex/Dockerfile b/flex/Dockerfile
index d047cab1e212..7d5874a60dd9 100644
--- a/flex/Dockerfile
+++ b/flex/Dockerfile
@@ -5,7 +5,7 @@ ARG CI=false
 SHELL ["/bin/bash", "-c"]
 
 
-RUN apt update && apt -y install locales && locale-gen en_US.UTF-8
+RUN apt-get update && apt-get -y install locales && locale-gen en_US.UTF-8
 ENV LANG en_US.UTF-8
 ENV LANGUAGE en_US:en
 ENV LC_ALL en_US.UTF-8
@@ -15,29 +15,32 @@ ENV TZ=Asia/Shanghai
 RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
 
 # install dependencies
-RUN apt install -y \
-      ninja-build ragel libhwloc-dev libnuma-dev libpciaccess-dev vim wget \
-      git g++ libgoogle-glog-dev cmake libopenmpi-dev default-jdk libcrypto++-dev \
-      libboost-all-dev libxml2-dev
-RUN apt install -y xfslibs-dev libgnutls28-dev liblz4-dev maven openssl pkg-config \
-      libsctp-dev gcc make python3 systemtap-sdt-dev libtool libyaml-cpp-dev \
-      libc-ares-dev stow libfmt-dev diffutils valgrind doxygen python3-pip net-tools
+RUN apt-get update && apt-get install -y vim wget \
+      git g++ libgoogle-glog-dev cmake libopenmpi-dev default-jdk \
+      libboost-all-dev
+RUN apt-get install -y maven openssl \
+      gcc make python3 libyaml-cpp-dev \
+      libc-ares-dev doxygen python3-pip net-tools curl
 
 # install libgrape-lite
-RUN cd /root && \
+RUN cd /tmp && \
     git clone https://github.com/alibaba/libgrape-lite.git -b v0.3.2 --single-branch && cd libgrape-lite && \
-    mkdir build && cd build && cmake .. && make -j && make install
+    mkdir build && cd build && cmake .. && make -j && make install && rm -rf /tmp/libgrape-lite
 
-RUN cp /usr/local/lib/libgrape-lite.so /usr/lib/libgrape-lite.so
-
-RUN git clone https://github.com/alibaba/hiactor.git -b v0.1.1 --single-branch && cd hiactor && \
+RUN cd /tmp && git clone https://github.com/alibaba/hiactor.git -b v0.1.1 --single-branch && cd hiactor && \
     git submodule update --init --recursive && ./seastar/seastar/install-dependencies.sh && mkdir build && cd build && \
     cmake -DHiactor_DEMOS=OFF -DHiactor_TESTING=OFF -DHiactor_DPDK=OFF -DHiactor_CXX_DIALECT=gnu++17 -DSeastar_CXX_FLAGS="-DSEASTAR_DEFAULT_ALLOCATOR -mno-avx512" .. && \
-    make -j && make install
+    make -j && make install && rm -rf /tmp/hiactor
 
 #install protobuf
 RUN apt-get install -y protobuf-compiler libprotobuf-dev
 
+#install arrow
+RUN cd /tmp && apt-get install -y -V ca-certificates lsb-release wget && \
+    curl -o apache-arrow-apt-source-latest.deb https://apache.jfrog.io/artifactory/arrow/$(lsb_release --id --short | tr 'A-Z' 'a-z')/apache-arrow-apt-source-latest-$(lsb_release --codename --short).deb && \
+    apt-get install -y ./apache-arrow-apt-source-latest.deb && \
+    apt-get update && apt-get install -y libarrow-dev=6.0.1-1
+    
 RUN apt-get install -y sudo
 
 # Add graphscope user with user id 1001
@@ -49,8 +52,9 @@ USER graphscope
 WORKDIR /home/graphscope
 
 RUN curl -sf -L https://static.rust-lang.org/rustup.sh | \
-  sh -s -- -y --profile minimal && \
+  sh -s -- -y --profile minimal --default-toolchain=1.70.0 && \
   chmod +x "$HOME/.cargo/env" && \
   echo "$source $HOME/.cargo/env" >> ~/.bashrc && \
   source "$HOME/.cargo/env" && \
+  echo "1.70.0" > rust-toolchain && \
   bash -c "rustup component add rustfmt"
diff --git a/flex/cmake/FindArrow.cmake b/flex/cmake/FindArrow.cmake
new file mode 100644
index 000000000000..b26ca9adf752
--- /dev/null
+++ b/flex/cmake/FindArrow.cmake
@@ -0,0 +1,438 @@
+# The file cmake/FindArrow.cmake is referered from project
+# https://github.com/apache/arrow
+#
+#   https://github.com/apache/arrow/blob/master/cpp/cmake_modules/FindArrow.cmake
+#
+# which has the following license:
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# - Find Arrow (arrow/api.h, libarrow.a, libarrow.so)
+# This module defines
+#  ARROW_FOUND, whether Arrow has been found
+#  ARROW_FULL_SO_VERSION, full shared object version of found Arrow "100.0.0"
+#  ARROW_IMPORT_LIB, path to libarrow's import library (Windows only)
+#  ARROW_INCLUDE_DIR, directory containing headers
+#  ARROW_LIBS, deprecated. Use ARROW_LIB_DIR instead
+#  ARROW_LIB_DIR, directory containing Arrow libraries
+#  ARROW_SHARED_IMP_LIB, deprecated. Use ARROW_IMPORT_LIB instead
+#  ARROW_SHARED_LIB, path to libarrow's shared library
+#  ARROW_SO_VERSION, shared object version of found Arrow such as "100"
+#  ARROW_STATIC_LIB, path to libarrow.a
+#  ARROW_VERSION, version of found Arrow
+#  ARROW_VERSION_MAJOR, major version of found Arrow
+#  ARROW_VERSION_MINOR, minor version of found Arrow
+#  ARROW_VERSION_PATCH, patch version of found Arrow
+
+if(DEFINED ARROW_FOUND)
+  return()
+endif()
+
+include(FindPkgConfig)
+include(FindPackageHandleStandardArgs)
+
+set(ARROW_SEARCH_LIB_PATH_SUFFIXES)
+if(CMAKE_LIBRARY_ARCHITECTURE)
+  list(APPEND ARROW_SEARCH_LIB_PATH_SUFFIXES "lib/${CMAKE_LIBRARY_ARCHITECTURE}")
+endif()
+list(APPEND ARROW_SEARCH_LIB_PATH_SUFFIXES
+            "lib64"
+            "lib32"
+            "lib"
+            "bin")
+set(ARROW_CONFIG_SUFFIXES
+    "_RELEASE"
+    "_RELWITHDEBINFO"
+    "_MINSIZEREL"
+    "_DEBUG"
+    "")
+if(CMAKE_BUILD_TYPE)
+  string(TOUPPER ${CMAKE_BUILD_TYPE} ARROW_CONFIG_SUFFIX_PREFERRED)
+  set(ARROW_CONFIG_SUFFIX_PREFERRED "_${ARROW_CONFIG_SUFFIX_PREFERRED}")
+  list(INSERT ARROW_CONFIG_SUFFIXES 0 "${ARROW_CONFIG_SUFFIX_PREFERRED}")
+endif()
+
+if(NOT DEFINED ARROW_MSVC_STATIC_LIB_SUFFIX)
+  if(MSVC)
+    set(ARROW_MSVC_STATIC_LIB_SUFFIX "_static")
+  else()
+    set(ARROW_MSVC_STATIC_LIB_SUFFIX "")
+  endif()
+endif()
+
+# Internal function.
+#
+# Set shared library name for ${base_name} to ${output_variable}.
+#
+# Example:
+#   arrow_build_shared_library_name(ARROW_SHARED_LIBRARY_NAME arrow)
+#   # -> ARROW_SHARED_LIBRARY_NAME=libarrow.so on Linux
+#   # -> ARROW_SHARED_LIBRARY_NAME=libarrow.dylib on macOS
+#   # -> ARROW_SHARED_LIBRARY_NAME=arrow.dll with MSVC on Windows
+#   # -> ARROW_SHARED_LIBRARY_NAME=libarrow.dll with MinGW on Windows
+function(arrow_build_shared_library_name output_variable base_name)
+  set(${output_variable}
+      "${CMAKE_SHARED_LIBRARY_PREFIX}${base_name}${CMAKE_SHARED_LIBRARY_SUFFIX}"
+      PARENT_SCOPE)
+endfunction()
+
+# Internal function.
+#
+# Set import library name for ${base_name} to ${output_variable}.
+# This is useful only for MSVC build. Import library is used only
+# with MSVC build.
+#
+# Example:
+#   arrow_build_import_library_name(ARROW_IMPORT_LIBRARY_NAME arrow)
+#   # -> ARROW_IMPORT_LIBRARY_NAME=arrow on Linux (meaningless)
+#   # -> ARROW_IMPORT_LIBRARY_NAME=arrow on macOS (meaningless)
+#   # -> ARROW_IMPORT_LIBRARY_NAME=arrow.lib with MSVC on Windows
+#   # -> ARROW_IMPORT_LIBRARY_NAME=libarrow.dll.a with MinGW on Windows
+function(arrow_build_import_library_name output_variable base_name)
+  set(${output_variable}
+      "${CMAKE_IMPORT_LIBRARY_PREFIX}${base_name}${CMAKE_IMPORT_LIBRARY_SUFFIX}"
+      PARENT_SCOPE)
+endfunction()
+
+# Internal function.
+#
+# Set static library name for ${base_name} to ${output_variable}.
+#
+# Example:
+#   arrow_build_static_library_name(ARROW_STATIC_LIBRARY_NAME arrow)
+#   # -> ARROW_STATIC_LIBRARY_NAME=libarrow.a on Linux
+#   # -> ARROW_STATIC_LIBRARY_NAME=libarrow.a on macOS
+#   # -> ARROW_STATIC_LIBRARY_NAME=arrow.lib with MSVC on Windows
+#   # -> ARROW_STATIC_LIBRARY_NAME=libarrow.dll.a with MinGW on Windows
+function(arrow_build_static_library_name output_variable base_name)
+  set(
+    ${output_variable}
+    "${CMAKE_STATIC_LIBRARY_PREFIX}${base_name}${ARROW_MSVC_STATIC_LIB_SUFFIX}${CMAKE_STATIC_LIBRARY_SUFFIX}"
+    PARENT_SCOPE)
+endfunction()
+
+# Internal function.
+#
+# Set macro value for ${macro_name} in ${header_content} to ${output_variable}.
+#
+# Example:
+#   arrow_extract_macro_value(version_major
+#                             "ARROW_VERSION_MAJOR"
+#                             "#define ARROW_VERSION_MAJOR 1.0.0")
+#   # -> version_major=1.0.0
+function(arrow_extract_macro_value output_variable macro_name header_content)
+  string(REGEX MATCH "#define +${macro_name} +[^\r\n]+" macro_definition
+               "${header_content}")
+  string(REGEX
+         REPLACE "^#define +${macro_name} +(.+)$" "\\1" macro_value "${macro_definition}")
+  set(${output_variable} "${macro_value}" PARENT_SCOPE)
+endfunction()
+
+# Internal macro only for arrow_find_package.
+#
+# Find package in HOME.
+macro(arrow_find_package_home)
+  find_path(${prefix}_include_dir "${header_path}"
+            PATHS "${home}"
+            PATH_SUFFIXES "include"
+            NO_DEFAULT_PATH)
+  set(include_dir "${${prefix}_include_dir}")
+  set(${prefix}_INCLUDE_DIR "${include_dir}" PARENT_SCOPE)
+
+  if(MSVC)
+    set(CMAKE_SHARED_LIBRARY_SUFFIXES_ORIGINAL ${CMAKE_FIND_LIBRARY_SUFFIXES})
+    # .dll isn't found by find_library with MSVC because .dll isn't included in
+    # CMAKE_FIND_LIBRARY_SUFFIXES.
+    list(APPEND CMAKE_FIND_LIBRARY_SUFFIXES "${CMAKE_SHARED_LIBRARY_SUFFIX}")
+  endif()
+  find_library(${prefix}_shared_lib
+               NAMES "${shared_lib_name}"
+               PATHS "${home}"
+               PATH_SUFFIXES ${ARROW_SEARCH_LIB_PATH_SUFFIXES}
+               NO_DEFAULT_PATH)
+  if(MSVC)
+    set(CMAKE_SHARED_LIBRARY_SUFFIXES ${CMAKE_FIND_LIBRARY_SUFFIXES_ORIGINAL})
+  endif()
+  set(shared_lib "${${prefix}_shared_lib}")
+  set(${prefix}_SHARED_LIB "${shared_lib}" PARENT_SCOPE)
+  if(shared_lib)
+    add_library(${target_shared} SHARED IMPORTED)
+    set_target_properties(${target_shared} PROPERTIES IMPORTED_LOCATION "${shared_lib}")
+    if(include_dir)
+      set_target_properties(${target_shared}
+                            PROPERTIES INTERFACE_INCLUDE_DIRECTORIES "${include_dir}")
+    endif()
+    find_library(${prefix}_import_lib
+                 NAMES "${import_lib_name}"
+                 PATHS "${home}"
+                 PATH_SUFFIXES ${ARROW_SEARCH_LIB_PATH_SUFFIXES}
+                 NO_DEFAULT_PATH)
+    set(import_lib "${${prefix}_import_lib}")
+    set(${prefix}_IMPORT_LIB "${import_lib}" PARENT_SCOPE)
+    if(import_lib)
+      set_target_properties(${target_shared} PROPERTIES IMPORTED_IMPLIB "${import_lib}")
+    endif()
+  endif()
+
+  find_library(${prefix}_static_lib
+               NAMES "${static_lib_name}"
+               PATHS "${home}"
+               PATH_SUFFIXES ${ARROW_SEARCH_LIB_PATH_SUFFIXES}
+               NO_DEFAULT_PATH)
+  set(static_lib "${${prefix}_static_lib}")
+  set(${prefix}_STATIC_LIB "${static_lib}" PARENT_SCOPE)
+  if(static_lib)
+    add_library(${target_static} STATIC IMPORTED)
+    set_target_properties(${target_static} PROPERTIES IMPORTED_LOCATION "${static_lib}")
+    if(include_dir)
+      set_target_properties(${target_static}
+                            PROPERTIES INTERFACE_INCLUDE_DIRECTORIES "${include_dir}")
+    endif()
+  endif()
+endmacro()
+
+# Internal macro only for arrow_find_package.
+#
+# Find package by CMake package configuration.
+macro(arrow_find_package_cmake_package_configuration)
+  find_package(${cmake_package_name} CONFIG)
+  if(${cmake_package_name}_FOUND)
+    set(${prefix}_USE_CMAKE_PACKAGE_CONFIG TRUE PARENT_SCOPE)
+    if(TARGET ${target_shared})
+      foreach(suffix ${ARROW_CONFIG_SUFFIXES})
+        get_target_property(shared_lib ${target_shared} IMPORTED_LOCATION${suffix})
+        if(shared_lib)
+          # Remove shared library version:
+          #   libarrow.so.100.0.0 -> libarrow.so
+          # Because ARROW_HOME and pkg-config approaches don't add
+          # shared library version.
+          string(REGEX
+                 REPLACE "(${CMAKE_SHARED_LIBRARY_SUFFIX})[.0-9]+$" "\\1" shared_lib
+                         "${shared_lib}")
+          set(${prefix}_SHARED_LIB "${shared_lib}" PARENT_SCOPE)
+          break()
+        endif()
+      endforeach()
+    endif()
+    if(TARGET ${target_static})
+      foreach(suffix ${ARROW_CONFIG_SUFFIXES})
+        get_target_property(static_lib ${target_static} IMPORTED_LOCATION${suffix})
+        if(static_lib)
+          set(${prefix}_STATIC_LIB "${static_lib}" PARENT_SCOPE)
+          break()
+        endif()
+      endforeach()
+    endif()
+  endif()
+endmacro()
+
+# Internal macro only for arrow_find_package.
+#
+# Find package by pkg-config.
+macro(arrow_find_package_pkg_config)
+  pkg_check_modules(${prefix}_PC ${pkg_config_name})
+  if(${prefix}_PC_FOUND)
+    set(${prefix}_USE_PKG_CONFIG TRUE PARENT_SCOPE)
+
+    set(include_dir "${${prefix}_PC_INCLUDEDIR}")
+    set(lib_dir "${${prefix}_PC_LIBDIR}")
+    set(shared_lib_paths "${${prefix}_PC_LINK_LIBRARIES}")
+    # Use the first shared library path as the IMPORTED_LOCATION
+    # for ${target_shared}. This assumes that the first shared library
+    # path is the shared library path for this module.
+    list(GET shared_lib_paths 0 first_shared_lib_path)
+    # Use the rest shared library paths as the INTERFACE_LINK_LIBRARIES
+    # for ${target_shared}. This assumes that the rest shared library
+    # paths are dependency library paths for this module.
+    list(LENGTH shared_lib_paths n_shared_lib_paths)
+    if(n_shared_lib_paths LESS_EQUAL 1)
+      set(rest_shared_lib_paths)
+    else()
+      list(SUBLIST
+           shared_lib_paths
+           1
+           -1
+           rest_shared_lib_paths)
+    endif()
+
+    set(${prefix}_VERSION "${${prefix}_PC_VERSION}" PARENT_SCOPE)
+    set(${prefix}_INCLUDE_DIR "${include_dir}" PARENT_SCOPE)
+    set(${prefix}_SHARED_LIB "${first_shared_lib_path}" PARENT_SCOPE)
+
+    add_library(${target_shared} SHARED IMPORTED)
+    set_target_properties(${target_shared}
+                          PROPERTIES INTERFACE_INCLUDE_DIRECTORIES
+                                     "${include_dir}"
+                                     INTERFACE_LINK_LIBRARIES
+                                     "${rest_shared_lib_paths}"
+                                     IMPORTED_LOCATION
+                                     "${first_shared_lib_path}")
+    get_target_property(shared_lib ${target_shared} IMPORTED_LOCATION)
+
+    find_library(${prefix}_static_lib
+                 NAMES "${static_lib_name}"
+                 PATHS "${lib_dir}"
+                 NO_DEFAULT_PATH)
+    set(static_lib "${${prefix}_static_lib}")
+    set(${prefix}_STATIC_LIB "${static_lib}" PARENT_SCOPE)
+    if(static_lib)
+      add_library(${target_static} STATIC IMPORTED)
+      set_target_properties(${target_static}
+                            PROPERTIES INTERFACE_INCLUDE_DIRECTORIES "${include_dir}"
+                                       IMPORTED_LOCATION "${static_lib}")
+    endif()
+  endif()
+endmacro()
+
+function(arrow_find_package
+         prefix
+         home
+         base_name
+         header_path
+         cmake_package_name
+         pkg_config_name)
+  arrow_build_shared_library_name(shared_lib_name ${base_name})
+  arrow_build_import_library_name(import_lib_name ${base_name})
+  arrow_build_static_library_name(static_lib_name ${base_name})
+
+  set(target_shared ${base_name}_shared)
+  set(target_static ${base_name}_static)
+
+  if(home)
+    arrow_find_package_home()
+    set(${prefix}_FIND_APPROACH "HOME: ${home}" PARENT_SCOPE)
+  else()
+    arrow_find_package_cmake_package_configuration()
+    if(${cmake_package_name}_FOUND)
+      set(${prefix}_FIND_APPROACH
+          "CMake package configuration: ${cmake_package_name}"
+          PARENT_SCOPE)
+    else()
+      arrow_find_package_pkg_config()
+      set(${prefix}_FIND_APPROACH "pkg-config: ${pkg_config_name}" PARENT_SCOPE)
+    endif()
+  endif()
+
+  if(NOT include_dir)
+    if(TARGET ${target_shared})
+      get_target_property(include_dir ${target_shared} INTERFACE_INCLUDE_DIRECTORIES)
+    elseif(TARGET ${target_static})
+      get_target_property(include_dir ${target_static} INTERFACE_INCLUDE_DIRECTORIES)
+    endif()
+  endif()
+  if(include_dir)
+    set(${prefix}_INCLUDE_DIR "${include_dir}" PARENT_SCOPE)
+  endif()
+
+  if(shared_lib)
+    get_filename_component(lib_dir "${shared_lib}" DIRECTORY)
+  elseif(static_lib)
+    get_filename_component(lib_dir "${static_lib}" DIRECTORY)
+  else()
+    set(lib_dir NOTFOUND)
+  endif()
+  set(${prefix}_LIB_DIR "${lib_dir}" PARENT_SCOPE)
+  # For backward compatibility
+  set(${prefix}_LIBS "${lib_dir}" PARENT_SCOPE)
+endfunction()
+
+if(NOT "$ENV{ARROW_HOME}" STREQUAL "")
+  file(TO_CMAKE_PATH "$ENV{ARROW_HOME}" ARROW_HOME)
+endif()
+arrow_find_package(ARROW
+                   "${ARROW_HOME}"
+                   arrow
+                   arrow/api.h
+                   Arrow
+                   arrow)
+
+if(ARROW_HOME)
+  if(ARROW_INCLUDE_DIR)
+    file(READ "${ARROW_INCLUDE_DIR}/arrow/util/config.h" ARROW_CONFIG_H_CONTENT)
+    arrow_extract_macro_value(ARROW_VERSION_MAJOR "ARROW_VERSION_MAJOR"
+                              "${ARROW_CONFIG_H_CONTENT}")
+    arrow_extract_macro_value(ARROW_VERSION_MINOR "ARROW_VERSION_MINOR"
+                              "${ARROW_CONFIG_H_CONTENT}")
+    arrow_extract_macro_value(ARROW_VERSION_PATCH "ARROW_VERSION_PATCH"
+                              "${ARROW_CONFIG_H_CONTENT}")
+    if("${ARROW_VERSION_MAJOR}" STREQUAL ""
+       OR "${ARROW_VERSION_MINOR}" STREQUAL ""
+       OR "${ARROW_VERSION_PATCH}" STREQUAL "")
+      set(ARROW_VERSION "0.0.0")
+    else()
+      set(ARROW_VERSION
+          "${ARROW_VERSION_MAJOR}.${ARROW_VERSION_MINOR}.${ARROW_VERSION_PATCH}")
+    endif()
+
+    arrow_extract_macro_value(ARROW_SO_VERSION_QUOTED "ARROW_SO_VERSION"
+                              "${ARROW_CONFIG_H_CONTENT}")
+    string(REGEX REPLACE "^\"(.+)\"$" "\\1" ARROW_SO_VERSION "${ARROW_SO_VERSION_QUOTED}")
+    arrow_extract_macro_value(ARROW_FULL_SO_VERSION_QUOTED "ARROW_FULL_SO_VERSION"
+                              "${ARROW_CONFIG_H_CONTENT}")
+    string(REGEX
+           REPLACE "^\"(.+)\"$" "\\1" ARROW_FULL_SO_VERSION
+                   "${ARROW_FULL_SO_VERSION_QUOTED}")
+  endif()
+else()
+  if(ARROW_USE_CMAKE_PACKAGE_CONFIG)
+    find_package(Arrow CONFIG)
+  elseif(ARROW_USE_PKG_CONFIG)
+    pkg_get_variable(ARROW_SO_VERSION arrow so_version)
+    pkg_get_variable(ARROW_FULL_SO_VERSION arrow full_so_version)
+  endif()
+endif()
+
+set(ARROW_ABI_VERSION ${ARROW_SO_VERSION})
+
+mark_as_advanced(ARROW_ABI_VERSION
+                 ARROW_CONFIG_SUFFIXES
+                 ARROW_FULL_SO_VERSION
+                 ARROW_IMPORT_LIB
+                 ARROW_INCLUDE_DIR
+                 ARROW_LIBS
+                 ARROW_LIB_DIR
+                 ARROW_SEARCH_LIB_PATH_SUFFIXES
+                 ARROW_SHARED_IMP_LIB
+                 ARROW_SHARED_LIB
+                 ARROW_SO_VERSION
+                 ARROW_STATIC_LIB
+                 ARROW_VERSION
+                 ARROW_VERSION_MAJOR
+                 ARROW_VERSION_MINOR
+                 ARROW_VERSION_PATCH)
+
+find_package_handle_standard_args(Arrow REQUIRED_VARS
+                                  # The first required variable is shown
+                                  # in the found message. So this list is
+                                  # not sorted alphabetically.
+                                  ARROW_INCLUDE_DIR
+                                  ARROW_LIB_DIR
+                                  ARROW_FULL_SO_VERSION
+                                  ARROW_SO_VERSION
+                                  VERSION_VAR
+                                  ARROW_VERSION)
+set(ARROW_FOUND ${Arrow_FOUND})
+
+if(Arrow_FOUND AND NOT Arrow_FIND_QUIETLY)
+  message(STATUS "Arrow version: ${ARROW_VERSION} (${ARROW_FIND_APPROACH})")
+  message(STATUS "Arrow SO and ABI version: ${ARROW_SO_VERSION}")
+  message(STATUS "Arrow full SO version: ${ARROW_FULL_SO_VERSION}")
+  message(STATUS "Found the Arrow core shared library: ${ARROW_SHARED_LIB}")
+  message(STATUS "Found the Arrow core import library: ${ARROW_IMPORT_LIB}")
+  message(STATUS "Found the Arrow core static library: ${ARROW_STATIC_LIB}")
+endif()
diff --git a/flex/engines/graph_db/database/graph_db.cc b/flex/engines/graph_db/database/graph_db.cc
index dd80d49f9ab0..0d0ce3681538 100644
--- a/flex/engines/graph_db/database/graph_db.cc
+++ b/flex/engines/graph_db/database/graph_db.cc
@@ -62,13 +62,17 @@ void GraphDB::Init(const Schema& schema, const LoadingConfig& load_config,
       LOG(INFO) << "Initializing graph db through bulk loading";
       {
         MutablePropertyFragment graph;
-        graph.Init(schema, load_config, thread_num);
+        auto loader = LoaderFactory::CreateFragmentLoader(schema, load_config,
+                                                          thread_num);
+        loader->LoadFragment(graph);
         graph.Serialize(data_dir_path.string());
       }
       graph_.Deserialize(data_dir_path.string());
     } else {
       LOG(INFO) << "Initializing empty graph db";
-      graph_.Init(schema, load_config, thread_num);
+      auto loader =
+          LoaderFactory::CreateFragmentLoader(schema, load_config, thread_num);
+      loader->LoadFragment(graph_);
       graph_.Serialize(data_dir_path.string());
     }
   } else {
diff --git a/flex/engines/graph_db/database/graph_db.h b/flex/engines/graph_db/database/graph_db.h
index 2029f5299208..a85bae4d99e3 100644
--- a/flex/engines/graph_db/database/graph_db.h
+++ b/flex/engines/graph_db/database/graph_db.h
@@ -30,6 +30,7 @@
 #include "flex/engines/graph_db/database/single_vertex_insert_transaction.h"
 #include "flex/engines/graph_db/database/update_transaction.h"
 #include "flex/engines/graph_db/database/version_manager.h"
+#include "flex/storages/rt_mutable_graph/loader/loader_factory.h"
 #include "flex/storages/rt_mutable_graph/loading_config.h"
 #include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
 
diff --git a/flex/engines/graph_db/grin/CMakeLists.txt b/flex/engines/graph_db/grin/CMakeLists.txt
index 3a066292b992..12b016724d28 100644
--- a/flex/engines/graph_db/grin/CMakeLists.txt
+++ b/flex/engines/graph_db/grin/CMakeLists.txt
@@ -33,6 +33,20 @@ if (GLOG_FOUND)
     set(CMAKE_REQUIRED_LIBRARIES "${CMAKE_REQUIRED_LIBRARIES} ${GLOG_LIBRARIES}")
 endif ()
 
+#find arrow----------------------------------------------------------------------
+include("../../../../flex/cmake/FindArrow.cmake")
+if (NOT ARROW_FOUND)
+    message(FATAL_ERROR "arrow not found, please install the arrow library")
+else ()
+    include_directories(SYSTEM ${ARROW_INCLUDE_DIRS})
+    if (TARGET arrow_shared)
+        set(ARROW_SHARED_LIB arrow_shared)
+    endif()
+    if (TARGET arrow_static)
+        set(ARROW_STATIC_LIB arrow_static)
+    endif()
+endif ()
+
 find_package(yaml-cpp REQUIRED)
 include_directories(SYSTEM ${yaml-cpp_INCLUDE_DIRS})
 
@@ -57,13 +71,18 @@ add_custom_target(grin_clformat
         COMMENT "Running clang-format."
         VERBATIM)
 
-file(GLOB SOURCES "src/*.cc" "src/topology/*.cc" "src/property/*.cc" "src/index/*.cc" "src/common/*.cc" "../../../utils/property/*.cc" "../../../storages/rt_mutable_graph/*.cc")
+file(GLOB SOURCES "src/*.cc" "src/topology/*.cc" "src/property/*.cc" "src/index/*.cc" "src/common/*.cc" "../../../utils/property/*.cc" "../../../utils/*.cc"
+                  "../../../storages/rt_mutable_graph/*.cc" "../../../storages/rt_mutable_graph/loader/*.cc")
 add_library(flex_grin SHARED ${SOURCES})
 target_link_libraries(flex_grin ${LIBGRAPELITE_LIBRARIES} ${GFLAGS_LIBRARIES} ${CMAKE_DL_LIBS}  ${YAML_CPP_LIBRARIES})
-
+if (ARROW_SHARED_LIB)
+        target_link_libraries(flex_grin ${ARROW_SHARED_LIB})
+else()
+        target_link_libraries(flex_grin ${ARROW_STATIC_LIB})
+endif()
 
 add_executable(run_grin_test test/test.c)
 
 target_include_directories(run_grin_test PRIVATE ${LIBGRAPELITE_INCLUDE_DIRS}/grape/analytical_apps fragment)
 target_link_libraries(run_grin_test flex_grin ${LIBGRAPELITE_LIBRARIES} ${GFLAGS_LIBRARIES} ${CMAKE_DL_LIBS})
-target_link_libraries(run_grin_test)
+
diff --git a/flex/engines/graph_db/grin/src/predefine.h b/flex/engines/graph_db/grin/src/predefine.h
index 4eb4020bf864..ad6df6d85555 100644
--- a/flex/engines/graph_db/grin/src/predefine.h
+++ b/flex/engines/graph_db/grin/src/predefine.h
@@ -1,6 +1,7 @@
 #include <vector>
 
 #include "grin/predefine.h"
+#include "storages/rt_mutable_graph/loader/loader_factory.h"
 #include "storages/rt_mutable_graph/loading_config.h"
 #include "storages/rt_mutable_graph/mutable_property_fragment.h"
 
diff --git a/flex/engines/graph_db/grin/src/topology/structure.cc b/flex/engines/graph_db/grin/src/topology/structure.cc
index f79b3cab40bf..5698fdc5291a 100644
--- a/flex/engines/graph_db/grin/src/topology/structure.cc
+++ b/flex/engines/graph_db/grin/src/topology/structure.cc
@@ -47,7 +47,9 @@ GRIN_GRAPH grin_get_graph_from_storage(const char* uri) {
       gs::LoadingConfig::ParseFromYaml(schema, bulk_load_config_path);
 
   GRIN_GRAPH_T* g = new GRIN_GRAPH_T();
-  g->g.Init(schema, loading_config);
+  auto loader =
+      gs::LoaderFactory::CreateFragmentLoader(schema, loading_config, 1);
+  loader->LoadFragment(g->g);
   init_cache(g);
   return g;
 }
diff --git a/flex/engines/hqps_db/core/operator/sink.h b/flex/engines/hqps_db/core/operator/sink.h
index 97f3c397c775..397b7cd6605e 100644
--- a/flex/engines/hqps_db/core/operator/sink.h
+++ b/flex/engines/hqps_db/core/operator/sink.h
@@ -119,7 +119,7 @@ template <typename T,
           typename std::enable_if<
               (std::is_same_v<T, std::string_view>)>::type* = nullptr>
 void template_set_value(common::Value* value, T v) {
-  value->set_str(v.data(), v.size());
+  value->mutable_str()->assign(v.data(), v.size());
 }
 
 template <typename T,
diff --git a/flex/engines/hqps_db/core/utils/hqps_utils.h b/flex/engines/hqps_db/core/utils/hqps_utils.h
index 894bb1e3ee99..0487533c9c48 100644
--- a/flex/engines/hqps_db/core/utils/hqps_utils.h
+++ b/flex/engines/hqps_db/core/utils/hqps_utils.h
@@ -31,6 +31,8 @@
 #include "flex/utils/property/column.h"
 #include "flex/utils/property/types.h"
 
+#include "arrow/api.h"
+
 namespace gs {
 
 // demangle a c++ variable's class name
@@ -44,35 +46,6 @@ std::string demangle(const T& t) {
   return ret;
 }
 
-// Get PropertyType from Cpp type
-template <typename T>
-struct CppTypeToPropertyType;
-
-template <>
-struct CppTypeToPropertyType<int32_t> {
-  static constexpr PropertyType value = PropertyType::kInt32;
-};
-
-template <>
-struct CppTypeToPropertyType<int64_t> {
-  static constexpr PropertyType value = PropertyType::kInt64;
-};
-
-template <>
-struct CppTypeToPropertyType<double> {
-  static constexpr PropertyType value = PropertyType::kDouble;
-};
-
-template <>
-struct CppTypeToPropertyType<std::string> {
-  static constexpr PropertyType value = PropertyType::kString;
-};
-
-template <>
-struct CppTypeToPropertyType<std::string_view> {
-  static constexpr PropertyType value = PropertyType::kString;
-};
-
 template <typename T>
 struct return_type;
 
diff --git a/flex/engines/hqps_db/structures/multi_edge_set/flat_edge_set.h b/flex/engines/hqps_db/structures/multi_edge_set/flat_edge_set.h
index 047f2f02e865..110e0b320bd9 100644
--- a/flex/engines/hqps_db/structures/multi_edge_set/flat_edge_set.h
+++ b/flex/engines/hqps_db/structures/multi_edge_set/flat_edge_set.h
@@ -648,112 +648,6 @@ class SingleLabelEdgeSet {
   Direction direction_;
 };
 
-// template <typename VID_T, typename LabelT>
-// class FlatEdgeSet<VID_T, LabelT, grape::EmptyType> {
-//  public:
-//   // TODO: use std::tuple<VID_T,VID_T> is enough
-//   using ele_tuple_t = std::tuple<VID_T, VID_T, grape::EmptyType>;
-//   using index_ele_tuple_t = std::tuple<size_t, ele_tuple_t>;
-//   using iterator = FlatEdgeSetIter<VID_T, grape::EmptyType>;
-//   using self_type_t = FlatEdgeSet<VID_T, LabelT, grape::EmptyType>;
-//   using flat_t = self_type_t;
-//   using data_tuple_t = ele_tuple_t;
-
-//   static constexpr bool is_multi_label = false;
-//   static constexpr bool is_collection = false;
-//   static constexpr bool is_edge_set = true;
-//   static constexpr bool is_multi_src = false;
-//   static constexpr bool is_multi_dst_label = false;
-
-//   FlatEdgeSet(std::vector<ele_tuple_t>&& vec,
-//               std::vector<std::array<LabelT, 3>>&& label_triplet,
-//               std::vector<LabelT>&& label_triplet_ind, Direction& dire)
-//       : vec_(std::move(vec)),
-//         label_triplet_(std::move(label_triplet)),
-//         label_triplet_ind_(std::move(label_triplet_ind)),
-//         direction_(dire) {
-//     CHECK(label_triplet_ind_.size() == vec_.size());
-//   }
-
-//   iterator begin() const { return iterator(vec_, 0); }
-
-//   iterator end() const { return iterator(vec_, vec_.size()); }
-
-//   template <size_t col_ind, typename... index_ele_tuple_t_>
-//   flat_t Flat(
-//       std::vector<std::tuple<index_ele_tuple_t_...>>& index_ele_tuple)
-//       const
-//       {
-//     std::vector<std::tuple<VID_T, VID_T, grape::EmptyType>> res;
-//     std::vector<int32_t> label_triplet_ind;
-//     res.reserve(index_ele_tuple.size());
-//     label_triplet_ind.reserve(index_ele_tuple.size());
-//     for (auto i = 0; i < index_ele_tuple.size(); ++i) {
-//       auto cur_ind_ele = std::get<col_ind>(index_ele_tuple[i]);
-//       res.push_back(std::get<1>(cur_ind_ele));
-//       label_triplet_ind.push_back(label_triplet_ind_[std::get<0>(cur_ind_ele)]);
-//     }
-//     return FlatEdgeSet(std::move(res), label_triplet_,
-//     std::move(label_triplet_ind));
-//   }
-//   template <typename... PropT>
-//   void fillBuiltinProps(std::vector<std::tuple<PropT...>>& tuples,
-//                         PropNameArray<PropT...>& prop_names,
-//                         std::vector<size_t>& repeat_array) {
-//     fillBuiltinPropsImpl(tuples, prop_names, repeat_array,
-//                          std::make_index_sequence<sizeof...(PropT)>());
-//   }
-
-//   // fill builtin props withour repeat array.
-//   template <typename... PropT>
-//   void fillBuiltinProps(std::vector<std::tuple<PropT...>>& tuples,
-//                         PropNameArray<PropT...>& prop_names) {
-//     std::vector<size_t> repeat_array(vec_.size(), 1);
-//     fillBuiltinPropsImpl(tuples, prop_names, repeat_array,
-//                          std::make_index_sequence<sizeof...(PropT)>());
-//   }
-
-//   template <typename EXPR, size_t num_labels>
-//   std::pair<RowVertexSet<LabelT, VID_T, grape::EmptyType>,
-//   std::vector<size_t>> GetVertices(VOpt v_opt, std::array<LabelT,
-//   num_labels>& labels,
-//               EXPR& expr) const {
-//     // We only contains one label for dst vertices.
-//     CHECK(v_opt == VOpt::End);
-//     std::vector<offset_t> offsets;
-//     std::vector<VID_T> vids;
-//     offsets.reserve(Size());
-//     offsets.emplace_back(0);
-//     // TODO: check labels.
-//     bool flag = false;
-//     for (auto l : labels) {
-//       if (l == dst_label_) {
-//         flag = true;
-//       }
-//     }
-//     if (flag) {
-//       for (auto iter : *this) {
-//         vids.emplace_back(iter.GetDst());
-//         offsets.emplace_back(vids.size());
-//       }
-//     } else {
-//       size_t size = Size();
-//       for (auto i = 0; i < size; ++i) {
-//         offsets.emplace_back(0);
-//       }
-//     }
-//     auto set = make_default_row_vertex_set(std::move(vids), dst_label_);
-//     return std::make_pair(std::move(set), std::move(offsets));
-//   }
-
-//   size_t Size() const { return vec_.size(); }
-
-//  private:
-//   std::vector<ele_tuple_t> vec_;
-//   std::vector<std::array<label_t, 3>> label_triplet_;
-//   std::vector<LabelT> label_triplet_ind_;
-//   Direction direction_;
-// };
 }  // namespace gs
 
 #endif  // ENGINES_HQPS_ENGINE_DS_MULTI_EDGE_SET_FLAT_EDGE_SET_H_
\ No newline at end of file
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 4e586b0287a5..64ff2f1501c3 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
@@ -23,6 +23,7 @@
 
 #include "flex/engines/hqps_db/core/utils/hqps_utils.h"
 #include "flex/engines/hqps_db/structures/multi_vertex_set/general_vertex_set.h"
+#include "flex/utils/arrow_utils.h"
 #include "grape/grape.h"
 
 namespace gs {
diff --git a/flex/interactive/docker/interactive-runtime.Dockerfile b/flex/interactive/docker/interactive-runtime.Dockerfile
index e5a480644a90..1b8ee02de93e 100755
--- a/flex/interactive/docker/interactive-runtime.Dockerfile
+++ b/flex/interactive/docker/interactive-runtime.Dockerfile
@@ -1,4 +1,4 @@
-FROM registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.4
+FROM registry.cn-hongkong.aliyuncs.com/graphscope/hqps-server-base:v0.0.6
 ARG CI=false
 
 # change bash as default
diff --git a/flex/resources/queries/ic/adhoc/ic6_adhoc.cypher b/flex/resources/queries/ic/adhoc/ic6_adhoc.cypher
index a5598d9e7ea9..be628ae543e5 100644
--- a/flex/resources/queries/ic/adhoc/ic6_adhoc.cypher
+++ b/flex/resources/queries/ic/adhoc/ic6_adhoc.cypher
@@ -1,3 +1,3 @@
-MATCH (p_:PERSON {id: 6597069812321})-[:KNOWS*1..3]-(other:PERSON)<-[:HASCREATOR]-(p:POST)-[:HASTAG]->(t:TAG {name: "William_Wordsworth"}),
+MATCH (p_:PERSON {id: 30786325579101})-[:KNOWS*1..3]-(other:PERSON)<-[:HASCREATOR]-(p:POST)-[:HASTAG]->(t:TAG {name: "Shakira"}),
 (p:POST)-[:HASTAG]->(otherTag:TAG) WHERE otherTag <> t RETURN otherTag.name as name, count(distinct p) as postCnt 
-ORDER BY postCnt desc, name asc LIMIT 10
\ No newline at end of file
+ORDER BY postCnt desc, name asc LIMIT 10;
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/CMakeLists.txt b/flex/storages/rt_mutable_graph/CMakeLists.txt
index ecd664f540b8..35ed3b1379fe 100644
--- a/flex/storages/rt_mutable_graph/CMakeLists.txt
+++ b/flex/storages/rt_mutable_graph/CMakeLists.txt
@@ -7,6 +7,12 @@ file(GLOB_RECURSE RT_MUTABLE_GRAPH_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
 add_library(flex_rt_mutable_graph SHARED ${RT_MUTABLE_GRAPH_SRC_FILES})
 target_link_libraries(flex_rt_mutable_graph ${LIBGRAPELITE_LIBRARIES} ${YAML_CPP_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
 
+if (ARROW_SHARED_LIB)
+        target_link_libraries(flex_rt_mutable_graph ${ARROW_SHARED_LIB})
+else()
+        target_link_libraries(flex_rt_mutable_graph ${ARROW_STATIC_LIB})
+endif()
+
 install(TARGETS flex_rt_mutable_graph
         RUNTIME DESTINATION bin
         ARCHIVE DESTINATION lib
diff --git a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc
new file mode 100644
index 000000000000..4da485d490d3
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.cc
@@ -0,0 +1,95 @@
+
+/** 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 "flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h"
+
+namespace gs {
+
+BasicFragmentLoader::BasicFragmentLoader(const Schema& schema)
+    : schema_(schema),
+      vertex_label_num_(schema_.vertex_label_num()),
+      edge_label_num_(schema_.edge_label_num()) {
+  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);
+  lf_indexers_.resize(vertex_label_num_);
+
+  init_vertex_data();
+}
+
+void BasicFragmentLoader::init_vertex_data() {
+  for (label_t v_label = 0; v_label < vertex_label_num_; v_label++) {
+    auto& v_data = vertex_data_[v_label];
+    auto label_name = schema_.get_vertex_label_name(v_label);
+    auto& property_types = schema_.get_vertex_properties(v_label);
+    auto& property_names = schema_.get_vertex_property_names(v_label);
+    v_data.init(property_names, property_types,
+                schema_.get_vertex_storage_strategies(label_name),
+                schema_.get_max_vnum(label_name));
+  }
+  VLOG(10) << "Finish init vertex data";
+}
+
+void BasicFragmentLoader::LoadFragment(MutablePropertyFragment& res_fragment) {
+  CHECK(res_fragment.ie_.empty()) << "Fragment is not empty";
+  CHECK(res_fragment.oe_.empty()) << "Fragment is not empty";
+  CHECK(res_fragment.vertex_data_.empty()) << "Fragment is not empty";
+
+  res_fragment.schema_ = schema_;
+  res_fragment.vertex_label_num_ = vertex_label_num_;
+  res_fragment.edge_label_num_ = edge_label_num_;
+  res_fragment.ie_.swap(ie_);
+  res_fragment.oe_.swap(oe_);
+  res_fragment.vertex_data_.swap(vertex_data_);
+  res_fragment.lf_indexers_.swap(lf_indexers_);
+  VLOG(10) << "Finish Building Fragment, " << res_fragment.vertex_label_num_
+           << " vertices labels, " << res_fragment.edge_label_num_
+           << " edges labels";
+}
+
+void BasicFragmentLoader::AddVertexBatch(
+    label_t v_label, const std::vector<vid_t>& vids,
+    const std::vector<std::vector<Any>>& props) {
+  auto& table = vertex_data_[v_label];
+  CHECK(props.size() == table.col_num());
+  for (auto i = 0; i < props.size(); ++i) {
+    CHECK(props[i].size() == vids.size())
+        << "vids size: " << vids.size() << ", props size: " << props.size()
+        << ", props[i] size: " << props[i].size();
+  }
+  auto dst_columns = table.column_ptrs();
+  for (auto j = 0; j < props.size(); ++j) {
+    auto& cur_vec = props[j];
+    for (auto i = 0; i < vids.size(); ++i) {
+      auto index = vids[i];
+      dst_columns[j]->set_any(index, cur_vec[i]);
+    }
+  }
+}
+
+void BasicFragmentLoader::FinishAddingVertex(
+    label_t v_label, const IdIndexer<oid_t, vid_t>& indexer) {
+  CHECK(v_label < vertex_label_num_);
+  build_lf_indexer(indexer, lf_indexers_[v_label]);
+}
+
+const LFIndexer<vid_t>& BasicFragmentLoader::GetLFIndexer(
+    label_t v_label) const {
+  CHECK(v_label < vertex_label_num_);
+  return lf_indexers_[v_label];
+}
+
+}  // namespace gs
diff --git a/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
new file mode 100644
index 000000000000..85fed7cd2c24
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h
@@ -0,0 +1,135 @@
+/** 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 STORAGES_RT_MUTABLE_GRAPH_LOADER_BASIC_FRAGMENT_LOADER_H_
+#define STORAGES_RT_MUTABLE_GRAPH_LOADER_BASIC_FRAGMENT_LOADER_H_
+
+#include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
+#include "flex/storages/rt_mutable_graph/schema.h"
+
+namespace gs {
+
+template <typename EDATA_T>
+TypedMutableCsrBase<EDATA_T>* create_typed_csr(EdgeStrategy es) {
+  if (es == EdgeStrategy::kSingle) {
+    return new SingleMutableCsr<EDATA_T>();
+  } else if (es == EdgeStrategy::kMultiple) {
+    return new MutableCsr<EDATA_T>();
+  } else if (es == EdgeStrategy::kNone) {
+    return new EmptyCsr<EDATA_T>();
+  }
+  LOG(FATAL) << "not support edge strategy or edge data type";
+}
+
+// FragmentLoader should use this BasicFragmentLoader to construct
+// mutable_csr_fragment.
+class BasicFragmentLoader {
+ public:
+  BasicFragmentLoader(const Schema& schema);
+
+  void LoadFragment(MutablePropertyFragment& res_fragment);
+
+  // props vector is column_num X batch_size
+  void AddVertexBatch(label_t v_label, const std::vector<vid_t>& vids,
+                      const std::vector<std::vector<Any>>& props);
+
+  inline void SetVertexProperty(label_t v_label, size_t col_ind, vid_t vid,
+                                Any&& prop) {
+    auto& table = vertex_data_[v_label];
+    auto dst_columns = table.column_ptrs();
+    CHECK(col_ind < dst_columns.size());
+    dst_columns[col_ind]->set_any(vid, prop);
+  }
+
+  void FinishAddingVertex(label_t v_label,
+                          const IdIndexer<oid_t, vid_t>& indexer);
+
+  template <typename EDATA_T>
+  void AddNoPropEdgeBatch(label_t src_label_id, label_t dst_label_id,
+                          label_t edge_label_id) {
+    size_t index = src_label_id * vertex_label_num_ * edge_label_num_ +
+                   dst_label_id * edge_label_num_ + edge_label_id;
+    CHECK(ie_[index] == NULL);
+    CHECK(oe_[index] == NULL);
+    auto src_label_name = schema_.get_vertex_label_name(src_label_id);
+    auto dst_label_name = schema_.get_vertex_label_name(dst_label_id);
+    auto edge_label_name = schema_.get_edge_label_name(edge_label_id);
+    EdgeStrategy oe_strategy = schema_.get_outgoing_edge_strategy(
+        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(0, {});
+    oe_[index]->batch_init(0, {});
+  }
+
+  template <typename EDATA_T>
+  void PutEdges(label_t src_label_id, label_t dst_label_id,
+                label_t edge_label_id,
+                const std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& edges,
+                const std::vector<int32_t>& ie_degree,
+                const std::vector<int32_t>& oe_degree) {
+    size_t index = src_label_id * vertex_label_num_ * edge_label_num_ +
+                   dst_label_id * edge_label_num_ + edge_label_id;
+    auto& src_indexer = lf_indexers_[src_label_id];
+    auto& dst_indexer = lf_indexers_[dst_label_id];
+    CHECK(ie_[index] == NULL);
+    CHECK(oe_[index] == NULL);
+    auto src_label_name = schema_.get_vertex_label_name(src_label_id);
+    auto dst_label_name = schema_.get_vertex_label_name(dst_label_id);
+    auto edge_label_name = schema_.get_edge_label_name(edge_label_id);
+    EdgeStrategy oe_strategy = schema_.get_outgoing_edge_strategy(
+        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);
+    CHECK(ie_degree.size() == dst_indexer.size());
+    CHECK(oe_degree.size() == src_indexer.size());
+
+    ie_csr->batch_init(dst_indexer.size(), ie_degree);
+    oe_csr->batch_init(src_indexer.size(), oe_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),
+                             std::get<2>(edge));
+    }
+    ie_[index] = ie_csr;
+    oe_[index] = oe_csr;
+    VLOG(10) << "Finish adding edge batch of size: " << edges.size();
+  }
+
+  Table& GetVertexTable(size_t ind) {
+    CHECK(ind < vertex_data_.size());
+    return vertex_data_[ind];
+  }
+
+  // get lf_indexer
+  const LFIndexer<vid_t>& GetLFIndexer(label_t v_label) const;
+
+ private:
+  void init_vertex_data();
+  const Schema& schema_;
+  size_t vertex_label_num_, edge_label_num_;
+  std::vector<LFIndexer<vid_t>> lf_indexers_;
+  std::vector<MutableCsrBase*> ie_, oe_;
+  std::vector<Table> vertex_data_;
+};
+}  // namespace gs
+
+#endif  // STORAGES_RT_MUTABLE_GRAPH_LOADER_BASIC_FRAGMENT_LOADER_H_
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.cc b/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.cc
new file mode 100644
index 000000000000..65606d9aec2c
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.cc
@@ -0,0 +1,1364 @@
+/** 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 "flex/storages/rt_mutable_graph/loader/csv_fragment_loader.h"
+#include "flex/engines/hqps_db/core/utils/hqps_utils.h"
+
+namespace gs {
+
+static void preprocess_line(char* line) {
+  size_t len = strlen(line);
+  while (len >= 0) {
+    if (line[len] != '\0' && line[len] != '\n' && line[len] != '\r' &&
+        line[len] != ' ' && line[len] != '\t') {
+      break;
+    } else {
+      --len;
+    }
+  }
+  line[len + 1] = '\0';
+}
+
+static std::vector<std::string> read_header(const std::string& file_name,
+                                            char delimiter) {
+  char line_buf[4096];
+  FILE* fin = fopen(file_name.c_str(), "r");
+  if (fgets(line_buf, 4096, fin) == NULL) {
+    LOG(FATAL) << "Failed to read header from file: " << file_name;
+  }
+  preprocess_line(line_buf);
+  const char* cur = line_buf;
+  std::vector<std::string> res_vec;
+  while (*cur != '\0') {
+    const char* tmp = cur;
+    while (*tmp != '\0' && *tmp != delimiter) {
+      ++tmp;
+    }
+
+    std::string_view sv(cur, tmp - cur);
+    res_vec.emplace_back(sv);
+    cur = tmp + 1;
+  }
+  return res_vec;
+}
+
+static void put_delimiter_option(const LoadingConfig& loading_config,
+                                 arrow::csv::ParseOptions& parse_options) {
+  auto delimiter_str = loading_config.GetDelimiter();
+  if (delimiter_str.size() != 1) {
+    LOG(FATAL) << "Delimiter should be a single character";
+  }
+  parse_options.delimiter = delimiter_str[0];
+}
+
+static bool put_skip_rows_option(const LoadingConfig& loading_config,
+                                 arrow::csv::ReadOptions& read_options) {
+  bool header_row = loading_config.GetHasHeaderRow();
+  if (header_row) {
+    read_options.skip_rows = 1;
+  } else {
+    read_options.skip_rows = 0;
+  }
+  return header_row;
+}
+
+static void put_escape_char_option(const LoadingConfig& loading_config,
+                                   arrow::csv::ParseOptions& parse_options) {
+  auto escape_str = loading_config.GetEscapeChar();
+  if (escape_str.size() != 1) {
+    LOG(FATAL) << "Escape char should be a single character";
+  }
+  parse_options.escape_char = escape_str[0];
+  parse_options.escaping = loading_config.GetIsEscaping();
+}
+
+static void put_block_size_option(const LoadingConfig& loading_config,
+                                  arrow::csv::ReadOptions& read_options) {
+  auto batch_size = loading_config.GetBatchSize();
+  if (batch_size <= 0) {
+    LOG(FATAL) << "Block size should be positive";
+  }
+  read_options.block_size = batch_size;
+}
+
+static void put_quote_char_option(const LoadingConfig& loading_config,
+                                  arrow::csv::ParseOptions& parse_options) {
+  auto quoting_str = loading_config.GetQuotingChar();
+  if (quoting_str.size() != 1) {
+    LOG(FATAL) << "Quote char should be a single character";
+  }
+  parse_options.quote_char = quoting_str[0];
+  parse_options.quoting = loading_config.GetIsQuoting();
+  parse_options.double_quote = loading_config.GetIsDoubleQuoting();
+}
+
+static void put_column_names_option(const LoadingConfig& loading_config,
+                                    bool header_row,
+                                    const std::string& file_path,
+                                    char delimiter,
+                                    arrow::csv::ReadOptions& read_options) {
+  std::vector<std::string> all_column_names;
+  if (header_row) {
+    all_column_names = read_header(file_path, delimiter);
+    // It is possible that there exists duplicate column names in the header,
+    // transform them to unique names
+    std::unordered_map<std::string, int> name_count;
+    for (auto& name : all_column_names) {
+      if (name_count.find(name) == name_count.end()) {
+        name_count[name] = 1;
+      } else {
+        name_count[name]++;
+      }
+    }
+    VLOG(10) << "before Got all column names: " << all_column_names.size()
+             << gs::to_string(all_column_names);
+    for (auto i = 0; i < all_column_names.size(); ++i) {
+      auto& name = all_column_names[i];
+      if (name_count[name] > 1) {
+        auto cur_cnt = name_count[name];
+        name_count[name] -= 1;
+        all_column_names[i] = name + "_" + std::to_string(cur_cnt);
+      }
+    }
+    VLOG(10) << "Got all column names: " << all_column_names.size()
+             << gs::to_string(all_column_names);
+  } else {
+    // just get the number of columns.
+    size_t num_cols = 0;
+    {
+      auto tmp = read_header(file_path, delimiter);
+      num_cols = tmp.size();
+    }
+    all_column_names.resize(num_cols);
+    for (auto i = 0; i < all_column_names.size(); ++i) {
+      all_column_names[i] = std::string("f") + std::to_string(i);
+    }
+  }
+  read_options.column_names = all_column_names;
+  VLOG(10) << "Got all column names: " << all_column_names.size()
+           << gs::to_string(all_column_names);
+}
+
+static void check_edge_invariant(
+    const Schema& schema,
+    const std::vector<std::tuple<size_t, std::string, std::string>>&
+        column_mappings,
+    size_t src_col_ind, size_t dst_col_ind, label_t src_label_i,
+    label_t dst_label_i, label_t edge_label_i) {
+  // TODO(zhanglei): Check column mappings after multiple property on edge is
+  // supported
+  if (column_mappings.size() > 1) {
+    LOG(FATAL) << "Edge column mapping must be less than 1";
+  }
+  if (column_mappings.size() > 0) {
+    auto& mapping = column_mappings[0];
+    if (std::get<0>(mapping) == src_col_ind ||
+        std::get<0>(mapping) == dst_col_ind) {
+      LOG(FATAL) << "Edge column mappings must not contain src_col_ind or "
+                    "dst_col_ind";
+    }
+    auto src_label_name = schema.get_vertex_label_name(src_label_i);
+    auto dst_label_name = schema.get_vertex_label_name(dst_label_i);
+    auto edge_label_name = schema.get_edge_label_name(edge_label_i);
+    // check property exists in schema
+    if (!schema.edge_has_property(src_label_name, dst_label_name,
+                                  edge_label_name, std::get<2>(mapping))) {
+      LOG(FATAL) << "property " << std::get<2>(mapping)
+                 << " not exists in schema for edge triplet " << src_label_name
+                 << " -> " << edge_label_name << " -> " << dst_label_name;
+    }
+  }
+}
+
+static void set_vertex_properties(gs::ColumnBase* col,
+                                  std::shared_ptr<arrow::ChunkedArray> array,
+                                  const std::vector<vid_t>& vids) {
+  auto type = array->type();
+  size_t cur_ind = 0;
+  if (type == arrow::int64()) {
+    for (auto j = 0; j < array->num_chunks(); ++j) {
+      auto casted =
+          std::static_pointer_cast<arrow::Int64Array>(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))));
+      }
+    }
+  } else if (type == arrow::int32()) {
+    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))));
+      }
+    }
+  } else if (type == arrow::float64()) {
+    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))));
+      }
+    }
+  } else if (type == 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)));
+      }
+      for (auto k = 0; k < std::min((int64_t) 10, casted->length()); ++k) {
+        VLOG(10) << "set vertex property: " << vids[k] << ", "
+                 << casted->GetString(k) << " "
+                 << col->get(vids[k]).to_string();
+      }
+    }
+  } else if (type == arrow::utf8()) {
+    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)));
+      }
+    }
+  } else {
+    LOG(FATAL) << "Not support type: " << type->ToString();
+  }
+}
+
+template <typename EDATA_T>
+static void append_edges(
+    std::shared_ptr<arrow::Int64Array> src_col,
+    std::shared_ptr<arrow::Int64Array> dst_col,
+    const LFIndexer<vid_t>& src_indexer, const LFIndexer<vid_t>& dst_indexer,
+    std::vector<std::shared_ptr<arrow::Array>>& edata_cols,
+    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());
+
+  auto old_size = parsed_edges.size();
+  parsed_edges.resize(old_size + src_col->length());
+  VLOG(10) << "resize parsed_edges from" << old_size << " to "
+           << parsed_edges.size();
+
+  auto src_col_thread = std::thread([&]() {
+    size_t cur_ind = old_size;
+    for (auto i = 0; i < src_col->length(); ++i) {
+      auto src_vid = src_indexer.get_index(src_col->Value(i));
+      std::get<0>(parsed_edges[cur_ind++]) = src_vid;
+      oe_degree[src_vid]++;
+    }
+  });
+  auto dst_col_thread = std::thread([&]() {
+    size_t cur_ind = old_size;
+    for (auto i = 0; i < dst_col->length(); ++i) {
+      auto dst_vid = dst_indexer.get_index(dst_col->Value(i));
+      std::get<1>(parsed_edges[cur_ind++]) = dst_vid;
+      ie_degree[dst_vid]++;
+    }
+  });
+  src_col_thread.join();
+  dst_col_thread.join();
+
+  // if EDATA_T is grape::EmptyType, no need to read columns
+  if constexpr (!std::is_same<EDATA_T, grape::EmptyType>::value) {
+    CHECK(edata_cols.size() == 1);
+    auto edata_col = edata_cols[0];
+    CHECK(src_col->length() == edata_col->length());
+    size_t cur_ind = old_size;
+    auto type = edata_col->type();
+    if (type != CppTypeToArrowType<EDATA_T>::TypeValue()) {
+      LOG(FATAL) << "Inconsistent data type, expect "
+                 << CppTypeToArrowType<EDATA_T>::TypeValue()->ToString()
+                 << ", but got " << type->ToString();
+    }
+
+    using arrow_array_type =
+        typename gs::CppTypeToArrowType<EDATA_T>::ArrayType;
+    // cast chunk to EDATA_T array
+    auto data = std::static_pointer_cast<arrow_array_type>(edata_col);
+    for (auto j = 0; j < edata_col->length(); ++j) {
+      if constexpr (std::is_same<arrow_array_type, arrow::StringArray>::value ||
+                    std::is_same<arrow_array_type,
+                                 arrow::LargeStringArray>::value) {
+        std::get<2>(parsed_edges[cur_ind++]) = data->GetString(j);
+      } else {
+        std::get<2>(parsed_edges[cur_ind++]) = data->Value(j);
+      }
+    }
+    VLOG(10) << "Finish inserting:  " << src_col->length() << " edges";
+  }
+}
+
+template <typename EDATA_T>
+static void append_edges(
+    std::shared_ptr<arrow::ChunkedArray> src_col,
+    std::shared_ptr<arrow::ChunkedArray> dst_col,
+    const LFIndexer<vid_t>& src_indexer, const LFIndexer<vid_t>& dst_indexer,
+    std::vector<std::shared_ptr<arrow::ChunkedArray>>& edata_cols,
+    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());
+  CHECK(src_col->type() == arrow::int64());
+  CHECK(dst_col->type() == arrow::int64());
+
+  auto old_size = parsed_edges.size();
+  parsed_edges.resize(old_size + src_col->length());
+  VLOG(10) << "resize parsed_edges from" << old_size << " to "
+           << parsed_edges.size();
+
+  auto src_col_thread = std::thread([&]() {
+    size_t cur_ind = old_size;
+    for (auto i = 0; i < src_col->num_chunks(); ++i) {
+      auto chunk = src_col->chunk(i);
+      CHECK(chunk->type() == arrow::int64());
+      auto casted_chunk = std::static_pointer_cast<arrow::Int64Array>(chunk);
+      for (auto j = 0; j < casted_chunk->length(); ++j) {
+        auto src_vid = src_indexer.get_index(casted_chunk->Value(j));
+        std::get<0>(parsed_edges[cur_ind++]) = src_vid;
+        oe_degree[src_vid]++;
+      }
+    }
+  });
+  auto dst_col_thread = std::thread([&]() {
+    size_t cur_ind = old_size;
+    for (auto i = 0; i < dst_col->num_chunks(); ++i) {
+      auto chunk = dst_col->chunk(i);
+      CHECK(chunk->type() == arrow::int64());
+      auto casted_chunk = std::static_pointer_cast<arrow::Int64Array>(chunk);
+      for (auto j = 0; j < casted_chunk->length(); ++j) {
+        auto dst_vid = dst_indexer.get_index(casted_chunk->Value(j));
+        std::get<1>(parsed_edges[cur_ind++]) = dst_vid;
+        ie_degree[dst_vid]++;
+      }
+    }
+  });
+
+  // if EDATA_T is grape::EmptyType, no need to read columns
+  auto edata_col_thread = std::thread([&]() {
+    if constexpr (!std::is_same<EDATA_T, grape::EmptyType>::value) {
+      CHECK(edata_cols.size() == 1);
+      auto edata_col = edata_cols[0];
+      CHECK(src_col->length() == edata_col->length());
+      // iterate and put data
+      size_t cur_ind = old_size;
+      auto type = edata_col->type();
+      if (type != CppTypeToArrowType<EDATA_T>::TypeValue()) {
+        LOG(FATAL) << "Inconsistent data type, expect "
+                   << CppTypeToArrowType<EDATA_T>::TypeValue()->ToString()
+                   << ", but got " << type->ToString();
+      }
+
+      using arrow_array_type =
+          typename gs::CppTypeToArrowType<EDATA_T>::ArrayType;
+      for (auto i = 0; i < edata_col->num_chunks(); ++i) {
+        auto chunk = edata_col->chunk(i);
+        auto casted_chunk = std::static_pointer_cast<arrow_array_type>(chunk);
+        for (auto j = 0; j < casted_chunk->length(); ++j) {
+          if constexpr (std::is_same<arrow_array_type,
+                                     arrow::StringArray>::value ||
+                        std::is_same<arrow_array_type,
+                                     arrow::LargeStringArray>::value) {
+            std::get<2>(parsed_edges[cur_ind++]) = casted_chunk->GetView(j);
+          } else {
+            std::get<2>(parsed_edges[cur_ind++]) = casted_chunk->Value(j);
+          }
+        }
+      }
+    }
+  });
+  src_col_thread.join();
+  dst_col_thread.join();
+  edata_col_thread.join();
+  VLOG(10) << "Finish inserting:  " << src_col->length() << " edges";
+}
+
+// Create VertexTableReader
+std::shared_ptr<arrow::csv::TableReader>
+CSVFragmentLoader::createVertexTableReader(label_t v_label,
+                                           const std::string& v_file) {
+  // Create options.
+  arrow::csv::ConvertOptions convert_options;
+  arrow::csv::ReadOptions read_options;
+  arrow::csv::ParseOptions parse_options;
+  fillVertexReaderMeta(read_options, parse_options, convert_options, v_file,
+                       v_label);
+
+  auto read_result = arrow::io::ReadableFile::Open(v_file);
+  if (!read_result.ok()) {
+    LOG(FATAL) << "Fail to open: " << v_file
+               << " error: " << read_result.status().message();
+  }
+  std::shared_ptr<arrow::io::ReadableFile> file = read_result.ValueOrDie();
+  auto res =
+      arrow::csv::TableReader::Make(arrow::io::IOContext(), file, read_options,
+                                    parse_options, convert_options);
+  if (!res.ok()) {
+    LOG(FATAL) << "Fail to create StreamingReader for file: " << v_file
+               << " error: " << res.status().message();
+  }
+  return res.ValueOrDie();
+}
+
+std::shared_ptr<arrow::csv::StreamingReader>
+CSVFragmentLoader::createVertexStreamReader(label_t v_label,
+                                            const std::string& v_file) {
+  arrow::csv::ConvertOptions convert_options;
+  arrow::csv::ReadOptions read_options;
+  arrow::csv::ParseOptions parse_options;
+  fillVertexReaderMeta(read_options, parse_options, convert_options, v_file,
+                       v_label);
+
+  auto read_result = arrow::io::ReadableFile::Open(v_file);
+  if (!read_result.ok()) {
+    LOG(FATAL) << "Fail to open: " << v_file
+               << " error: " << read_result.status().message();
+  }
+  std::shared_ptr<arrow::io::ReadableFile> file = read_result.ValueOrDie();
+  auto res = arrow::csv::StreamingReader::Make(arrow::io::IOContext(), file,
+                                               read_options, parse_options,
+                                               convert_options);
+  if (!res.ok()) {
+    LOG(FATAL) << "Fail to create StreamingReader for file: " << v_file
+               << " error: " << res.status().message();
+  }
+  return res.ValueOrDie();
+}
+
+std::shared_ptr<arrow::csv::StreamingReader>
+CSVFragmentLoader::createEdgeStreamReader(label_t src_label_id,
+                                          label_t dst_label_id,
+                                          label_t label_id,
+                                          const std::string& e_file) {
+  arrow::csv::ConvertOptions convert_options;
+  arrow::csv::ReadOptions read_options;
+  arrow::csv::ParseOptions parse_options;
+
+  fillEdgeReaderMeta(read_options, parse_options, convert_options, e_file,
+                     src_label_id, dst_label_id, label_id);
+
+  auto read_result = arrow::io::ReadableFile::Open(e_file);
+  if (!read_result.ok()) {
+    LOG(FATAL) << "Fail to open: " << e_file
+               << " error: " << read_result.status().message();
+  }
+  std::shared_ptr<arrow::io::ReadableFile> file = read_result.ValueOrDie();
+  auto res = arrow::csv::StreamingReader::Make(arrow::io::IOContext(), file,
+                                               read_options, parse_options,
+                                               convert_options);
+  if (!res.ok()) {
+    LOG(FATAL) << "Fail to create StreamingReader for file: " << e_file
+               << " error: " << res.status().message();
+  }
+  return res.ValueOrDie();
+}
+
+std::shared_ptr<arrow::csv::TableReader>
+CSVFragmentLoader::createEdgeTableReader(label_t src_label_id,
+                                         label_t dst_label_id, label_t label_id,
+                                         const std::string& e_file) {
+  arrow::csv::ConvertOptions convert_options;
+  arrow::csv::ReadOptions read_options;
+  arrow::csv::ParseOptions parse_options;
+
+  fillEdgeReaderMeta(read_options, parse_options, convert_options, e_file,
+                     src_label_id, dst_label_id, label_id);
+
+  auto read_result = arrow::io::ReadableFile::Open(e_file);
+  if (!read_result.ok()) {
+    LOG(FATAL) << "Fail to open: " << e_file
+               << " error: " << read_result.status().message();
+  }
+  std::shared_ptr<arrow::io::ReadableFile> file = read_result.ValueOrDie();
+  auto res =
+      arrow::csv::TableReader::Make(arrow::io::IOContext(), file, read_options,
+                                    parse_options, convert_options);
+  if (!res.ok()) {
+    LOG(FATAL) << "Fail to create TableReader for file: " << e_file
+               << " error: " << res.status().message();
+  }
+  return res.ValueOrDie();
+}
+
+void CSVFragmentLoader::addVertexBatch(
+    label_t v_label_id, IdIndexer<oid_t, vid_t>& indexer,
+    std::shared_ptr<arrow::Array>& primary_key_col,
+    const std::vector<std::shared_ptr<arrow::Array>>& property_cols) {
+  size_t row_num = primary_key_col->length();
+  CHECK_EQ(primary_key_col->type()->id(), arrow::Type::INT64);
+  auto col_num = property_cols.size();
+  for (size_t i = 0; i < col_num; ++i) {
+    CHECK_EQ(property_cols[i]->length(), row_num);
+  }
+  auto casted_array =
+      std::static_pointer_cast<arrow::Int64Array>(primary_key_col);
+  std::vector<std::vector<Any>> prop_vec(property_cols.size());
+
+  double t = -grape::GetCurrentTime();
+  vid_t vid;
+  std::vector<vid_t> vids;
+  vids.reserve(row_num);
+  for (auto i = 0; i < row_num; ++i) {
+    if (!indexer.add(casted_array->Value(i), vid)) {
+      LOG(FATAL) << "Duplicate vertex id: " << casted_array->Value(i) << " for "
+                 << schema_.get_vertex_label_name(v_label_id);
+    }
+    vids.emplace_back(vid);
+  }
+
+  t += grape::GetCurrentTime();
+  for (double tmp = convert_to_internal_vertex_time_;
+       !convert_to_internal_vertex_time_.compare_exchange_weak(tmp, tmp + t);) {
+  }
+
+  t = -grape::GetCurrentTime();
+  for (auto j = 0; j < property_cols.size(); ++j) {
+    auto array = property_cols[j];
+    auto chunked_array = std::make_shared<arrow::ChunkedArray>(array);
+    set_vertex_properties(
+        basic_fragment_loader_.GetVertexTable(v_label_id).column_ptrs()[j],
+        chunked_array, vids);
+  }
+
+  t += grape::GetCurrentTime();
+  for (double tmp = basic_frag_loader_vertex_time_;
+       !basic_frag_loader_vertex_time_.compare_exchange_weak(tmp, tmp + t);) {}
+
+  VLOG(10) << "Insert rows: " << row_num;
+}
+
+void CSVFragmentLoader::addVertexBatch(
+    label_t v_label_id, IdIndexer<oid_t, vid_t>& indexer,
+    std::shared_ptr<arrow::ChunkedArray>& primary_key_col,
+    const std::vector<std::shared_ptr<arrow::ChunkedArray>>& property_cols) {
+  size_t row_num = primary_key_col->length();
+  std::vector<vid_t> vids;
+  vids.reserve(row_num);
+  CHECK_EQ(primary_key_col->type()->id(), arrow::Type::INT64);
+  // check row num
+  auto col_num = property_cols.size();
+  for (size_t i = 0; i < col_num; ++i) {
+    CHECK_EQ(property_cols[i]->length(), row_num);
+  }
+  std::vector<std::vector<Any>> prop_vec(property_cols.size());
+
+  double t = -grape::GetCurrentTime();
+  for (auto i = 0; i < primary_key_col->num_chunks(); ++i) {
+    auto chunk = primary_key_col->chunk(i);
+    auto casted_array = std::static_pointer_cast<arrow::Int64Array>(chunk);
+    for (auto j = 0; j < casted_array->length(); ++j) {
+      vid_t vid;
+      if (!indexer.add(casted_array->Value(j), vid)) {
+        LOG(FATAL) << "Duplicate vertex id: " << casted_array->Value(j)
+                   << " for " << schema_.get_vertex_label_name(v_label_id);
+      }
+      vids.emplace_back(vid);
+    }
+  }
+
+  t += grape::GetCurrentTime();
+  for (double tmp = convert_to_internal_vertex_time_;
+       !convert_to_internal_vertex_time_.compare_exchange_weak(tmp, tmp + t);) {
+  }
+
+  t = -grape::GetCurrentTime();
+  for (auto i = 0; i < property_cols.size(); ++i) {
+    auto array = property_cols[i];
+    auto& table = basic_fragment_loader_.GetVertexTable(v_label_id);
+    auto& col_ptrs = table.column_ptrs();
+    set_vertex_properties(col_ptrs[i], array, vids);
+  }
+  t += grape::GetCurrentTime();
+  for (double tmp = basic_frag_loader_vertex_time_;
+       !basic_frag_loader_vertex_time_.compare_exchange_weak(tmp, tmp + t);) {}
+
+  VLOG(10) << "Insert rows: " << row_num;
+}
+
+void CSVFragmentLoader::addVerticesImplWithTableReader(
+    const std::string& v_file, label_t v_label_id,
+    IdIndexer<oid_t, vid_t>& indexer) {
+  auto vertex_column_mappings =
+      loading_config_.GetVertexColumnMappings(v_label_id);
+  auto primary_key = schema_.get_vertex_primary_key(v_label_id)[0];
+  size_t primary_key_ind = std::get<2>(primary_key);
+  auto reader = createVertexTableReader(v_label_id, v_file);
+  std::shared_ptr<arrow::Table> table;
+  double t = -grape::GetCurrentTime();
+  auto result = reader->Read();
+  t += grape::GetCurrentTime();
+  for (double tmp = read_vertex_table_time_;
+       !read_vertex_table_time_.compare_exchange_weak(tmp, tmp + t);) {}
+
+  auto status = result.status();
+  if (!status.ok()) {
+    LOG(FATAL) << "Failed to read next batch from file " << v_file
+               << status.message();
+  }
+  table = result.ValueOrDie();
+  if (table == nullptr) {
+    LOG(FATAL) << "Empty file: " << v_file;
+  }
+  auto header = table->schema()->field_names();
+  auto schema_column_names = schema_.get_vertex_property_names(v_label_id);
+  CHECK(schema_column_names.size() + 1 == header.size());
+  VLOG(10) << "Find header of size: " << header.size();
+
+  auto columns = table->columns();
+  CHECK(primary_key_ind < columns.size());
+  auto primary_key_column = columns[primary_key_ind];
+  auto other_columns_array = columns;
+  other_columns_array.erase(other_columns_array.begin() + primary_key_ind);
+  VLOG(10) << "Reading record batch of size: " << table->num_rows();
+  addVertexBatch(v_label_id, indexer, primary_key_column, other_columns_array);
+}
+
+void CSVFragmentLoader::addVerticesImplWithStreamReader(
+    const std::string& v_file, label_t v_label_id,
+    IdIndexer<oid_t, vid_t>& indexer) {
+  auto vertex_column_mappings =
+      loading_config_.GetVertexColumnMappings(v_label_id);
+  auto primary_key = schema_.get_vertex_primary_key(v_label_id)[0];
+  auto primary_key_name = std::get<1>(primary_key);
+  size_t primary_key_ind = std::get<2>(primary_key);
+  auto reader = createVertexStreamReader(v_label_id, v_file);
+  std::shared_ptr<arrow::RecordBatch> record_batch;
+  bool first_batch = true;
+  while (true) {
+    double t = -grape::GetCurrentTime();
+    auto status = reader->ReadNext(&record_batch);
+    t += grape::GetCurrentTime();
+    for (double tmp = read_vertex_table_time_;
+         !read_vertex_table_time_.compare_exchange_weak(tmp, tmp + t);) {}
+    if (!status.ok()) {
+      LOG(FATAL) << "Failed to read next batch from file " << v_file
+                 << status.message();
+    }
+    if (record_batch == nullptr) {
+      break;
+    }
+    if (first_batch) {
+      // get header
+      auto header = record_batch->schema()->field_names();
+      auto schema_column_names = schema_.get_vertex_property_names(v_label_id);
+      CHECK(schema_column_names.size() + 1 == header.size());
+      VLOG(10) << "Find header of size: " << header.size();
+      first_batch = false;
+    }
+
+    auto columns = record_batch->columns();
+    CHECK(primary_key_ind < columns.size());
+    auto primary_key_column = columns[primary_key_ind];
+    auto other_columns_array = columns;
+    other_columns_array.erase(other_columns_array.begin() + primary_key_ind);
+    VLOG(10) << "Reading record batch of size: " << record_batch->num_rows();
+    addVertexBatch(v_label_id, indexer, primary_key_column,
+                   other_columns_array);
+  }
+}
+
+void CSVFragmentLoader::addVerticesImpl(label_t v_label_id,
+                                        const std::string& v_label_name,
+                                        const std::vector<std::string> v_files,
+                                        IdIndexer<oid_t, vid_t>& indexer) {
+  VLOG(10) << "Parsing vertex file:" << v_files.size() << " for label "
+           << v_label_name;
+
+  for (auto& v_file : v_files) {
+    if (loading_config_.GetIsBatchReader()) {
+      addVerticesImplWithStreamReader(v_file, v_label_id, indexer);
+    } else {
+      addVerticesImplWithTableReader(v_file, v_label_id, indexer);
+    }
+  }
+
+  VLOG(10) << "Finish parsing vertex file:" << v_files.size() << " for label "
+           << v_label_name;
+}
+
+void CSVFragmentLoader::addVertices(label_t v_label_id,
+                                    const std::vector<std::string>& v_files) {
+  auto primary_keys = schema_.get_vertex_primary_key(v_label_id);
+
+  if (primary_keys.size() != 1) {
+    LOG(FATAL) << "Only support one primary key for vertex.";
+  }
+  if (std::get<0>(primary_keys[0]) != PropertyType::kInt64) {
+    LOG(FATAL) << "Only support int64_t primary key for vertex.";
+  }
+
+  std::string v_label_name = schema_.get_vertex_label_name(v_label_id);
+  VLOG(10) << "Start init vertices for label " << v_label_name << " with "
+           << v_files.size() << " files.";
+
+  IdIndexer<oid_t, vid_t> indexer;
+
+  addVerticesImpl(v_label_id, v_label_name, v_files, indexer);
+
+  if (indexer.bucket_count() == 0) {
+    indexer._rehash(schema_.get_max_vnum(v_label_name));
+  }
+  basic_fragment_loader_.FinishAddingVertex(v_label_id, indexer);
+
+  VLOG(10) << "Finish init vertices for label " << v_label_name;
+}
+
+template <typename EDATA_T>
+void CSVFragmentLoader::addEdgesImplWithTableReader(
+    const std::string& filename, label_t src_label_id, label_t dst_label_id,
+    label_t e_label_id, std::vector<int32_t>& ie_degree,
+    std::vector<int32_t>& oe_degree,
+    std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& parsed_edges) {
+  const auto& src_indexer = basic_fragment_loader_.GetLFIndexer(src_label_id);
+  const auto& dst_indexer = basic_fragment_loader_.GetLFIndexer(dst_label_id);
+  auto reader =
+      createEdgeTableReader(src_label_id, dst_label_id, e_label_id, filename);
+  std::shared_ptr<arrow::Table> table;
+  double t = -grape::GetCurrentTime();
+  auto result = reader->Read();
+  t += grape::GetCurrentTime();
+  for (double tmp = read_edge_table_time_;
+       !read_edge_table_time_.compare_exchange_weak(tmp, tmp + t);) {}
+
+  auto status = result.status();
+  if (!status.ok()) {
+    LOG(FATAL) << "Failed to read Table from file " << filename
+               << status.message();
+  }
+  table = result.ValueOrDie();
+  if (table == nullptr) {
+    LOG(FATAL) << "Empty file: " << filename;
+  }
+  auto header = table->schema()->field_names();
+  auto schema_column_names =
+      schema_.get_edge_property_names(src_label_id, dst_label_id, e_label_id);
+  auto schema_column_types =
+      schema_.get_edge_properties(src_label_id, dst_label_id, e_label_id);
+  CHECK(schema_column_names.size() + 2 == header.size());
+  CHECK(schema_column_types.size() + 2 == header.size());
+  VLOG(10) << "Find header of size: " << header.size();
+
+  auto columns = table->columns();
+  CHECK(columns.size() >= 2);
+  auto src_col = columns[0];
+  auto dst_col = columns[1];
+  CHECK(src_col->type() == arrow::int64())
+      << "src_col type: " << src_col->type()->ToString();
+  CHECK(dst_col->type() == arrow::int64())
+      << "dst_col type: " << dst_col->type()->ToString();
+
+  std::vector<std::shared_ptr<arrow::ChunkedArray>> property_cols;
+  for (auto i = 2; i < columns.size(); ++i) {
+    property_cols.emplace_back(columns[i]);
+  }
+  CHECK(property_cols.size() <= 1)
+      << "Currently only support at most one property on edge";
+  {
+    CHECK(src_col->length() == dst_col->length());
+    CHECK(src_col->type() == arrow::int64());
+    CHECK(dst_col->type() == arrow::int64());
+    t = -grape::GetCurrentTime();
+    append_edges(src_col, dst_col, src_indexer, dst_indexer, property_cols,
+                 parsed_edges, ie_degree, oe_degree);
+    t += grape::GetCurrentTime();
+    for (double tmp = convert_to_internal_edge_time_;
+         !convert_to_internal_edge_time_.compare_exchange_weak(tmp, tmp + t);) {
+    }
+  }
+}
+
+template <typename EDATA_T>
+void CSVFragmentLoader::addEdgesImplWithStreamReader(
+    const std::string& filename, label_t src_label_id, label_t dst_label_id,
+    label_t e_label_id, std::vector<int32_t>& ie_degree,
+    std::vector<int32_t>& oe_degree,
+    std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& parsed_edges) {
+  const auto& src_indexer = basic_fragment_loader_.GetLFIndexer(src_label_id);
+  const auto& dst_indexer = basic_fragment_loader_.GetLFIndexer(dst_label_id);
+  auto reader =
+      createEdgeStreamReader(src_label_id, dst_label_id, e_label_id, filename);
+  std::shared_ptr<arrow::RecordBatch> record_batch;
+  // read first batch
+  bool first_batch = true;
+  while (true) {
+    double t = -grape::GetCurrentTime();
+    auto status = reader->ReadNext(&record_batch);
+    t += grape::GetCurrentTime();
+    for (double tmp = read_edge_table_time_;
+         !read_edge_table_time_.compare_exchange_weak(tmp, tmp + t);) {}
+    if (!status.ok()) {
+      LOG(FATAL) << "Failed to read next batch from file " << filename
+                 << status.message();
+    }
+    if (record_batch == nullptr) {
+      break;
+    }
+    if (first_batch) {
+      auto header = record_batch->schema()->field_names();
+      auto schema_column_names = schema_.get_edge_property_names(
+          src_label_id, dst_label_id, e_label_id);
+      auto schema_column_types =
+          schema_.get_edge_properties(src_label_id, dst_label_id, e_label_id);
+      CHECK(schema_column_names.size() + 2 == header.size())
+          << "schema size: " << schema_column_names.size()
+          << " header size: " << header.size();
+      CHECK(schema_column_types.size() + 2 == header.size())
+          << "schema size: " << schema_column_types.size()
+          << " header size: " << header.size();
+      VLOG(10) << "Find header of size: " << header.size();
+      first_batch = false;
+    }
+
+    // copy the table to csr.
+    auto columns = record_batch->columns();
+    // We assume the src_col and dst_col will always be put at front.
+    CHECK(columns.size() >= 2);
+    auto src_col = columns[0];
+    auto dst_col = columns[1];
+    CHECK(src_col->type() == arrow::int64())
+        << "src_col type: " << src_col->type()->ToString();
+    CHECK(dst_col->type() == arrow::int64())
+        << "dst_col type: " << dst_col->type()->ToString();
+
+    std::vector<std::shared_ptr<arrow::Array>> property_cols;
+    for (auto i = 2; i < columns.size(); ++i) {
+      property_cols.emplace_back(columns[i]);
+    }
+    CHECK(property_cols.size() <= 1)
+        << "Currently only support at most one property on edge";
+    {
+      // add edges to vector
+      CHECK(src_col->length() == dst_col->length());
+      CHECK(src_col->type() == arrow::int64());
+      CHECK(dst_col->type() == arrow::int64());
+      auto src_casted_array =
+          std::static_pointer_cast<arrow::Int64Array>(src_col);
+      auto dst_casted_array =
+          std::static_pointer_cast<arrow::Int64Array>(dst_col);
+      t = -grape::GetCurrentTime();
+      append_edges(src_casted_array, dst_casted_array, src_indexer, dst_indexer,
+                   property_cols, parsed_edges, ie_degree, oe_degree);
+      t += grape::GetCurrentTime();
+      for (double tmp = convert_to_internal_edge_time_;
+           !convert_to_internal_edge_time_.compare_exchange_weak(tmp, tmp + t);
+           tmp = convert_to_internal_edge_time_) {}
+    }
+  }
+}
+
+template <typename EDATA_T>
+void CSVFragmentLoader::addEdgesImpl(label_t src_label_id, label_t dst_label_id,
+                                     label_t e_label_id,
+                                     const std::vector<std::string>& e_files) {
+  auto edge_column_mappings = loading_config_.GetEdgeColumnMappings(
+      src_label_id, dst_label_id, e_label_id);
+  auto src_dst_col_pair =
+      loading_config_.GetEdgeSrcDstCol(src_label_id, dst_label_id, e_label_id);
+  if (src_dst_col_pair.first.size() != 1 ||
+      src_dst_col_pair.second.size() != 1) {
+    LOG(FATAL) << "We currently only support one src primary key and one "
+                  "dst primary key";
+  }
+  size_t src_col_ind = src_dst_col_pair.first[0];
+  size_t dst_col_ind = src_dst_col_pair.second[0];
+  CHECK(src_col_ind != dst_col_ind);
+
+  check_edge_invariant(schema_, edge_column_mappings, src_col_ind, dst_col_ind,
+                       src_label_id, dst_label_id, e_label_id);
+
+  std::vector<std::tuple<vid_t, vid_t, EDATA_T>> parsed_edges;
+  std::vector<int32_t> ie_degree, oe_degree;
+  const auto& src_indexer = basic_fragment_loader_.GetLFIndexer(src_label_id);
+  const auto& dst_indexer = basic_fragment_loader_.GetLFIndexer(dst_label_id);
+  ie_degree.resize(dst_indexer.size());
+  oe_degree.resize(src_indexer.size());
+  VLOG(10) << "src indexer size: " << src_indexer.size()
+           << " dst indexer size: " << dst_indexer.size();
+
+  for (auto filename : e_files) {
+    VLOG(10) << "processing " << filename << " with src_col_id " << src_col_ind
+             << " and dst_col_id " << dst_col_ind;
+    if (loading_config_.GetIsBatchReader()) {
+      VLOG(1) << "Using batch reader";
+      addEdgesImplWithStreamReader(filename, src_label_id, dst_label_id,
+                                   e_label_id, ie_degree, oe_degree,
+                                   parsed_edges);
+    } else {
+      VLOG(1) << "Using table reader";
+      addEdgesImplWithTableReader(filename, src_label_id, dst_label_id,
+                                  e_label_id, ie_degree, oe_degree,
+                                  parsed_edges);
+    }
+  }
+  double t = -grape::GetCurrentTime();
+  basic_fragment_loader_.PutEdges(src_label_id, dst_label_id, e_label_id,
+                                  parsed_edges, ie_degree, oe_degree);
+  t += grape::GetCurrentTime();
+  // basic_frag_loader_edge_time_.fetch_add(t);
+  for (double tmp = basic_frag_loader_edge_time_;
+       !basic_frag_loader_edge_time_.compare_exchange_weak(tmp, tmp + t);) {}
+  VLOG(10) << "Finish putting: " << parsed_edges.size() << " edges";
+}
+
+void CSVFragmentLoader::addEdges(label_t src_label_i, label_t dst_label_i,
+                                 label_t edge_label_i,
+                                 const std::vector<std::string>& filenames) {
+  auto src_label_name = schema_.get_vertex_label_name(src_label_i);
+  auto dst_label_name = schema_.get_vertex_label_name(dst_label_i);
+  auto edge_label_name = schema_.get_edge_label_name(edge_label_i);
+  if (filenames.size() <= 0) {
+    LOG(FATAL) << "No edge files found for src label: " << src_label_name
+               << " dst label: " << dst_label_name
+               << " edge label: " << edge_label_name;
+  }
+  if (filenames.size() <= 0) {
+    LOG(FATAL) << "No edge files found for src label: " << src_label_name
+               << " dst label: " << dst_label_name
+               << " edge label: " << edge_label_name;
+  }
+  VLOG(10) << "Init edges src label: " << src_label_name
+           << " dst label: " << dst_label_name
+           << " edge label: " << edge_label_name
+           << " filenames: " << filenames.size();
+  auto& property_types = schema_.get_edge_properties(
+      src_label_name, dst_label_name, edge_label_name);
+  size_t col_num = property_types.size();
+  CHECK_LE(col_num, 1) << "Only single or no property is supported for edge.";
+
+  if (col_num == 0) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<grape::EmptyType>(
+          src_label_i, dst_label_i, edge_label_i);
+    } else {
+      addEdgesImpl<grape::EmptyType>(src_label_i, dst_label_i, edge_label_i,
+                                     filenames);
+    }
+  } else if (property_types[0] == PropertyType::kDate) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<Date>(src_label_i, dst_label_i,
+                                                      edge_label_i);
+    } else {
+      addEdgesImpl<Date>(src_label_i, dst_label_i, edge_label_i, filenames);
+    }
+  } else if (property_types[0] == PropertyType::kInt32) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<int>(src_label_i, dst_label_i,
+                                                     edge_label_i);
+    } else {
+      addEdgesImpl<int>(src_label_i, dst_label_i, edge_label_i, filenames);
+    }
+  } else if (property_types[0] == PropertyType::kInt64) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<int64_t>(
+          src_label_i, dst_label_i, edge_label_i);
+    } else {
+      addEdgesImpl<int64_t>(src_label_i, dst_label_i, edge_label_i, filenames);
+    }
+  } else if (property_types[0] == PropertyType::kString) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<std::string>(
+          src_label_i, dst_label_i, edge_label_i);
+    } else {
+      LOG(FATAL) << "Unsupported edge property type.";
+    }
+  } else if (property_types[0] == PropertyType::kDouble) {
+    if (filenames.empty()) {
+      basic_fragment_loader_.AddNoPropEdgeBatch<double>(
+          src_label_i, dst_label_i, edge_label_i);
+    } else {
+      addEdgesImpl<double>(src_label_i, dst_label_i, edge_label_i, filenames);
+    }
+  } else {
+    LOG(FATAL) << "Unsupported edge property type." << property_types[0];
+  }
+}
+
+void CSVFragmentLoader::loadVertices() {
+  auto vertex_sources = loading_config_.GetVertexLoadingMeta();
+  if (vertex_sources.empty()) {
+    LOG(INFO) << "Skip loading vertices since no vertex source is specified.";
+    return;
+  }
+
+  if (thread_num_ == 1) {
+    LOG(INFO) << "Loading vertices with single thread...";
+    for (auto iter = vertex_sources.begin(); iter != vertex_sources.end();
+         ++iter) {
+      auto v_label_id = iter->first;
+      auto v_files = iter->second;
+      addVertices(v_label_id, v_files);
+    }
+  } else {
+    // copy vertex_sources and edge sources to vector, since we need to
+    // use multi-thread loading.
+    std::vector<std::pair<label_t, std::vector<std::string>>> vertex_files;
+    for (auto iter = vertex_sources.begin(); iter != vertex_sources.end();
+         ++iter) {
+      vertex_files.emplace_back(iter->first, iter->second);
+    }
+    LOG(INFO) << "Parallel loading with " << thread_num_ << " threads, "
+              << " " << vertex_files.size() << " vertex files, ";
+    std::atomic<size_t> v_ind(0);
+    std::vector<std::thread> threads(thread_num_);
+    for (int i = 0; i < thread_num_; ++i) {
+      threads[i] = std::thread([&]() {
+        while (true) {
+          size_t cur = v_ind.fetch_add(1);
+          if (cur >= vertex_files.size()) {
+            break;
+          }
+          auto v_label_id = vertex_files[cur].first;
+          addVertices(v_label_id, vertex_files[cur].second);
+        }
+      });
+    }
+    for (auto& thread : threads) {
+      thread.join();
+    }
+
+    LOG(INFO) << "Finished loading vertices";
+  }
+}
+
+void CSVFragmentLoader::fillVertexReaderMeta(
+    arrow::csv::ReadOptions& read_options,
+    arrow::csv::ParseOptions& parse_options,
+    arrow::csv::ConvertOptions& convert_options, const std::string& v_file,
+    label_t v_label) const {
+  auto time_stamp_parser = arrow::TimestampParser::MakeISO8601();
+  convert_options.timestamp_parsers.emplace_back(time_stamp_parser);
+
+  put_delimiter_option(loading_config_, parse_options);
+  bool header_row = put_skip_rows_option(loading_config_, read_options);
+  put_column_names_option(loading_config_, header_row, v_file,
+                          parse_options.delimiter, read_options);
+  put_escape_char_option(loading_config_, parse_options);
+  put_quote_char_option(loading_config_, parse_options);
+  put_block_size_option(loading_config_, read_options);
+
+  // parse all column_names
+
+  std::vector<std::string> included_col_names;
+  std::vector<size_t> included_col_indices;
+  std::vector<std::string> mapped_property_names;
+
+  auto cur_label_col_mapping = loading_config_.GetVertexColumnMappings(v_label);
+  auto primary_keys = schema_.get_vertex_primary_key(v_label);
+  CHECK(primary_keys.size() == 1);
+  auto primary_key = primary_keys[0];
+
+  if (cur_label_col_mapping.size() == 0) {
+    // use default mapping, we assume the order of the columns in the file is
+    // the same as the order of the properties in the schema, except for
+    // primary key.
+    auto primary_key_name = std::get<1>(primary_key);
+    auto primary_key_ind = std::get<2>(primary_key);
+    auto property_names = schema_.get_vertex_property_names(v_label);
+    // for example, schema is : (name,age)
+    // file header is (id,name,age), the primary key is id.
+    // so, the mapped_property_names are: (id,name,age)
+    CHECK(property_names.size() + 1 == read_options.column_names.size());
+    // insert primary_key to property_names
+    property_names.insert(property_names.begin() + primary_key_ind,
+                          primary_key_name);
+
+    for (auto i = 0; i < read_options.column_names.size(); ++i) {
+      included_col_names.emplace_back(read_options.column_names[i]);
+      included_col_indices.emplace_back(i);
+      // We assume the order of the columns in the file is the same as the
+      // order of the properties in the schema, except for primary key.
+      mapped_property_names.emplace_back(property_names[i]);
+    }
+  } else {
+    for (auto i = 0; i < cur_label_col_mapping.size(); ++i) {
+      auto& [col_id, col_name, property_name] = cur_label_col_mapping[i];
+      if (col_name.empty()) {
+        // use default mapping
+        col_name = read_options.column_names[col_id];
+      }
+      included_col_names.emplace_back(col_name);
+      included_col_indices.emplace_back(col_id);
+      mapped_property_names.emplace_back(property_name);
+    }
+  }
+
+  VLOG(10) << "Include columns: " << included_col_names.size()
+           << gs::to_string(included_col_names);
+  // if empty, then means need all columns
+  convert_options.include_columns = included_col_names;
+
+  // put column_types, col_name : col_type
+  std::unordered_map<std::string, std::shared_ptr<arrow::DataType>> arrow_types;
+  {
+    auto property_types = schema_.get_vertex_properties(v_label);
+    auto property_names = schema_.get_vertex_property_names(v_label);
+    CHECK(property_types.size() == property_names.size());
+
+    for (auto i = 0; i < property_types.size(); ++i) {
+      // for each schema' property name, get the index of the column in
+      // vertex_column mapping, and bind the type with the column name
+      auto property_type = property_types[i];
+      auto property_name = property_names[i];
+      size_t ind = mapped_property_names.size();
+      for (auto i = 0; i < mapped_property_names.size(); ++i) {
+        if (mapped_property_names[i] == property_name) {
+          ind = i;
+          break;
+        }
+      }
+      if (ind == mapped_property_names.size()) {
+        LOG(FATAL) << "The specified property name: " << property_name
+                   << " does not exist in the vertex column mapping for "
+                      "vertex label: "
+                   << schema_.get_vertex_label_name(v_label)
+                   << " please "
+                      "check your configuration";
+      }
+      VLOG(10) << "vertex_label: " << schema_.get_vertex_label_name(v_label)
+               << " property_name: " << property_name
+               << " property_type: " << property_type << " ind: " << ind;
+      arrow_types.insert(
+          {included_col_names[ind], PropertyTypeToArrowType(property_type)});
+    }
+    {
+      // add primary key types;
+      auto primary_key_name = std::get<1>(primary_key);
+      auto primary_key_type = std::get<0>(primary_key);
+      size_t ind = mapped_property_names.size();
+      for (auto i = 0; i < mapped_property_names.size(); ++i) {
+        if (mapped_property_names[i] == primary_key_name) {
+          ind = i;
+          break;
+        }
+      }
+      if (ind == mapped_property_names.size()) {
+        LOG(FATAL) << "The specified property name: " << primary_key_name
+                   << " does not exist in the vertex column mapping, please "
+                      "check your configuration";
+      }
+      arrow_types.insert(
+          {included_col_names[ind], PropertyTypeToArrowType(primary_key_type)});
+    }
+
+    convert_options.column_types = arrow_types;
+  }
+}
+
+void CSVFragmentLoader::fillEdgeReaderMeta(
+    arrow::csv::ReadOptions& read_options,
+    arrow::csv::ParseOptions& parse_options,
+    arrow::csv::ConvertOptions& convert_options, const std::string& e_file,
+    label_t src_label_id, label_t dst_label_id, label_t label_id) const {
+  auto time_stamp_parser =
+      arrow::TimestampParser::MakeISO8601();  // 2011-08-17T14:26:59.961+0000
+  convert_options.timestamp_parsers.emplace_back(time_stamp_parser);
+
+  put_delimiter_option(loading_config_, parse_options);
+  bool header_row = put_skip_rows_option(loading_config_, read_options);
+  put_column_names_option(loading_config_, header_row, e_file,
+                          parse_options.delimiter, read_options);
+  put_escape_char_option(loading_config_, parse_options);
+  put_quote_char_option(loading_config_, parse_options);
+  put_block_size_option(loading_config_, read_options);
+
+  auto src_dst_cols =
+      loading_config_.GetEdgeSrcDstCol(src_label_id, dst_label_id, label_id);
+
+  // parse all column_names
+  // Get all column names(header, and always skip the first row)
+  std::vector<std::string> included_col_names;
+  std::vector<std::string> mapped_property_names;
+
+  {
+    // add src and dst primary col, to included_columns, put src_col and
+    // dst_col at the first of included_columns.
+    CHECK(src_dst_cols.first.size() == 1 && src_dst_cols.second.size() == 1);
+    auto src_col_ind = src_dst_cols.first[0];
+    auto dst_col_ind = src_dst_cols.second[0];
+    CHECK(src_col_ind >= 0 && src_col_ind < read_options.column_names.size());
+    CHECK(dst_col_ind >= 0 && dst_col_ind < read_options.column_names.size());
+
+    included_col_names.emplace_back(read_options.column_names[src_col_ind]);
+    included_col_names.emplace_back(read_options.column_names[dst_col_ind]);
+  }
+
+  auto cur_label_col_mapping = loading_config_.GetEdgeColumnMappings(
+      src_label_id, dst_label_id, label_id);
+  if (cur_label_col_mapping.empty()) {
+    // use default mapping, we assume the order of the columns in the file is
+    // the same as the order of the properties in the schema,
+    auto edge_prop_names =
+        schema_.get_edge_property_names(src_label_id, dst_label_id, label_id);
+    for (auto i = 0; i < edge_prop_names.size(); ++i) {
+      auto property_name = edge_prop_names[i];
+      included_col_names.emplace_back(property_name);
+      mapped_property_names.emplace_back(property_name);
+    }
+  } else {
+    // add the property columns into the included columns
+    for (auto i = 0; i < cur_label_col_mapping.size(); ++i) {
+      // TODO: make the property column's names are in same order with schema.
+      auto& [col_id, col_name, property_name] = cur_label_col_mapping[i];
+      if (col_name.empty()) {
+        // use default mapping
+        col_name = read_options.column_names[col_id];
+      }
+      included_col_names.emplace_back(col_name);
+      mapped_property_names.emplace_back(property_name);
+    }
+  }
+
+  VLOG(10) << "Include Edge columns: " << gs::to_string(included_col_names);
+  // if empty, then means need all columns
+  convert_options.include_columns = included_col_names;
+
+  // put column_types, col_name : col_type
+  std::unordered_map<std::string, std::shared_ptr<arrow::DataType>> arrow_types;
+  {
+    auto property_types =
+        schema_.get_edge_properties(src_label_id, dst_label_id, label_id);
+    auto property_names =
+        schema_.get_edge_property_names(src_label_id, dst_label_id, label_id);
+    CHECK(property_types.size() == property_names.size());
+    for (auto i = 0; i < property_types.size(); ++i) {
+      // for each schema' property name, get the index of the column in
+      // vertex_column mapping, and bind the type with the column name
+      auto property_type = property_types[i];
+      auto property_name = property_names[i];
+      size_t ind = mapped_property_names.size();
+      for (auto i = 0; i < mapped_property_names.size(); ++i) {
+        if (mapped_property_names[i] == property_name) {
+          ind = i;
+          break;
+        }
+      }
+      if (ind == mapped_property_names.size()) {
+        LOG(FATAL) << "The specified property name: " << property_name
+                   << " does not exist in the vertex column mapping, please "
+                      "check your configuration";
+      }
+      VLOG(10) << "vertex_label: " << schema_.get_edge_label_name(label_id)
+               << " property_name: " << property_name
+               << " property_type: " << property_type << " ind: " << ind;
+      arrow_types.insert({included_col_names[ind + 2],
+                          PropertyTypeToArrowType(property_type)});
+    }
+    {
+      // add src and dst primary col, to included_columns and column types.
+      auto src_dst_cols = loading_config_.GetEdgeSrcDstCol(
+          src_label_id, dst_label_id, label_id);
+      CHECK(src_dst_cols.first.size() == 1 && src_dst_cols.second.size() == 1);
+      auto src_col_ind = src_dst_cols.first[0];
+      auto dst_col_ind = src_dst_cols.second[0];
+      CHECK(src_col_ind >= 0 && src_col_ind < read_options.column_names.size());
+      CHECK(dst_col_ind >= 0 && dst_col_ind < read_options.column_names.size());
+      PropertyType src_col_type, dst_col_type;
+      {
+        auto src_primary_keys = schema_.get_vertex_primary_key(src_label_id);
+        CHECK(src_primary_keys.size() == 1);
+        src_col_type = std::get<0>(src_primary_keys[0]);
+        arrow_types.insert({read_options.column_names[src_col_ind],
+                            PropertyTypeToArrowType(src_col_type)});
+      }
+      {
+        auto dst_primary_keys = schema_.get_vertex_primary_key(dst_label_id);
+        CHECK(dst_primary_keys.size() == 1);
+        dst_col_type = std::get<0>(dst_primary_keys[0]);
+        arrow_types.insert({read_options.column_names[dst_col_ind],
+                            PropertyTypeToArrowType(dst_col_type)});
+      }
+    }
+
+    convert_options.column_types = arrow_types;
+
+    VLOG(10) << "Column types: ";
+    for (auto iter : arrow_types) {
+      VLOG(10) << iter.first << " : " << iter.second->ToString();
+    }
+  }
+}
+
+void CSVFragmentLoader::loadEdges() {
+  auto& edge_sources = loading_config_.GetEdgeLoadingMeta();
+
+  if (edge_sources.empty()) {
+    LOG(INFO) << "Skip loading edges since no edge source is specified.";
+    return;
+  }
+
+  if (thread_num_ == 1) {
+    LOG(INFO) << "Loading edges with single thread...";
+    for (auto iter = edge_sources.begin(); iter != edge_sources.end(); ++iter) {
+      auto& src_label_id = std::get<0>(iter->first);
+      auto& dst_label_id = std::get<1>(iter->first);
+      auto& e_label_id = std::get<2>(iter->first);
+      auto& e_files = iter->second;
+
+      addEdges(src_label_id, dst_label_id, e_label_id, e_files);
+    }
+  } else {
+    std::vector<std::pair<typename LoadingConfig::edge_triplet_type,
+                          std::vector<std::string>>>
+        edge_files;
+    for (auto iter = edge_sources.begin(); iter != edge_sources.end(); ++iter) {
+      edge_files.emplace_back(iter->first, iter->second);
+    }
+    LOG(INFO) << "Parallel loading with " << thread_num_ << " threads, "
+              << edge_files.size() << " edge files.";
+    std::atomic<size_t> e_ind(0);
+    std::vector<std::thread> threads(thread_num_);
+    for (int i = 0; i < thread_num_; ++i) {
+      threads[i] = std::thread([&]() {
+        while (true) {
+          size_t cur = e_ind.fetch_add(1);
+          if (cur >= edge_files.size()) {
+            break;
+          }
+          auto& edge_file = edge_files[cur];
+          auto src_label_id = std::get<0>(edge_file.first);
+          auto dst_label_id = std::get<1>(edge_file.first);
+          auto e_label_id = std::get<2>(edge_file.first);
+          auto& file_names = edge_file.second;
+          addEdges(src_label_id, dst_label_id, e_label_id, file_names);
+        }
+      });
+    }
+    for (auto& thread : threads) {
+      thread.join();
+    }
+    LOG(INFO) << "Finished loading edges";
+  }
+}
+
+void CSVFragmentLoader::LoadFragment(MutablePropertyFragment& fragment) {
+  loadVertices();
+  loadEdges();
+
+  return basic_fragment_loader_.LoadFragment(fragment);
+}
+
+}  // namespace gs
diff --git a/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.h b/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.h
new file mode 100644
index 000000000000..54433e3876ae
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/csv_fragment_loader.h
@@ -0,0 +1,153 @@
+
+/** 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 STORAGES_RT_MUTABLE_GRAPH_LOADER_CSV_FRAGMENT_LOADER_H_
+#define STORAGES_RT_MUTABLE_GRAPH_LOADER_CSV_FRAGMENT_LOADER_H_
+
+#include "flex/storages/rt_mutable_graph/loader/basic_fragment_loader.h"
+#include "flex/storages/rt_mutable_graph/loader/i_fragment_loader.h"
+#include "flex/storages/rt_mutable_graph/loading_config.h"
+#include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
+
+#include <arrow/api.h>
+#include <arrow/csv/api.h>
+#include <arrow/io/api.h>
+#include "arrow/util/value_parsing.h"
+
+#include "grape/util.h"
+
+namespace gs {
+
+// LoadFragment for csv files.
+class CSVFragmentLoader : public IFragmentLoader {
+ public:
+  CSVFragmentLoader(const Schema& schema, const LoadingConfig& loading_config,
+                    int32_t thread_num)
+      : loading_config_(loading_config),
+        schema_(schema),
+        thread_num_(thread_num),
+        basic_fragment_loader_(schema_),
+        read_vertex_table_time_(0),
+        read_edge_table_time_(0),
+        convert_to_internal_vertex_time_(0),
+        convert_to_internal_edge_time_(0),
+        basic_frag_loader_vertex_time_(0),
+        basic_frag_loader_edge_time_(0) {
+    vertex_label_num_ = schema_.vertex_label_num();
+    edge_label_num_ = schema_.edge_label_num();
+  }
+
+  ~CSVFragmentLoader() {}
+
+  FragmentLoaderType GetFragmentLoaderType() const override {
+    return FragmentLoaderType::kCSVFragmentLoader;
+  }
+
+  void LoadFragment(MutablePropertyFragment& fragment) override;
+
+ private:
+  void loadVertices();
+
+  void loadEdges();
+
+  void addVertices(label_t v_label_id, const std::vector<std::string>& v_files);
+
+  void addVerticesImpl(label_t v_label_id, const std::string& v_label_name,
+                       const std::vector<std::string> v_file,
+                       IdIndexer<oid_t, vid_t>& indexer);
+
+  void addVerticesImplWithStreamReader(const std::string& filename,
+                                       label_t v_label_id,
+                                       IdIndexer<oid_t, vid_t>& indexer);
+
+  void addVerticesImplWithTableReader(const std::string& filename,
+                                      label_t v_label_id,
+                                      IdIndexer<oid_t, vid_t>& indexer);
+
+  void addVertexBatch(
+      label_t v_label_id, IdIndexer<oid_t, vid_t>& indexer,
+      std::shared_ptr<arrow::Array>& primary_key_col,
+      const std::vector<std::shared_ptr<arrow::Array>>& property_cols);
+
+  void addVertexBatch(
+      label_t v_label_id, IdIndexer<oid_t, vid_t>& indexer,
+      std::shared_ptr<arrow::ChunkedArray>& primary_key_col,
+      const std::vector<std::shared_ptr<arrow::ChunkedArray>>& property_cols);
+
+  void addEdges(label_t src_label_id, label_t dst_label_id, label_t e_label_id,
+                const std::vector<std::string>& e_files);
+
+  template <typename EDATA_T>
+  void addEdgesImpl(label_t src_label_id, label_t dst_label_id,
+                    label_t e_label_id,
+                    const std::vector<std::string>& e_files);
+
+  template <typename EDATA_T>
+  void addEdgesImplWithStreamReader(
+      const std::string& file_name, label_t src_label_id, label_t dst_label_id,
+      label_t e_label_id, std::vector<int32_t>& ie_degree,
+      std::vector<int32_t>& oe_degree,
+      std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& edges);
+
+  template <typename EDATA_T>
+  void addEdgesImplWithTableReader(
+      const std::string& filename, label_t src_label_id, label_t dst_label_id,
+      label_t e_label_id, std::vector<int32_t>& ie_degree,
+      std::vector<int32_t>& oe_degree,
+      std::vector<std::tuple<vid_t, vid_t, EDATA_T>>& edges);
+
+  std::shared_ptr<arrow::csv::StreamingReader> createVertexStreamReader(
+      label_t v_label, const std::string& v_file);
+
+  std::shared_ptr<arrow::csv::TableReader> createVertexTableReader(
+      label_t v_label, const std::string& v_file);
+
+  std::shared_ptr<arrow::csv::StreamingReader> createEdgeStreamReader(
+      label_t src_label_id, label_t dst_label_id, label_t e_label,
+      const std::string& e_file);
+
+  std::shared_ptr<arrow::csv::TableReader> createEdgeTableReader(
+      label_t src_label_id, label_t dst_label_id, label_t e_label,
+      const std::string& e_file);
+
+  void fillEdgeReaderMeta(arrow::csv::ReadOptions& read_options,
+                          arrow::csv::ParseOptions& parse_options,
+                          arrow::csv::ConvertOptions& convert_options,
+                          const std::string& e_file, label_t src_label_id,
+                          label_t dst_label_id, label_t label_id) const;
+
+  void fillVertexReaderMeta(arrow::csv::ReadOptions& read_options,
+                            arrow::csv::ParseOptions& parse_options,
+                            arrow::csv::ConvertOptions& convert_options,
+                            const std::string& v_file, label_t v_label) const;
+
+  const LoadingConfig& loading_config_;
+  const Schema& schema_;
+  size_t vertex_label_num_, edge_label_num_;
+  int32_t thread_num_;
+
+  mutable BasicFragmentLoader basic_fragment_loader_;
+
+  std::atomic<double> read_vertex_table_time_, read_edge_table_time_;
+  std::atomic<double> convert_to_internal_vertex_time_,
+      convert_to_internal_edge_time_;
+  std::atomic<double> basic_frag_loader_vertex_time_,
+      basic_frag_loader_edge_time_;
+};
+
+}  // namespace gs
+
+#endif  // STORAGES_RT_MUTABLE_GRAPH_LOADER_CSV_FRAGMENT_LOADER_H_
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/loader/i_fragment_loader.h b/flex/storages/rt_mutable_graph/loader/i_fragment_loader.h
new file mode 100644
index 000000000000..c332fa35e7e0
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/i_fragment_loader.h
@@ -0,0 +1,36 @@
+/** 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 STORAGES_RT_MUTABLE_GRAPH_LOADER_FRAGMENT_LOADER_H_
+#define STORAGES_RT_MUTABLE_GRAPH_LOADER_FRAGMENT_LOADER_H_
+
+#include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
+
+namespace gs {
+
+enum class FragmentLoaderType { kCSVFragmentLoader };
+
+// For different input format, we should implement different fragment loader.
+class IFragmentLoader {
+ public:
+  virtual ~IFragmentLoader() = default;
+  // get the fragment loader type
+  virtual FragmentLoaderType GetFragmentLoaderType() const = 0;
+  virtual void LoadFragment(MutablePropertyFragment& fragment) = 0;
+};
+
+}  // namespace gs
+
+#endif  // STORAGES_RT_MUTABLE_GRAPH_LOADER_FRAGMENT_LOADER_H_
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/loader/loader_factory.cc b/flex/storages/rt_mutable_graph/loader/loader_factory.cc
new file mode 100644
index 000000000000..4cf80e049eaa
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/loader_factory.cc
@@ -0,0 +1,30 @@
+/** 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 "flex/storages/rt_mutable_graph/loader/loader_factory.h"
+
+namespace gs {
+
+std::shared_ptr<IFragmentLoader> LoaderFactory::CreateFragmentLoader(
+    const Schema& schema, const LoadingConfig& loading_config, int thread_num) {
+  if (loading_config.GetFormat() == "csv") {
+    return std::make_shared<CSVFragmentLoader>(schema, loading_config,
+                                               thread_num);
+  } else {
+    LOG(FATAL) << "Unsupported format: " << loading_config.GetFormat();
+  }
+}
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/loader/loader_factory.h b/flex/storages/rt_mutable_graph/loader/loader_factory.h
new file mode 100644
index 000000000000..7c45081402fb
--- /dev/null
+++ b/flex/storages/rt_mutable_graph/loader/loader_factory.h
@@ -0,0 +1,33 @@
+/** 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 STORAGES_RT_MUTABLE_GRAPH_LOADER_LOADER_FACTORY_H_
+#define STORAGES_RT_MUTABLE_GRAPH_LOADER_LOADER_FACTORY_H_
+
+#include <memory>
+#include "flex/storages/rt_mutable_graph/loader/csv_fragment_loader.h"
+#include "flex/storages/rt_mutable_graph/loader/i_fragment_loader.h"
+#include "flex/storages/rt_mutable_graph/loading_config.h"
+
+namespace gs {
+class LoaderFactory {
+ public:
+  static std::shared_ptr<IFragmentLoader> CreateFragmentLoader(
+      const Schema& schema, const LoadingConfig& loading_config,
+      int thread_num);
+};
+}  // namespace gs
+
+#endif  // STORAGES_RT_MUTABLE_GRAPH_LOADER_LOADER_FACTORY_H_
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/loading_config.cc b/flex/storages/rt_mutable_graph/loading_config.cc
index b6153dc10d6e..39e99ecb13e0 100644
--- a/flex/storages/rt_mutable_graph/loading_config.cc
+++ b/flex/storages/rt_mutable_graph/loading_config.cc
@@ -14,13 +14,16 @@
  */
 
 #include "flex/storages/rt_mutable_graph/loading_config.h"
-
 #include <yaml-cpp/yaml.h>
 #include <filesystem>
+#include <iostream>
+#include <sstream>
 #include <string>
 #include <tuple>
+#include "flex/engines/hqps_db/core/utils/hqps_utils.h"
 
 namespace gs {
+
 namespace config_parsing {
 
 // fetch the primary key of the src and dst vertex label in the edge file,
@@ -47,8 +50,9 @@ static bool fetch_src_dst_column_mapping(const Schema& schema, YAML::Node node,
     for (auto i = 0; i < column_mappings.size(); ++i) {
       auto column_mapping = column_mappings[i]["column"];
       auto name = column_mapping["name"].as<std::string>();
-      if (name != schema_primary_key[i].second) {
-        LOG(ERROR) << "Expect column name [" << schema_primary_key[i].second
+      if (name != std::get<1>(schema_primary_key[i])) {
+        LOG(ERROR) << "Expect column name ["
+                   << std::get<1>(schema_primary_key[i])
                    << "] for source vertex mapping, at index: " << i
                    << ", got: " << name;
         return false;
@@ -65,11 +69,36 @@ static bool fetch_src_dst_column_mapping(const Schema& schema, YAML::Node node,
   return true;
 }
 
+// Function to parse memory size represented as a string
+uint64_t parse_block_size(const std::string& memorySizeStr) {
+  // Create a stringstream to parse the memory size string
+  std::istringstream ss(memorySizeStr);
+
+  // Extract the numeric part of the string
+  uint64_t memorySize;
+  ss >> memorySize;
+
+  // Handle unit prefixes (e.g., KB, MB, GB, etc.)
+  std::string unit;
+  ss >> unit;
+
+  // Convert the value to bytes based on the unit
+  if (unit == "KB") {
+    memorySize *= 1024;
+  } else if (unit == "MB") {
+    memorySize *= 1024 * 1024;
+  } else if (unit == "GB") {
+    memorySize *= 1024 * 1024 * 1024;
+  }
+
+  return memorySize;
+}
+
 // Parse the mappings and check whether property exists in the schema.
 template <typename FUNC>
 static bool parse_column_mappings(
     YAML::Node node, const Schema& schema, const std::string& label_name,
-    std::vector<std::pair<size_t, std::string>>& column_mappings,
+    std::vector<std::tuple<size_t, std::string, std::string>>& column_mappings,
     FUNC condition) {
   if (!node.IsSequence()) {
     LOG(ERROR) << "column_mappings should be a sequence";
@@ -87,6 +116,12 @@ static bool parse_column_mappings(
       LOG(ERROR) << "Expect column index for column mapping";
       return false;
     }
+    std::string column_name;
+    if (!get_scalar(column_mapping, "name", column_name)) {
+      VLOG(10) << "Column name for col_id: " << column_id
+               << " is not set, make it empty";
+    }
+
     std::string property_name;  // property name is optional.
     if (!get_scalar(node[i], "property", property_name)) {
       LOG(ERROR) << "Expect property name for column mapping";
@@ -97,7 +132,7 @@ static bool parse_column_mappings(
                  << "the schema for label : " << label_name;
       return false;
     }
-    column_mappings.emplace_back(column_id, property_name);
+    column_mappings.emplace_back(column_id, column_name, property_name);
   }
   // If no column mapping is set, use default mapping.
   return true;
@@ -107,7 +142,8 @@ static bool parse_column_mappings(
 static bool parse_vertex_files(
     YAML::Node node, const Schema& schema, const std::string& data_location,
     std::unordered_map<label_t, std::vector<std::string>>& files,
-    std::unordered_map<label_t, std::vector<std::pair<size_t, std::string>>>&
+    std::unordered_map<
+        label_t, std::vector<std::tuple<size_t, std::string, std::string>>>&
         vertex_mapping) {
   std::string label_name;
   if (!get_scalar(node, "type_name", label_name)) {
@@ -149,8 +185,8 @@ static bool parse_vertex_files(
   } else {
     // if no column_mappings is given, use default mapping.
     VLOG(10) << "No vertex mapping is given, use default mapping";
-    vertex_mapping.emplace(label_id,
-                           std::vector<std::pair<size_t, std::string>>());
+    vertex_mapping.emplace(
+        label_id, std::vector<std::tuple<size_t, std::string, std::string>>());
   }
   if (files_node) {
     if (!files_node.IsSequence()) {
@@ -177,7 +213,8 @@ static bool parse_vertex_files(
 static bool parse_vertices_files_schema(
     YAML::Node node, const Schema& schema, const std::string& data_location,
     std::unordered_map<label_t, std::vector<std::string>>& files,
-    std::unordered_map<label_t, std::vector<std::pair<size_t, std::string>>>&
+    std::unordered_map<
+        label_t, std::vector<std::tuple<size_t, std::string, std::string>>>&
         column_mappings) {
   if (!node.IsSequence()) {
     LOG(ERROR) << "vertex is not set properly";
@@ -198,10 +235,10 @@ static bool parse_edge_files(
     std::unordered_map<
         std::tuple<label_t, label_t, label_t>, std::vector<std::string>,
         boost::hash<std::tuple<label_t, label_t, label_t>>>& files,
-    std::unordered_map<typename LoadingConfig::edge_triplet_type,
-                       std::vector<std::pair<size_t, std::string>>,
-                       boost::hash<typename LoadingConfig::edge_triplet_type>>&
-        edge_mapping,
+    std::unordered_map<
+        typename LoadingConfig::edge_triplet_type,
+        std::vector<std::tuple<size_t, std::string, std::string>>,
+        boost::hash<typename LoadingConfig::edge_triplet_type>>& edge_mapping,
     std::unordered_map<typename LoadingConfig::edge_triplet_type,
                        std::pair<std::vector<size_t>, std::vector<size_t>>,
                        boost::hash<typename LoadingConfig::edge_triplet_type>>&
@@ -302,8 +339,9 @@ static bool parse_edge_files(
   } else {
     VLOG(10) << "No edge column mapping is given, use default mapping";
     // use default mapping
-    edge_mapping.emplace(std::tuple{src_label_id, dst_label_id, edge_label_id},
-                         std::vector<std::pair<size_t, std::string>>{});
+    edge_mapping.emplace(
+        std::tuple{src_label_id, dst_label_id, edge_label_id},
+        std::vector<std::tuple<size_t, std::string, std::string>>{});
   }
 
   YAML::Node files_node = node["inputs"];
@@ -335,10 +373,10 @@ static bool parse_edges_files_schema(
     std::unordered_map<
         std::tuple<label_t, label_t, label_t>, std::vector<std::string>,
         boost::hash<std::tuple<label_t, label_t, label_t>>>& files,
-    std::unordered_map<typename LoadingConfig::edge_triplet_type,
-                       std::vector<std::pair<size_t, std::string>>,
-                       boost::hash<typename LoadingConfig::edge_triplet_type>>&
-        edge_mapping,
+    std::unordered_map<
+        typename LoadingConfig::edge_triplet_type,
+        std::vector<std::tuple<size_t, std::string, std::string>>,
+        boost::hash<typename LoadingConfig::edge_triplet_type>>& edge_mapping,
     std::unordered_map<std::tuple<label_t, label_t, label_t>,
                        std::pair<std::vector<size_t>, std::vector<size_t>>,
                        boost::hash<typename LoadingConfig::edge_triplet_type>>&
@@ -365,8 +403,8 @@ static bool parse_bulk_load_config_file(const std::string& config_file,
   std::string data_location;
   load_config.scheme_ = "file";  // default data source is file
   load_config.method_ = "init";
-  load_config.delimiter_ = "|";
   load_config.format_ = "csv";
+
   if (root["loading_config"]) {
     auto loading_config_node = root["loading_config"];
     if (loading_config_node["data_source"]) {
@@ -379,9 +417,44 @@ static bool parse_bulk_load_config_file(const std::string& config_file,
     if (format_node) {
       get_scalar(format_node, "type", load_config.format_);
       if (load_config.format_ == "csv") {
-        if (format_node["meta_data"]) {
-          get_scalar(format_node["meta_data"], "delimiter",
-                     load_config.delimiter_);
+        // set default delimiter before we parsing meta_data
+        load_config.metadata_[reader_options::DELIMITER] = "|";
+        load_config.metadata_[reader_options::HEADER_ROW] = "true";
+        load_config.metadata_[reader_options::QUOTING] = "false";
+        load_config.metadata_[reader_options::QUOTE_CHAR] = "\"";
+        load_config.metadata_[reader_options::DOUBLE_QUOTE] = "false";
+        load_config.metadata_[reader_options::ESCAPE_CHAR] = "\\";
+        load_config.metadata_[reader_options::ESCAPING] = "false";
+        load_config.metadata_[reader_options::BATCH_SIZE_KEY] =
+            std::to_string(reader_options::DEFAULT_BLOCK_SIZE);
+        load_config.metadata_[reader_options::BATCH_READER] = "false";
+        // put all key values in meta_data into metadata_
+        if (format_node["metadata"]) {
+          auto meta_data_node = format_node["metadata"];
+          if (!meta_data_node.IsMap()) {
+            LOG(ERROR) << "metadata should be a map";
+            return false;
+          }
+          for (auto it = meta_data_node.begin(); it != meta_data_node.end();
+               ++it) {
+            // override previous settings.
+            auto key = it->first.as<std::string>();
+            VLOG(1) << "Got metadata key: " << key
+                    << " value: " << it->second.as<std::string>();
+            if (reader_options::CSV_META_KEY_WORDS.find(key) !=
+                reader_options::CSV_META_KEY_WORDS.end()) {
+              if (key == reader_options::BATCH_SIZE_KEY) {
+                // special case for block size
+                // parse block size (MB, b, KB, B) to bytes
+                auto block_size_str = it->second.as<std::string>();
+                auto block_size = parse_block_size(block_size_str);
+                load_config.metadata_[reader_options::BATCH_SIZE_KEY] =
+                    std::to_string(block_size);
+              } else {
+                load_config.metadata_[key] = it->second.as<std::string>();
+              }
+            }
+          }
         }
       } else {
         LOG(ERROR) << "Only support csv format now";
@@ -389,13 +462,6 @@ static bool parse_bulk_load_config_file(const std::string& config_file,
       }
     }
   }
-  // only delimeter with | is supported now
-  if (load_config.GetValidDelimiters().find(load_config.delimiter_) ==
-      load_config.GetValidDelimiters().end()) {
-    LOG(ERROR) << "Not valid delimiter: " << load_config.delimiter_
-               << ", supported delimeters: '|'";
-    return false;
-  }
   if (load_config.method_ != "init") {
     LOG(ERROR) << "Only support init method now";
     return false;
@@ -412,8 +478,10 @@ static bool parse_bulk_load_config_file(const std::string& config_file,
   }
   LOG(INFO) << "scheme: " << load_config.scheme_
             << ", data_location: " << data_location
-            << ", method: " << load_config.method_
-            << ", delimiter: " << load_config.delimiter_;
+            << ", method: " << load_config.method_ << ", delimiter: "
+            << load_config.metadata_[reader_options::DELIMITER]
+            << ", include header row: "
+            << load_config.metadata_[reader_options::HEADER_ROW];
 
   if (root["vertex_mappings"]) {
     VLOG(10) << "vertex_mappings is set";
@@ -453,22 +521,16 @@ LoadingConfig LoadingConfig::ParseFromYaml(const Schema& schema,
 }
 
 LoadingConfig::LoadingConfig(const Schema& schema)
-    : schema_(schema),
-      scheme_("file"),
-      delimiter_("|"),
-      method_("init"),
-      format_("csv") {}
+    : schema_(schema), scheme_("file"), method_("init"), format_("csv") {}
 
 LoadingConfig::LoadingConfig(const Schema& schema,
                              const std::string& data_source,
                              const std::string& delimiter,
                              const std::string& method,
                              const std::string& format)
-    : schema_(schema),
-      scheme_(data_source),
-      delimiter_(delimiter),
-      method_(method),
-      format_(format) {}
+    : schema_(schema), scheme_(data_source), method_(method), format_(format) {
+  metadata_[reader_options::DELIMITER] = delimiter;
+}
 
 bool LoadingConfig::AddVertexSources(const std::string& label,
                                      const std::string& file_path) {
@@ -495,18 +557,63 @@ bool LoadingConfig::AddEdgeSources(const std::string& src_label,
 }
 
 void LoadingConfig::SetScheme(const std::string& scheme) { scheme_ = scheme; }
-void LoadingConfig::SetDelimiter(const std::string& delimiter) {
-  delimiter_ = delimiter;
+void LoadingConfig::SetDelimiter(const char& delimiter) {
+  metadata_[reader_options::DELIMITER] = std::string(1, delimiter);
 }
 void LoadingConfig::SetMethod(const std::string& method) { method_ = method; }
 
 // getters
 const std::string& LoadingConfig::GetScheme() const { return scheme_; }
 
-const std::string& LoadingConfig::GetDelimiter() const { return delimiter_; }
+const std::string& LoadingConfig::GetDelimiter() const {
+  return metadata_.at(reader_options::DELIMITER);
+}
+
+bool LoadingConfig::GetHasHeaderRow() const {
+  auto str = metadata_.at(reader_options::HEADER_ROW);
+  return str == "true" || str == "True" || str == "TRUE";
+}
+
+const std::string& LoadingConfig::GetFormat() const { return format_; }
 
 const std::string& LoadingConfig::GetMethod() const { return method_; }
 
+const std::string& LoadingConfig::GetEscapeChar() const {
+  return metadata_.at(reader_options::ESCAPE_CHAR);
+}
+
+bool LoadingConfig::GetIsEscaping() const {
+  auto str = metadata_.at(reader_options::ESCAPING);
+  return str == "true" || str == "True" || str == "TRUE";
+}
+
+const std::string& LoadingConfig::GetQuotingChar() const {
+  return metadata_.at(reader_options::QUOTE_CHAR);
+}
+
+bool LoadingConfig::GetIsQuoting() const {
+  auto str = metadata_.at(reader_options::QUOTING);
+  return str == "true" || str == "True" || str == "TRUE";
+}
+
+bool LoadingConfig::GetIsDoubleQuoting() const {
+  auto str = metadata_.at(reader_options::DOUBLE_QUOTE);
+  return str == "true" || str == "True" || str == "TRUE";
+}
+
+int32_t LoadingConfig::GetBatchSize() const {
+  if (metadata_.find(reader_options::BATCH_SIZE_KEY) == metadata_.end()) {
+    return reader_options::DEFAULT_BLOCK_SIZE;
+  }
+  auto str = metadata_.at(reader_options::BATCH_SIZE_KEY);
+  return std::stoi(str);
+}
+
+bool LoadingConfig::GetIsBatchReader() const {
+  auto str = metadata_.at(reader_options::BATCH_READER);
+  return str == "true" || str == "True" || str == "TRUE";
+}
+
 const std::unordered_map<LoadingConfig::schema_label_type,
                          std::vector<std::string>>&
 LoadingConfig::GetVertexLoadingMeta() const {
@@ -520,14 +627,14 @@ LoadingConfig::GetEdgeLoadingMeta() const {
   return edge_loading_meta_;
 }
 
-const std::vector<std::pair<size_t, std::string>>&
+const std::vector<std::tuple<size_t, std::string, std::string>>&
 LoadingConfig::GetVertexColumnMappings(label_t label_id) const {
   CHECK(vertex_column_mappings_.find(label_id) !=
         vertex_column_mappings_.end());
   return vertex_column_mappings_.at(label_id);
 }
 
-const std::vector<std::pair<size_t, std::string>>&
+const std::vector<std::tuple<size_t, std::string, std::string>>&
 LoadingConfig::GetEdgeColumnMappings(label_t src_label_id, label_t dst_label_id,
                                      label_t edge_label_id) const {
   auto key = std::make_tuple(src_label_id, dst_label_id, edge_label_id);
@@ -543,10 +650,4 @@ LoadingConfig::GetEdgeSrcDstCol(label_t src_label_id, label_t dst_label_id,
   return edge_src_dst_col_.at(key);
 }
 
-const std::unordered_set<std::string>& LoadingConfig::GetValidDelimiters() {
-  return LoadingConfig::valid_delimiter_;
-}
-// define delimeter here
-const std::unordered_set<std::string> LoadingConfig::valid_delimiter_ = {"|"};
-
 }  // namespace gs
diff --git a/flex/storages/rt_mutable_graph/loading_config.h b/flex/storages/rt_mutable_graph/loading_config.h
index f5193e6105f4..15f2544fae2b 100644
--- a/flex/storages/rt_mutable_graph/loading_config.h
+++ b/flex/storages/rt_mutable_graph/loading_config.h
@@ -22,11 +22,41 @@
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
+#include "arrow/api.h"
+#include "arrow/csv/options.h"
 #include "flex/storages/rt_mutable_graph/schema.h"
+#include "flex/utils/arrow_utils.h"
 #include "flex/utils/yaml_utils.h"
 
+#include "boost/algorithm/string.hpp"
+
 namespace gs {
 
+namespace reader_options {
+static const int32_t DEFAULT_BLOCK_SIZE = (1 << 20);  // 1MB
+
+// KEY_WORDS for configurations
+static const char* DELIMITER = "delimiter";
+static const char* HEADER_ROW = "header_row";
+static const char* INCLUDE_COLUMNS = "include_columns";
+static const char* COLUMN_TYPES = "column_types";
+static const char* ESCAPING = "escaping";
+static const char* ESCAPE_CHAR = "escape_char";
+static const char* QUOTING = "quoting";
+static const char* QUOTE_CHAR = "quote_char";
+static const char* DOUBLE_QUOTE = "double_quote";
+static const char* BATCH_SIZE_KEY = "batch_size";
+// whether or not to use record batch reader. If true, the reader will read
+// data in batches, otherwise, the reader will read data row by row.
+static const char* BATCH_READER = "batch_reader";
+
+static const std::unordered_set<std::string> CSV_META_KEY_WORDS = {
+    DELIMITER,    HEADER_ROW,     INCLUDE_COLUMNS, COLUMN_TYPES,
+    ESCAPING,     ESCAPE_CHAR,    QUOTING,         QUOTE_CHAR,
+    DOUBLE_QUOTE, BATCH_SIZE_KEY, BATCH_READER};
+
+}  // namespace reader_options
+
 class LoadingConfig;
 
 namespace config_parsing {
@@ -42,7 +72,6 @@ class LoadingConfig {
   using edge_triplet_type =
       std::tuple<schema_label_type, schema_label_type,
                  schema_label_type>;  // src_label_t, dst_label_t, edge_label_t
-  static const std::unordered_set<std::string> valid_delimiter_;
 
   // Check whether loading config file is consistent with schema
   static LoadingConfig ParseFromYaml(const Schema& schema,
@@ -66,13 +95,22 @@ class LoadingConfig {
                       size_t dst_pri_key_ind, const std::string& file_path);
 
   void SetScheme(const std::string& data_source);
-  void SetDelimiter(const std::string& delimiter);
+  void SetDelimiter(const char& delimiter);
   void SetMethod(const std::string& method);
 
   // getters
   const std::string& GetScheme() const;
   const std::string& GetDelimiter() const;
   const std::string& GetMethod() const;
+  const std::string& GetFormat() const;
+  bool GetHasHeaderRow() const;
+  const std::string& GetEscapeChar() const;
+  bool GetIsEscaping() const;
+  const std::string& GetQuotingChar() const;
+  bool GetIsQuoting() const;
+  bool GetIsDoubleQuoting() const;
+  int32_t GetBatchSize() const;
+  bool GetIsBatchReader() const;
   const std::unordered_map<schema_label_type, std::vector<std::string>>&
   GetVertexLoadingMeta() const;
   const std::unordered_map<edge_triplet_type, std::vector<std::string>,
@@ -81,33 +119,36 @@ class LoadingConfig {
 
   // Get vertex column mappings. Each element in the vector is a pair of
   // <column_index, property_name>.
-  const std::vector<std::pair<size_t, std::string>>& GetVertexColumnMappings(
-      label_t label_id) const;
+  const std::vector<std::tuple<size_t, std::string, std::string>>&
+  GetVertexColumnMappings(label_t label_id) const;
 
   // Get edge column mappings. Each element in the vector is a pair of
-  // <column_index, schema_property_name>.
-  const std::vector<std::pair<size_t, std::string>>& GetEdgeColumnMappings(
-      label_t src_label_id, label_t dst_label_id, label_t edge_label_id) const;
+  // <column_index,column_name, schema_property_name>.
+  const std::vector<std::tuple<size_t, std::string, std::string>>&
+  GetEdgeColumnMappings(label_t src_label_id, label_t dst_label_id,
+                        label_t edge_label_id) const;
 
   // Get src_id and dst_id column index for edge label.
   const std::pair<std::vector<size_t>, std::vector<size_t>>& GetEdgeSrcDstCol(
       label_t src_label_id, label_t dst_label_id, label_t edge_label_id) const;
 
-  static const std::unordered_set<std::string>& GetValidDelimiters();
-
  private:
   const Schema& schema_;
-  std::string scheme_;     // "file", "hdfs", "oss", "s3"
-  std::string delimiter_;  // "\t", ",", " ", "|"
-  std::string method_;     // init, append, overwrite
-  std::string format_;     // csv, tsv, json, parquet
+  std::string scheme_;  // "file", "hdfs", "oss", "s3"
+  std::string method_;  // init, append, overwrite
+  std::string format_;  // csv, tsv, json, parquet
+
+  // meta_data, stores all the meta info about loading
+  std::unordered_map<std::string, std::string> metadata_;
 
   std::unordered_map<schema_label_type, std::vector<std::string>>
       vertex_loading_meta_;  // <vertex_label_id, std::vector<file_path_>>
   std::unordered_map<schema_label_type,
-                     std::vector<std::pair<size_t, std::string>>>
+                     std::vector<std::tuple<size_t, std::string, std::string>>>
       vertex_column_mappings_;  // match which column in file to which property
-                                // in schema
+                                // in schema. {col_ind, col_name,
+                                // schema_prop_name}
+                                // col_name can be empty
 
   std::unordered_map<edge_triplet_type, std::vector<std::string>,
                      boost::hash<edge_triplet_type>>
@@ -116,10 +157,11 @@ class LoadingConfig {
                            // <file_path>
   // All Edge Files share the same File schema.
   std::unordered_map<edge_triplet_type,
-                     std::vector<std::pair<size_t, std::string>>,
+                     std::vector<std::tuple<size_t, std::string, std::string>>,
                      boost::hash<edge_triplet_type>>
       edge_column_mappings_;  // match which column in file to which property in
-                              // schema
+                              // schema, {col_ind, col_name, schema_prop_name}
+                              // col_name can be empty
 
   std::unordered_map<edge_triplet_type,
                      std::pair<std::vector<size_t>, std::vector<size_t>>,
diff --git a/flex/storages/rt_mutable_graph/modern_graph/bulk_load.yaml b/flex/storages/rt_mutable_graph/modern_graph/bulk_load.yaml
index b4e690f5bf49..8d9085d81aca 100644
--- a/flex/storages/rt_mutable_graph/modern_graph/bulk_load.yaml
+++ b/flex/storages/rt_mutable_graph/modern_graph/bulk_load.yaml
@@ -8,6 +8,15 @@ loading_config:
     type: csv
     metadata:
       delimiter: "|"  # other loading configuration places here
+      header_row: true # whether to use the first row as the header
+      quoting: false
+      quote_char: '"'
+      double_quote: true
+      escape_char: '\'
+      escaping: false
+      block_size: 4MB
+      batch_reader: true
+
 vertex_mappings:
   - type_name: person  # must align with the schema
     inputs:
@@ -31,7 +40,7 @@ vertex_mappings:
     column_mappings:
       - column:
           index: 0      # can be omitted if the index is the same as the property index
-          name: id    # can be omitted if the name is not known
+          name: id    # can be omitted if the name is not known or header_row is set to true.
         property: id  # must align with the schema
       - column:
           index: 1
diff --git a/flex/storages/rt_mutable_graph/mutable_csr.h b/flex/storages/rt_mutable_graph/mutable_csr.h
index 5fa6b245103a..2ef6569eeff5 100644
--- a/flex/storages/rt_mutable_graph/mutable_csr.h
+++ b/flex/storages/rt_mutable_graph/mutable_csr.h
@@ -703,6 +703,51 @@ class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
   mmap_array<nbr_t> nbr_list_;
 };
 
+template <typename EDATA_T>
+class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
+  using slice_t = MutableNbrSlice<EDATA_T>;
+
+ public:
+  EmptyCsr() = default;
+  ~EmptyCsr() = default;
+
+  void batch_init(vid_t vnum, const std::vector<int>& degree) override {}
+
+  slice_t get_edges(vid_t i) const override { return slice_t::empty(); }
+
+  void put_generic_edge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
+                        ArenaAllocator& alloc) override {}
+
+  void Serialize(const std::string& path) override {}
+
+  void Deserialize(const std::string& path) override {}
+
+  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
+                      timestamp_t ts = 0) override {}
+
+  void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
+                   ArenaAllocator& alloc) override {
+    EDATA_T value;
+    arc >> value;
+  }
+
+  void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
+                        const timestamp_t ts, ArenaAllocator& alloc) override {}
+
+  std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
+      vid_t v) const override {
+    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(
+        MutableNbrSlice<EDATA_T>::empty());
+  }
+  MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
+    return new TypedMutableCsrConstEdgeIter<EDATA_T>(
+        MutableNbrSlice<EDATA_T>::empty());
+  }
+  std::shared_ptr<MutableCsrEdgeIterBase> edge_iter_mut(vid_t v) override {
+    return std::make_shared<TypedMutableCsrEdgeIter<EDATA_T>>(
+        MutableNbrSliceMut<EDATA_T>::empty());
+  }
+};
 }  // 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 6312cb52f822..b4b6142ab7a8 100644
--- a/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
+++ b/flex/storages/rt_mutable_graph/mutable_property_fragment.cc
@@ -15,44 +15,10 @@
 
 #include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
 
-namespace gs {
-
-void preprocess_line(char* line) {
-  size_t len = strlen(line);
-  while (len >= 0) {
-    if (line[len] != '\0' && line[len] != '\n' && line[len] != '\r' &&
-        line[len] != ' ' && line[len] != '\t') {
-      break;
-    } else {
-      --len;
-    }
-  }
-  line[len + 1] = '\0';
-}
+#include "flex/engines/hqps_db/core/utils/hqps_utils.h"
+#include "flex/utils/property/types.h"
 
-void get_header_row(const std::string& file_name, std::vector<Any>& header) {
-  char line_buf[4096];
-  FILE* fin = fopen(file_name.c_str(), "r");
-  if (fgets(line_buf, 4096, fin) == NULL) {
-    LOG(FATAL) << "Failed to read header from file: " << file_name;
-    return;
-  }
-  preprocess_line(line_buf);
-  ParseRecord(line_buf, header);
-}
-
-std::vector<std::pair<size_t, std::string>> generate_default_column_mapping(
-    const std::string& file_name, std::string primary_key_name,
-    const std::vector<std::string>& column_names) {
-  std::vector<std::pair<size_t, std::string>> column_mapping;
-  for (size_t i = 0; i < column_names.size(); ++i) {
-    auto col_name = column_names[i];
-    if (col_name != primary_key_name) {
-      column_mapping.emplace_back(i, col_name);
-    }
-  }
-  return column_mapping;
-}
+namespace gs {
 
 MutablePropertyFragment::MutablePropertyFragment() {}
 
@@ -69,471 +35,6 @@ MutablePropertyFragment::~MutablePropertyFragment() {
   }
 }
 
-// vertex_column_mappings is a vector of pairs, each pair is (column_ind in
-// file, the cooresponding property name in schema).
-void MutablePropertyFragment::initVertices(
-    label_t v_label_i, const std::vector<std::string>& filenames,
-    const std::vector<std::pair<size_t, std::string>>& vertex_column_mappings) {
-  // Check primary key num and type.
-  auto primary_keys = schema_.get_vertex_primary_key(v_label_i);
-  if (primary_keys.size() != 1) {
-    LOG(FATAL) << "Only support one primary key for vertex.";
-  }
-  if (primary_keys[0].first != PropertyType::kInt64) {
-    LOG(FATAL) << "Only support int64_t primary key for vertex.";
-  }
-  IdIndexer<oid_t, vid_t> indexer;
-  std::string v_label_name = schema_.get_vertex_label_name(v_label_i);
-  VLOG(10) << "Start init vertices for label " << v_label_i << " with "
-           << filenames.size() << " files.";
-  auto& table = vertex_data_[v_label_i];
-  auto& property_types = schema_.get_vertex_properties(v_label_name);
-  auto& property_names = schema_.get_vertex_property_names(v_label_name);
-
-  // col num should be property num - 1, because one column will be used as
-  // primary key
-  CHECK(property_types.size() > 0);
-  size_t col_num = property_types.size();
-
-  // create real property_types_vec for table
-  VLOG(10) << "Init table for table: " << v_label_name
-           << ", with property num: " << col_num;
-  table.init(property_names, property_types,
-             schema_.get_vertex_storage_strategies(v_label_name),
-             schema_.get_max_vnum(v_label_name));
-  // Match the records read from the file with the schema
-  parseVertexFiles(v_label_name, filenames, vertex_column_mappings, indexer);
-  if (indexer.bucket_count() == 0) {
-    indexer._rehash(schema_.get_max_vnum(v_label_name));
-  }
-  build_lf_indexer(indexer, lf_indexers_[v_label_i]);
-}
-
-template <typename EDATA_T>
-class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
-  using slice_t = MutableNbrSlice<EDATA_T>;
-
- public:
-  EmptyCsr() = default;
-  ~EmptyCsr() = default;
-
-  void batch_init(vid_t vnum, const std::vector<int>& degree) override {}
-
-  slice_t get_edges(vid_t i) const override { return slice_t::empty(); }
-
-  void put_generic_edge(vid_t src, vid_t dst, const Any& data, timestamp_t ts,
-                        ArenaAllocator& alloc) override {}
-
-  void Serialize(const std::string& path) override {}
-
-  void Deserialize(const std::string& path) override {}
-
-  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
-                      timestamp_t ts = 0) override {}
-
-  void ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc, timestamp_t ts,
-                   ArenaAllocator& alloc) override {
-    EDATA_T value;
-    arc >> value;
-  }
-
-  void peek_ingest_edge(vid_t src, vid_t dst, grape::OutArchive& arc,
-                        const timestamp_t ts, ArenaAllocator& alloc) override {}
-
-  std::shared_ptr<MutableCsrConstEdgeIterBase> edge_iter(
-      vid_t v) const override {
-    return std::make_shared<TypedMutableCsrConstEdgeIter<EDATA_T>>(
-        MutableNbrSlice<EDATA_T>::empty());
-  }
-  MutableCsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
-    return new TypedMutableCsrConstEdgeIter<EDATA_T>(
-        MutableNbrSlice<EDATA_T>::empty());
-  }
-  std::shared_ptr<MutableCsrEdgeIterBase> edge_iter_mut(vid_t v) override {
-    return std::make_shared<TypedMutableCsrEdgeIter<EDATA_T>>(
-        MutableNbrSliceMut<EDATA_T>::empty());
-  }
-};
-
-template <typename EDATA_T>
-TypedMutableCsrBase<EDATA_T>* create_typed_csr(EdgeStrategy es) {
-  if (es == EdgeStrategy::kSingle) {
-    return new SingleMutableCsr<EDATA_T>();
-  } else if (es == EdgeStrategy::kMultiple) {
-    return new MutableCsr<EDATA_T>();
-  } else if (es == EdgeStrategy::kNone) {
-    return new EmptyCsr<EDATA_T>();
-  }
-  LOG(FATAL) << "not support edge strategy or edge data type";
-}
-
-template <typename EDATA_T>
-std::pair<MutableCsrBase*, MutableCsrBase*> construct_empty_csr(
-    EdgeStrategy ie_strategy, EdgeStrategy oe_strategy) {
-  TypedMutableCsrBase<EDATA_T>* ie_csr = create_typed_csr<EDATA_T>(ie_strategy);
-  TypedMutableCsrBase<EDATA_T>* oe_csr = create_typed_csr<EDATA_T>(oe_strategy);
-  ie_csr->batch_init(0, {});
-  oe_csr->batch_init(0, {});
-  return std::make_pair(ie_csr, oe_csr);
-}
-
-// each file name is tuple <src_column_ind, dst_column_id, edata(currently only
-// one edata)> src_column_id indicate which column is src id, dst_column_id
-// indicate which column is dst id default is 0, 1
-template <typename EDATA_T>
-std::pair<MutableCsrBase*, MutableCsrBase*> construct_csr(
-    const Schema& schema, const std::vector<std::string>& filenames,
-    size_t src_col_ind, size_t dst_col_ind,
-    const std::vector<PropertyType>& property_types,
-    const std::vector<std::pair<size_t, std::string>>& column_mappings,
-    EdgeStrategy ie_strategy, EdgeStrategy oe_strategy,
-    const LFIndexer<vid_t>& src_indexer, const LFIndexer<vid_t>& dst_indexer) {
-  TypedMutableCsrBase<EDATA_T>* ie_csr = create_typed_csr<EDATA_T>(ie_strategy);
-  TypedMutableCsrBase<EDATA_T>* oe_csr = create_typed_csr<EDATA_T>(oe_strategy);
-
-  std::vector<int> odegree(src_indexer.size(), 0);
-  std::vector<int> idegree(dst_indexer.size(), 0);
-
-  std::vector<std::tuple<vid_t, vid_t, EDATA_T>> parsed_edges;
-  vid_t src_index, dst_index;
-  char line_buf[4096];
-  oid_t src, dst;
-  EDATA_T data;
-
-  size_t col_num = property_types.size();
-  std::vector<Any> header(col_num + 2);
-  for (auto& item : header) {
-    item.type = PropertyType::kString;
-  }
-
-  // fetch header first
-  get_header_row(filenames[0], header);  // filenames must not be empty
-  // check header matches schema
-
-  for (auto filename : filenames) {
-    VLOG(10) << "processing " << filename << " with src_col_id " << src_col_ind
-             << " and dst_col_id " << dst_col_ind;
-    FILE* fin = fopen(filename.c_str(), "r");
-    if (fgets(line_buf, 4096, fin) == NULL) {
-      continue;
-    }
-    preprocess_line(line_buf);  // do nothing
-
-    // if match the default configuration, use ParseRecordX to fasten the
-    // parsing
-    if (src_col_ind == 0 && dst_col_ind == 1) {
-      while (fgets(line_buf, 4096, fin) != NULL) {
-        // ParseRecord src_id, dst_id, data from row.
-        ParseRecordX(line_buf, src, dst, data);
-        src_index = src_indexer.get_index(src);
-        dst_index = dst_indexer.get_index(dst);
-        ++idegree[dst_index];
-        ++odegree[src_index];
-        parsed_edges.emplace_back(src_index, dst_index, data);
-      }
-    } else {
-      std::vector<Any> row(col_num + 2);
-      CHECK(src_col_ind < row.size() && dst_col_ind < row.size());
-      row[src_col_ind].type = PropertyType::kInt64;
-      row[dst_col_ind].type = PropertyType::kInt64;
-      int32_t data_col_id = -1;
-      // the left column is must the edata.
-      for (auto i = 0; i < row.size(); ++i) {
-        if (row[i].type == PropertyType::kEmpty) {
-          // The index 0's type must exists
-          row[i].type = property_types[0];
-          data_col_id = i;
-          break;
-        }
-      }
-      CHECK(data_col_id != -1);
-      while (fgets(line_buf, 4096, fin) != NULL) {
-        // ParseRecord src_id, dst_id, data from row.
-        ParseRecord(line_buf, row);
-        src_index = src_indexer.get_index(row[src_col_ind].AsInt64());
-        dst_index = dst_indexer.get_index(row[dst_col_ind].AsInt64());
-        ConvertAny<EDATA_T>::to(row[data_col_id], data);
-        ++idegree[dst_index];
-        ++odegree[src_index];
-        parsed_edges.emplace_back(src_index, dst_index, data);
-      }
-    }
-
-    fclose(fin);
-  }
-
-  ie_csr->batch_init(dst_indexer.size(), idegree);
-  oe_csr->batch_init(src_indexer.size(), odegree);
-
-  for (auto& edge : parsed_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),
-                           std::get<2>(edge));
-  }
-
-  return std::make_pair(ie_csr, oe_csr);
-}
-
-void MutablePropertyFragment::initEdges(
-    label_t src_label_i, label_t dst_label_i, label_t edge_label_i,
-    const std::vector<std::string>& filenames,
-    const std::vector<std::pair<size_t, std::string>>& column_mappings,
-    size_t src_col_ind, size_t dst_col_ind) {
-  auto src_label_name = schema_.get_vertex_label_name(src_label_i);
-  auto dst_label_name = schema_.get_vertex_label_name(dst_label_i);
-  auto edge_label_name = schema_.get_edge_label_name(edge_label_i);
-  if (filenames.size() <= 0) {
-    LOG(FATAL) << "No edge files found for src label: " << src_label_name
-               << " dst label: " << dst_label_name
-               << " edge label: " << edge_label_name;
-  }
-  if (filenames.size() <= 0) {
-    LOG(FATAL) << "No edge files found for src label: " << src_label_name
-               << " dst label: " << dst_label_name
-               << " edge label: " << edge_label_name;
-  }
-  VLOG(10) << "Init edges src label: " << src_label_name
-           << " dst label: " << dst_label_name
-           << " edge label: " << edge_label_name
-           << " filenames: " << filenames.size();
-  auto& property_types = schema_.get_edge_properties(
-      src_label_name, dst_label_name, edge_label_name);
-  size_t col_num = property_types.size();
-  CHECK_LE(col_num, 1) << "Only single or no property is supported for edge.";
-
-  size_t index = src_label_i * vertex_label_num_ * edge_label_num_ +
-                 dst_label_i * edge_label_num_ + edge_label_i;
-  EdgeStrategy oe_strtagy = schema_.get_outgoing_edge_strategy(
-      src_label_name, dst_label_name, edge_label_name);
-  EdgeStrategy ie_strtagy = schema_.get_incoming_edge_strategy(
-      src_label_name, dst_label_name, edge_label_name);
-
-  {
-    // check column mappings consistent,
-    // TODO(zhanglei): Check column mappings after multiple property on edge is
-    // supported
-    if (column_mappings.size() > 1) {
-      LOG(FATAL) << "Edge column mapping must be less than 1";
-    }
-    if (column_mappings.size() > 0) {
-      auto& mapping = column_mappings[0];
-      if (mapping.first == src_col_ind || mapping.first == dst_col_ind) {
-        LOG(FATAL) << "Edge column mappings must not contain src_col_ind or "
-                      "dst_col_ind";
-      }
-      // check property exists in schema
-      if (!schema_.edge_has_property(src_label_name, dst_label_name,
-                                     edge_label_name, mapping.second)) {
-        LOG(FATAL) << "property " << mapping.second
-                   << " not exists in schema for edge triplet "
-                   << src_label_name << " -> " << edge_label_name << " -> "
-                   << dst_label_name;
-      }
-    }
-  }
-
-  if (col_num == 0) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<grape::EmptyType>(ie_strtagy, oe_strtagy);
-    } else {
-      std::tie(ie_[index], oe_[index]) = construct_csr<grape::EmptyType>(
-          schema_, filenames, src_col_ind, dst_col_ind, property_types,
-          column_mappings, ie_strtagy, oe_strtagy, lf_indexers_[src_label_i],
-          lf_indexers_[dst_label_i]);
-    }
-  } else if (property_types[0] == PropertyType::kDate) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<Date>(ie_strtagy, oe_strtagy);
-    } else {
-      std::tie(ie_[index], oe_[index]) = construct_csr<Date>(
-          schema_, filenames, src_col_ind, dst_col_ind, property_types,
-          column_mappings, ie_strtagy, oe_strtagy, lf_indexers_[src_label_i],
-          lf_indexers_[dst_label_i]);
-    }
-  } else if (property_types[0] == PropertyType::kInt32) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<int>(ie_strtagy, oe_strtagy);
-    } else {
-      std::tie(ie_[index], oe_[index]) = construct_csr<int>(
-          schema_, filenames, src_col_ind, dst_col_ind, property_types,
-          column_mappings, ie_strtagy, oe_strtagy, lf_indexers_[src_label_i],
-          lf_indexers_[dst_label_i]);
-    }
-  } else if (property_types[0] == PropertyType::kInt64) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<int64_t>(ie_strtagy, oe_strtagy);
-    } else {
-      std::tie(ie_[index], oe_[index]) = construct_csr<int64_t>(
-          schema_, filenames, src_col_ind, dst_col_ind, property_types,
-          column_mappings, ie_strtagy, oe_strtagy, lf_indexers_[src_label_i],
-          lf_indexers_[dst_label_i]);
-    }
-  } else if (property_types[0] == PropertyType::kString) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<std::string>(ie_strtagy, oe_strtagy);
-    } else {
-      LOG(FATAL) << "Unsupported edge property type.";
-    }
-  } else if (property_types[0] == PropertyType::kDouble) {
-    if (filenames.empty()) {
-      std::tie(ie_[index], oe_[index]) =
-          construct_empty_csr<double>(ie_strtagy, oe_strtagy);
-    } else {
-      std::tie(ie_[index], oe_[index]) = construct_csr<double>(
-          schema_, filenames, src_col_ind, dst_col_ind, property_types,
-          column_mappings, ie_strtagy, oe_strtagy, lf_indexers_[src_label_i],
-          lf_indexers_[dst_label_i]);
-    }
-  } else {
-    LOG(FATAL) << "Unsupported edge property type.";
-  }
-}
-
-void MutablePropertyFragment::Init(const Schema& schema,
-                                   const LoadingConfig& loading_config,
-                                   int thread_num) {
-  schema_ = schema;
-  vertex_label_num_ = schema_.vertex_label_num();
-  edge_label_num_ = schema_.edge_label_num();
-  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);
-  lf_indexers_.resize(vertex_label_num_);
-
-  auto& vertex_sources = loading_config.GetVertexLoadingMeta();
-  auto& edge_sources = loading_config.GetEdgeLoadingMeta();
-
-  if (thread_num == 1) {
-    if (vertex_sources.empty()) {
-      LOG(INFO) << "Skip loading vertices since no vertex source is specified.";
-    } else {
-      for (auto iter = vertex_sources.begin(); iter != vertex_sources.end();
-           ++iter) {
-        auto v_label_id = iter->first;
-        auto v_files = iter->second;
-        initVertices(v_label_id, v_files,
-                     loading_config.GetVertexColumnMappings(v_label_id));
-      }
-    }
-
-    if (edge_sources.empty()) {
-      LOG(INFO) << "Skip loading edges since no edge source is specified.";
-    } else {
-      LOG(INFO) << "Loading edges...";
-      for (auto iter = edge_sources.begin(); iter != edge_sources.end();
-           ++iter) {
-        // initEdges(iter->first, iter->second);
-        auto& src_label_id = std::get<0>(iter->first);
-        auto& dst_label_id = std::get<1>(iter->first);
-        auto& e_label_id = std::get<2>(iter->first);
-        auto& e_files = iter->second;
-        auto src_dst_col_pair = loading_config.GetEdgeSrcDstCol(
-            src_label_id, dst_label_id, e_label_id);
-        // We currenly only support one src primary key and one dst primary key
-        if (src_dst_col_pair.first.size() != 1 ||
-            src_dst_col_pair.second.size() != 1) {
-          LOG(FATAL) << "We currenly only support one src primary key and one "
-                        "dst primary key";
-        }
-        initEdges(src_label_id, dst_label_id, e_label_id, e_files,
-                  loading_config.GetEdgeColumnMappings(
-                      src_label_id, dst_label_id, e_label_id),
-                  src_dst_col_pair.first[0], src_dst_col_pair.second[0]);
-      }
-    }
-
-  } else {
-    // copy vertex_sources and edge sources to vector, since we need to
-    // use multi-thread loading.
-    std::vector<std::pair<label_t, std::vector<std::string>>> vertex_files;
-    for (auto iter = vertex_sources.begin(); iter != vertex_sources.end();
-         ++iter) {
-      vertex_files.emplace_back(iter->first, iter->second);
-    }
-    std::vector<std::pair<typename LoadingConfig::edge_triplet_type,
-                          std::vector<std::string>>>
-        edge_files;
-    for (auto iter = edge_sources.begin(); iter != edge_sources.end(); ++iter) {
-      edge_files.emplace_back(iter->first, iter->second);
-    }
-    LOG(INFO) << "Parallel loading with " << thread_num << " threads, "
-              << " " << vertex_files.size() << " vertex files, "
-              << edge_files.size() << " edge files.";
-    {
-      if (vertex_sources.empty()) {
-        LOG(INFO)
-            << "Skip loading vertices since no vertex source is specified.";
-      } else {
-        std::atomic<size_t> v_ind(0);
-        std::vector<std::thread> threads(thread_num);
-        for (int i = 0; i < thread_num; ++i) {
-          threads[i] = std::thread([&]() {
-            while (true) {
-              size_t cur = v_ind.fetch_add(1);
-              if (cur >= vertex_files.size()) {
-                break;
-              }
-              auto v_label_id = vertex_files[cur].first;
-              initVertices(v_label_id, vertex_files[cur].second,
-                           loading_config.GetVertexColumnMappings(v_label_id));
-            }
-          });
-        }
-        for (auto& thrd : threads) {
-          thrd.join();
-        }
-
-        LOG(INFO) << "finished loading vertices";
-      }
-    }
-    {
-      if (edge_sources.empty()) {
-        LOG(INFO) << "Skip loading edges since no edge source is specified.";
-      } else {
-        std::atomic<size_t> e_ind(0);
-        std::vector<std::thread> threads(thread_num);
-        for (int i = 0; i < thread_num; ++i) {
-          threads[i] = std::thread([&]() {
-            while (true) {
-              size_t cur = e_ind.fetch_add(1);
-              if (cur >= edge_files.size()) {
-                break;
-              }
-              auto& edge_file = edge_files[cur];
-              auto src_label_id = std::get<0>(edge_file.first);
-              auto dst_label_id = std::get<1>(edge_file.first);
-              auto e_label_id = std::get<2>(edge_file.first);
-              auto& file_names = edge_file.second;
-              auto src_dst_col_pair = loading_config.GetEdgeSrcDstCol(
-                  src_label_id, dst_label_id, e_label_id);
-              if (src_dst_col_pair.first.size() != 1 ||
-                  src_dst_col_pair.second.size() != 1) {
-                LOG(FATAL)
-                    << "We currenly only support one src primary key and one "
-                       "dst primary key";
-              }
-              initEdges(src_label_id, dst_label_id, e_label_id, file_names,
-                        loading_config.GetEdgeColumnMappings(
-                            src_label_id, dst_label_id, e_label_id),
-                        src_dst_col_pair.first[0], src_dst_col_pair.second[0]);
-            }
-          });
-        }
-        for (auto& thrd : threads) {
-          thrd.join();
-        }
-        LOG(INFO) << "finished loading edges";
-      }
-    }
-  }
-}
-
 void MutablePropertyFragment::IngestEdge(label_t src_label, vid_t src_lid,
                                          label_t dst_label, vid_t dst_lid,
                                          label_t edge_label, timestamp_t ts,
@@ -802,150 +303,4 @@ const MutableCsrBase* MutablePropertyFragment::get_ie_csr(
   return ie_[index];
 }
 
-void MutablePropertyFragment::parseVertexFiles(
-    const std::string& vertex_label, const std::vector<std::string>& filenames,
-    const std::vector<std::pair<size_t, std::string>>&
-        in_vertex_column_mappings,
-    IdIndexer<oid_t, vid_t>& indexer) {
-  if (filenames.empty()) {
-    return;
-  }
-  LOG(INFO) << "Parsing vertex files for label " << vertex_label;
-  auto vertex_column_mappings = in_vertex_column_mappings;
-
-  size_t label_index = schema_.get_vertex_label_id(vertex_label);
-  auto& table = vertex_data_[label_index];
-  auto& property_types = schema_.get_vertex_properties(vertex_label);
-  size_t col_num = property_types.size();
-  auto primary_key = schema_.get_vertex_primary_key(label_index)[0];
-  auto primary_key_name = primary_key.second;
-
-  // vertex_column_mappings can be empty, empty means the each column in the
-  // file is mapped to the same column in the table.
-  std::vector<Any> header(col_num + 1);
-  {
-    for (auto i = 0; i < header.size(); ++i) {
-      header[i].type = PropertyType::kString;
-    }
-  }
-  std::vector<Any> properties(col_num + 1);
-  std::vector<std::string> column_names(col_num + 1);
-  size_t primary_key_ind = col_num + 1;
-
-  // First get header
-  get_header_row(filenames[0], header);
-  // construct column_names
-  for (auto i = 0; i < header.size(); ++i) {
-    column_names[i] =
-        std::string(header[i].value.s.data(), header[i].value.s.size());
-  }
-
-  if (vertex_column_mappings.empty()) {
-    vertex_column_mappings = generate_default_column_mapping(
-        filenames[0], primary_key_name, column_names);
-    VLOG(10) << "vertex_column_mappings is empty, "
-                "generate_default_column_mapping returns "
-             << vertex_column_mappings.size() << " mappings";
-  }
-  for (auto i = 0; i < properties.size(); ++i) {
-    if (column_names[i] == primary_key_name) {
-      primary_key_ind = i;
-      break;
-    }
-    VLOG(10) << " compare: " << column_names[i] << " " << primary_key_name;
-  }
-  CHECK(primary_key_ind != col_num + 1);
-  {
-    // reset header of table with primary key removed
-    std::vector<std::string> header_col_names;
-    for (auto i = 0; i < column_names.size(); ++i) {
-      if (i != primary_key_ind) {
-        header_col_names.emplace_back(column_names[i]);
-      }
-    }
-    table.reset_header(header_col_names);
-    VLOG(10) << "reset header of table with primary key removed: "
-             << header_col_names.size();
-  }
-
-  for (auto i = 0; i < properties.size(); ++i) {
-    if (i < primary_key_ind) {
-      properties[i].type = property_types[i];
-    } else if (i > primary_key_ind) {
-      properties[i].type = property_types[i - 1];
-    } else {
-      properties[i].type = primary_key.first;
-    }
-  }
-
-  char line_buf[4096];
-  // we can't assume oid will be the first column.
-  oid_t oid;
-  vid_t v_index;
-
-  std::vector<int32_t> file_col_to_schema_col_ind;
-  {
-    // parse from vertex_column_mappings, vertex_column_mappings doesn't
-    // contains primary key.
-    size_t max_ind = 0;
-    for (auto& pair : vertex_column_mappings) {
-      max_ind = std::max(max_ind, pair.first);
-    }
-    file_col_to_schema_col_ind.resize(max_ind + 1, -1);
-    for (auto& pair : vertex_column_mappings) {
-      // if meet primary key, skip it.
-      if (pair.second == primary_key_name) {
-        VLOG(10) << "Skip primary key column " << pair.first << ", "
-                 << pair.second;
-        continue;
-      }
-      if (file_col_to_schema_col_ind[pair.first] == -1) {
-        if (schema_.vertex_has_property(vertex_label, pair.second)) {
-          auto& prop_names = schema_.get_vertex_property_names(vertex_label);
-          // find index of pair.second in prop_names
-          auto iter =
-              std::find(prop_names.begin(), prop_names.end(), pair.second);
-          // must be a valid iter.
-          if (iter == prop_names.end()) {
-            LOG(FATAL) << "Column " << pair.first << " is mapped to a column "
-                       << "that does not exist in schema: " << pair.second;
-          }
-          file_col_to_schema_col_ind[pair.first] =
-              std::distance(prop_names.begin(), iter);
-          VLOG(10) << "Column " << std::to_string(pair.first)
-                   << " is mapped to column " << pair.second << " in schema.: "
-                   << std::to_string(file_col_to_schema_col_ind[pair.first]);
-        } else {
-          LOG(FATAL) << "Column " << pair.first << " is mapped to a column "
-                     << "that does not exist in schema: " << pair.second;
-        }
-      } else {
-        LOG(FATAL) << "Column " << pair.first << " is mapped to multiple "
-                   << "columns in bulk loading file.";
-      }
-    }
-  }
-
-  for (auto filename : filenames) {
-    VLOG(10) << "Processing file: " << filename;
-    FILE* fin = fopen(filename.c_str(), "r");
-    // Just read first line, and do nothing, the header of file is not needed.
-    if (fgets(line_buf, 4096, fin) == NULL) {
-      continue;
-    }
-    preprocess_line(line_buf);
-    while (fgets(line_buf, 4096, fin) != NULL) {
-      preprocess_line(line_buf);
-      ParseRecord(line_buf, properties);
-      oid = properties[primary_key_ind].AsInt64();
-      if (indexer.add(oid, v_index)) {
-        // insert properties except for primary_key_ind
-        table.insert(v_index, properties, file_col_to_schema_col_ind);
-      }
-    }
-
-    fclose(fin);
-  }
-}
-
 }  // namespace gs
diff --git a/flex/storages/rt_mutable_graph/mutable_property_fragment.h b/flex/storages/rt_mutable_graph/mutable_property_fragment.h
index 67e34df41cdd..6fd2a7f4ae59 100644
--- a/flex/storages/rt_mutable_graph/mutable_property_fragment.h
+++ b/flex/storages/rt_mutable_graph/mutable_property_fragment.h
@@ -22,9 +22,9 @@
 
 #include "flex/storages/rt_mutable_graph/schema.h"
 
-#include "flex/storages/rt_mutable_graph/loading_config.h"
 #include "flex/storages/rt_mutable_graph/mutable_csr.h"
 #include "flex/storages/rt_mutable_graph/types.h"
+#include "flex/utils/arrow_utils.h"
 #include "flex/utils/id_indexer.h"
 #include "flex/utils/property/table.h"
 #include "flex/utils/yaml_utils.h"
@@ -39,9 +39,6 @@ class MutablePropertyFragment {
 
   ~MutablePropertyFragment();
 
-  void Init(const Schema& schema, const LoadingConfig& loading_config,
-            int thread_num = 1);
-
   void IngestEdge(label_t src_label, vid_t src_lid, label_t dst_label,
                   vid_t dst_lid, label_t edge_label, timestamp_t ts,
                   grape::OutArchive& arc, ArenaAllocator& alloc);
@@ -95,23 +92,6 @@ class MutablePropertyFragment {
   const MutableCsrBase* get_ie_csr(label_t label, label_t neighbor_label,
                                    label_t edge_label) const;
 
-  void parseVertexFiles(
-      const std::string& vertex_label,
-      const std::vector<std::string>& filenames,
-      const std::vector<std::pair<size_t, std::string>>& vertex_column_mappings,
-      IdIndexer<oid_t, vid_t>& indexer);
-
-  void initVertices(label_t v_label_i,
-                    const std::vector<std::string>& vertex_files,
-                    const std::vector<std::pair<size_t, std::string>>&
-                        vertex_column_mappings);
-
-  void initEdges(
-      label_t src_label_i, label_t dst_label_i, label_t edge_label_i,
-      const std::vector<std::string>& edge_files,
-      const std::vector<std::pair<size_t, std::string>>& edge_col_mappings,
-      size_t src_col_id, size_t dst_col_id);
-
   Schema schema_;
   std::vector<LFIndexer<vid_t>> lf_indexers_;
   std::vector<MutableCsrBase*> ie_, oe_;
diff --git a/flex/storages/rt_mutable_graph/schema.cc b/flex/storages/rt_mutable_graph/schema.cc
index 26e5bc7e8d1b..b42f344dd519 100644
--- a/flex/storages/rt_mutable_graph/schema.cc
+++ b/flex/storages/rt_mutable_graph/schema.cc
@@ -25,7 +25,8 @@ Schema::~Schema() = default;
 void Schema::add_vertex_label(
     const std::string& label, const std::vector<PropertyType>& property_types,
     const std::vector<std::string>& property_names,
-    const std::vector<std::pair<PropertyType, std::string>>& primary_key,
+    const std::vector<std::tuple<PropertyType, std::string, size_t>>&
+        primary_key,
     const std::vector<StorageStrategy>& strategies, size_t max_vnum) {
   label_t v_label_id = vertex_label_to_index(label);
   vproperties_[v_label_id] = property_types;
@@ -103,6 +104,12 @@ const std::vector<std::string>& Schema::get_vertex_property_names(
   return vprop_names_[index];
 }
 
+const std::vector<std::string>& Schema::get_vertex_property_names(
+    label_t label) const {
+  CHECK(label < vprop_names_.size());
+  return vprop_names_[label];
+}
+
 const std::vector<StorageStrategy>& Schema::get_vertex_storage_strategies(
     const std::string& label) const {
   label_t index;
@@ -137,6 +144,15 @@ const std::vector<PropertyType>& Schema::get_edge_properties(
   return eproperties_.at(index);
 }
 
+const std::vector<PropertyType>& Schema::get_edge_properties(
+    label_t src_label, label_t dst_label, label_t label) const {
+  CHECK(src_label < vlabel_indexer_.size());
+  CHECK(dst_label < vlabel_indexer_.size());
+  CHECK(label < elabel_indexer_.size());
+  uint32_t index = generate_edge_label(src_label, dst_label, label);
+  return eproperties_.at(index);
+}
+
 PropertyType Schema::get_edge_property(label_t src, label_t dst,
                                        label_t edge) const {
   uint32_t index = generate_edge_label(src, dst, edge);
@@ -154,6 +170,16 @@ const std::vector<std::string>& Schema::get_edge_property_names(
   return eprop_names_.at(index);
 }
 
+const std::vector<std::string>& Schema::get_edge_property_names(
+    const label_t& src_label, const label_t& dst_label,
+    const label_t& label) const {
+  CHECK(src_label < vlabel_indexer_.size());
+  CHECK(dst_label < vlabel_indexer_.size());
+  CHECK(label < elabel_indexer_.size());
+  uint32_t index = generate_edge_label(src_label, dst_label, label);
+  return eprop_names_.at(index);
+}
+
 bool Schema::valid_edge_property(const std::string& src_label,
                                  const std::string& dst_label,
                                  const std::string& label) const {
@@ -210,7 +236,7 @@ std::string Schema::get_edge_label_name(label_t index) const {
   return ret;
 }
 
-const std::vector<std::pair<PropertyType, std::string>>&
+const std::vector<std::tuple<PropertyType, std::string, size_t>>&
 Schema::get_vertex_primary_key(label_t index) const {
   CHECK(v_primary_keys_.size() > index);
   return v_primary_keys_.at(index);
@@ -511,7 +537,7 @@ static bool parse_vertex_schema(YAML::Node node, Schema& schema) {
   // remove primary key from properties.
 
   std::vector<int> primary_key_inds(primary_key_node.size(), -1);
-  std::vector<std::pair<PropertyType, std::string>> primary_keys;
+  std::vector<std::tuple<PropertyType, std::string, size_t>> primary_keys;
   for (auto i = 0; i < primary_key_node.size(); ++i) {
     auto cur_primary_key = primary_key_node[i];
     std::string primary_key_name = primary_key_node[0].as<std::string>();
@@ -531,7 +557,8 @@ static bool parse_vertex_schema(YAML::Node node, Schema& schema) {
       return false;
     }
     primary_keys.emplace_back(property_types[primary_key_inds[i]],
-                              property_names[primary_key_inds[i]]);
+                              property_names[primary_key_inds[i]],
+                              primary_key_inds[i]);
     // remove primary key from properties.
     property_names.erase(property_names.begin() + primary_key_inds[i]);
     property_types.erase(property_types.begin() + primary_key_inds[i]);
@@ -738,7 +765,7 @@ bool Schema::vertex_has_primary_key(const std::string& label,
   CHECK(v_label_id < vprop_names_.size());
   auto& keys = v_primary_keys_[v_label_id];
   for (auto i = 0; i < keys.size(); ++i) {
-    if (keys[i].second == prop) {
+    if (std::get<1>(keys[i]) == prop) {
       return true;
     }
   }
diff --git a/flex/storages/rt_mutable_graph/schema.h b/flex/storages/rt_mutable_graph/schema.h
index fe217b3a2bc7..c0a017088d79 100644
--- a/flex/storages/rt_mutable_graph/schema.h
+++ b/flex/storages/rt_mutable_graph/schema.h
@@ -23,11 +23,6 @@
 
 namespace gs {
 
-static constexpr const char* DT_SIGNED_INT32 = "DT_SIGNED_INT32";
-static constexpr const char* DT_STRING = "DT_STRING";
-static constexpr const char* DT_SIGNED_INT64 = "DT_SIGNED_INT64";
-static constexpr const char* DT_DOUBLE = "DT_DOUBLE";
-
 class Schema {
  public:
   using label_type = label_t;
@@ -37,7 +32,8 @@ class Schema {
   void add_vertex_label(
       const std::string& label, const std::vector<PropertyType>& property_types,
       const std::vector<std::string>& property_names,
-      const std::vector<std::pair<PropertyType, std::string>>& primary_key,
+      const std::vector<std::tuple<PropertyType, std::string, size_t>>&
+          primary_key,
       const std::vector<StorageStrategy>& strategies = {},
       size_t max_vnum = static_cast<size_t>(1) << 32);
 
@@ -69,6 +65,9 @@ class Schema {
 
   const std::vector<PropertyType>& get_vertex_properties(label_t label) const;
 
+  const std::vector<std::string>& get_vertex_property_names(
+      label_t label) const;
+
   const std::vector<StorageStrategy>& get_vertex_storage_strategies(
       const std::string& label) const;
 
@@ -81,12 +80,20 @@ class Schema {
       const std::string& src_label, const std::string& dst_label,
       const std::string& label) const;
 
+  const std::vector<PropertyType>& get_edge_properties(label_t src_label,
+                                                       label_t dst_label,
+                                                       label_t label) const;
+
   PropertyType get_edge_property(label_t src, label_t dst, label_t edge) const;
 
   const std::vector<std::string>& get_edge_property_names(
       const std::string& src_label, const std::string& dst_label,
       const std::string& label) const;
 
+  const std::vector<std::string>& get_edge_property_names(
+      const label_t& src_label, const label_t& dst_label,
+      const label_t& label) const;
+
   bool vertex_has_property(const std::string& label,
                            const std::string& prop) const;
 
@@ -124,9 +131,11 @@ class Schema {
 
   std::string get_edge_label_name(label_t index) const;
 
-  const std::vector<std::pair<PropertyType, std::string>>&
+  const std::vector<std::tuple<PropertyType, std::string, size_t>>&
   get_vertex_primary_key(label_t index) const;
 
+  const std::string& get_vertex_primary_key_name(label_t index) const;
+
   void Serialize(std::unique_ptr<grape::LocalIOAdaptor>& writer);
 
   void Deserialize(std::unique_ptr<grape::LocalIOAdaptor>& reader);
@@ -150,8 +159,9 @@ class Schema {
   IdIndexer<std::string, label_t> elabel_indexer_;
   std::vector<std::vector<PropertyType>> vproperties_;
   std::vector<std::vector<std::string>> vprop_names_;
-  std::vector<std::vector<std::pair<PropertyType, std::string>>>
-      v_primary_keys_;
+  std::vector<std::vector<std::tuple<PropertyType, std::string, size_t>>>
+      v_primary_keys_;  // the third element is the index of the property in the
+                        // vertex property list
   std::vector<std::vector<StorageStrategy>> vprop_storage_;
   std::map<uint32_t, std::vector<PropertyType>> eproperties_;
   std::map<uint32_t, std::vector<std::string>> eprop_names_;
diff --git a/flex/storages/rt_mutable_graph/types.h b/flex/storages/rt_mutable_graph/types.h
index 3b9700b4c772..f73c7cc2fcd3 100644
--- a/flex/storages/rt_mutable_graph/types.h
+++ b/flex/storages/rt_mutable_graph/types.h
@@ -31,6 +31,11 @@ using vid_t = uint32_t;
 using oid_t = int64_t;
 using label_t = uint8_t;
 
+static constexpr const char* DT_SIGNED_INT32 = "DT_SIGNED_INT32";
+static constexpr const char* DT_SIGNED_INT64 = "DT_SIGNED_INT64";
+static constexpr const char* DT_DOUBLE = "DT_DOUBLE";
+static constexpr const char* DT_STRING = "DT_STRING";
+
 }  // namespace gs
 
 #endif  // STORAGES_RT_MUTABLE_GRAPH_TYPES_H_
diff --git a/flex/utils/arrow_utils.cc b/flex/utils/arrow_utils.cc
new file mode 100644
index 000000000000..50356db839eb
--- /dev/null
+++ b/flex/utils/arrow_utils.cc
@@ -0,0 +1,50 @@
+/** 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 "flex/utils/arrow_utils.h"
+
+namespace gs {
+std::shared_ptr<arrow::DataType> PropertyTypeToArrowType(PropertyType type) {
+  switch (type) {
+  case PropertyType::kInt32:
+    return arrow::int32();
+  case PropertyType::kInt64:
+    return arrow::int64();
+  case PropertyType::kDouble:
+    return arrow::float64();
+  case PropertyType::kDate:
+    return arrow::timestamp(arrow::TimeUnit::MILLI);
+  case PropertyType::kString:
+    return arrow::large_utf8();
+  case PropertyType::kEmpty:
+    return arrow::null();
+  default:
+    LOG(FATAL) << "Unexpected property type: " << static_cast<int>(type);
+    return nullptr;
+  }
+}
+
+template <typename T>
+void emplace_into_vector(const std::shared_ptr<arrow::ChunkedArray>& array,
+                         std::vector<Any>& vec) {
+  using arrow_array_type = typename gs::CppTypeToArrowType<T>::ArrayType;
+  for (auto i = 0; i < array->num_chunks(); ++i) {
+    auto casted = std::static_pointer_cast<arrow_array_type>(array->chunk(i));
+    for (auto k = 0; k < casted->length(); ++k) {
+      vec.emplace_back(AnyConverter<T>::to_any(casted->Value(k)));
+    }
+  }
+}
+
+}  // namespace gs
diff --git a/flex/utils/arrow_utils.h b/flex/utils/arrow_utils.h
new file mode 100644
index 000000000000..d022ce56dbcc
--- /dev/null
+++ b/flex/utils/arrow_utils.h
@@ -0,0 +1,91 @@
+/** 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 UTILS_ARROW_UTILS_H_
+#define UTILS_ARROW_UTILS_H_
+
+#include <arrow/api.h>
+#include <memory>
+#include "flex/utils/property/types.h"
+
+namespace gs {
+
+// arrow related;
+
+// convert c++ type to arrow type. support other types likes emptyType, Date
+template <typename T>
+struct CppTypeToArrowType {};
+
+template <>
+struct CppTypeToArrowType<int64_t> {
+  using Type = arrow::Int64Type;
+  using ArrayType = arrow::Int64Array;
+  static std::shared_ptr<arrow::DataType> TypeValue() { return arrow::int64(); }
+};
+
+template <>
+struct CppTypeToArrowType<int32_t> {
+  using Type = arrow::Int32Type;
+  using ArrayType = arrow::Int32Array;
+  static std::shared_ptr<arrow::DataType> TypeValue() { return arrow::int32(); }
+};
+
+template <>
+struct CppTypeToArrowType<double> {
+  using Type = arrow::DoubleType;
+  using ArrayType = arrow::DoubleArray;
+  static std::shared_ptr<arrow::DataType> TypeValue() {
+    return arrow::float64();
+  }
+};
+
+template <>
+struct CppTypeToArrowType<Date> {
+  using Type = arrow::Int64Type;
+  using ArrayType = arrow::Int64Array;
+  static std::shared_ptr<arrow::DataType> TypeValue() { return arrow::int64(); }
+};
+
+template <typename T>
+struct CppTypeToPropertyType;
+
+template <>
+struct CppTypeToPropertyType<int32_t> {
+  static constexpr PropertyType value = PropertyType::kInt32;
+};
+
+template <>
+struct CppTypeToPropertyType<int64_t> {
+  static constexpr PropertyType value = PropertyType::kInt64;
+};
+
+template <>
+struct CppTypeToPropertyType<double> {
+  static constexpr PropertyType value = PropertyType::kDouble;
+};
+
+template <>
+struct CppTypeToPropertyType<std::string> {
+  static constexpr PropertyType value = PropertyType::kString;
+};
+
+template <>
+struct CppTypeToPropertyType<std::string_view> {
+  static constexpr PropertyType value = PropertyType::kString;
+};
+
+std::shared_ptr<arrow::DataType> PropertyTypeToArrowType(PropertyType type);
+}  // namespace gs
+
+#endif  // UTILS_ARROW_UTILS_H_
diff --git a/flex/utils/property/table.cc b/flex/utils/property/table.cc
index ec60f06e42c1..20a2aa836f75 100644
--- a/flex/utils/property/table.cc
+++ b/flex/utils/property/table.cc
@@ -134,6 +134,8 @@ const std::shared_ptr<ColumnBase> Table::get_column_by_id(size_t index) const {
 
 size_t Table::col_num() const { return columns_.size(); }
 std::vector<std::shared_ptr<ColumnBase>>& Table::columns() { return columns_; }
+// get column pointers
+std::vector<ColumnBase*>& Table::column_ptrs() { return column_ptrs_; }
 
 void Table::insert(size_t index, const std::vector<Any>& values) {
   assert(values.size() == columns_.size());
diff --git a/flex/utils/property/table.h b/flex/utils/property/table.h
index 5f5b1eca4aeb..f6f05aa5248a 100644
--- a/flex/utils/property/table.h
+++ b/flex/utils/property/table.h
@@ -59,6 +59,7 @@ class Table {
 
   size_t col_num() const;
   std::vector<std::shared_ptr<ColumnBase>>& columns();
+  std::vector<ColumnBase*>& column_ptrs();
 
   void insert(size_t index, const std::vector<Any>& values);
 
diff --git a/flex/utils/property/types.cc b/flex/utils/property/types.cc
index 461dcd165238..1412378bde84 100644
--- a/flex/utils/property/types.cc
+++ b/flex/utils/property/types.cc
@@ -20,97 +20,6 @@
 
 namespace gs {
 
-inline void ParseInt32(const std::string_view& str, int& val) {
-  sscanf(str.data(), "%d", &val);
-}
-
-inline void ParseInt64(const std::string_view& str, int64_t& val) {
-#ifdef __APPLE__
-  sscanf(str.data(), "%lld", &val);
-#else
-  sscanf(str.data(), "%" SCNd64, &val);
-#endif
-}
-
-inline void ParseDate(const std::string_view& str, Date& date) {
-  date.reset(str.data());
-}
-
-inline void ParseString(const std::string_view& str, std::string_view& val) {
-  val = str;
-}
-
-inline void ParseDouble(const std::string_view& str, double& val) {
-  sscanf(str.data(), "%lf", &val);
-}
-
-void ParseRecord(const char* line, std::vector<Any>& rec) {
-  const char* cur = line;
-  for (auto& item : rec) {
-    const char* ptr = cur;
-    while (*ptr != '\0' && *ptr != '|') {
-      ++ptr;
-    }
-    std::string_view sv(cur, ptr - cur);
-    if (item.type == PropertyType::kInt32) {
-      ParseInt32(sv, item.value.i);
-    } else if (item.type == PropertyType::kInt64) {
-      ParseInt64(sv, item.value.l);
-    } else if (item.type == PropertyType::kDate) {
-      ParseDate(sv, item.value.d);
-    } else if (item.type == PropertyType::kString) {
-      ParseString(sv, item.value.s);
-    } else if (item.type == PropertyType::kDouble) {
-      ParseDouble(sv, item.value.db);
-    }
-    cur = ptr + 1;
-  }
-}
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, int& prop) {
-#ifdef __APPLE__
-  sscanf(line, "%lld|%lld|%d", &src, &dst, &prop);
-#else
-  sscanf(line, "%" SCNd64 "|%" SCNd64 "|%d", &src, &dst, &prop);
-#endif
-}
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, Date& prop) {
-#ifdef __APPLE__
-  sscanf(line, "%lld|%lld", &src, &dst);
-#else
-  sscanf(line, "%" SCNd64 "|%" SCNd64 "", &src, &dst);
-#endif
-  const char* ptr = strrchr(line, '|') + 1;
-  prop.reset(ptr);
-}
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst,
-                  grape::EmptyType& prop) {
-#ifdef __APPLE__
-  sscanf(line, "%lld|%lld", &src, &dst);
-#else
-  sscanf(line, "%" SCNd64 "|%" SCNd64 "", &src, &dst);
-#endif
-}
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, double& prop) {
-#ifdef __APPLE__
-  sscanf(line, "%lld|%lld|%lf", &src, &dst, &prop);
-#else
-  sscanf(line, "%" SCNd64 "|%" SCNd64 "|%lf", &src, &dst, &prop);
-#endif
-}
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, int64_t& prop) {
-#ifdef __APPLE__
-  // parseRecordX for edge with int64 property
-  sscanf(line, "%lld|%lld|%lld", &src, &dst, &prop);
-#else
-  sscanf(line, "%" SCNd64 "|%" SCNd64 "|%" SCNd64 "", &src, &dst, &prop);
-#endif
-}
-
 grape::InArchive& operator<<(grape::InArchive& in_archive, const Any& value) {
   switch (value.type) {
   case PropertyType::kInt32:
@@ -177,67 +86,7 @@ grape::OutArchive& operator>>(grape::OutArchive& out_archive,
   return out_archive;
 }
 
-// date format:
-// YYYY-MM-DD'T'hh:mm:ss.SSSZZZZ
-// 2010-04-25T05:45:11.772+0000
-
-inline static uint32_t char_to_digit(char c) { return (c - '0'); }
-
-inline static uint32_t str_4_to_number(const char* str) {
-  return char_to_digit(str[0]) * 1000u + char_to_digit(str[1]) * 100u +
-         char_to_digit(str[2]) * 10u + char_to_digit(str[3]);
-}
-
-inline static uint32_t str_3_to_number(const char* str) {
-  return char_to_digit(str[0]) * 100u + char_to_digit(str[1]) * 10u +
-         char_to_digit(str[2]);
-}
-
-inline static uint32_t str_2_to_number(const char* str) {
-  return char_to_digit(str[0]) * 10u + char_to_digit(str[1]);
-}
-
 Date::Date(int64_t x) : milli_second(x) {}
-Date::Date(const char* str) { reset(str); }
-
-void Date::reset(const char* str) {
-  if (str[4] == '-') {
-    struct tm v;
-    memset(&v, 0, sizeof(v));
-    v.tm_year = str_4_to_number(str) - 1900;
-    v.tm_mon = str_2_to_number(&str[5]) - 1;
-    v.tm_mday = str_2_to_number(&str[8]);
-    if (str[10] == '|') {
-      milli_second = mktime(&v);
-      milli_second *= 1000l;
-      milli_second += 8 * 60 * 60 * 1000l;
-      return;
-    }
-    v.tm_hour = str_2_to_number(&str[11]);
-    v.tm_min = str_2_to_number(&str[14]);
-    v.tm_sec = str_2_to_number(&str[17]);
-
-    milli_second = (mktime(&v));
-
-    milli_second *= 1000l;
-    milli_second += str_3_to_number(&str[20]);
-    bool zone_flag = (str[23] == '+') ? 1u : 0u;
-    uint32_t zone_hour = str_2_to_number(&str[24]);
-    uint32_t zone_minute = str_2_to_number(&str[26]);
-    milli_second += 8 * 60 * 60 * 1000l;
-    if (zone_flag) {
-      milli_second += (zone_hour * 60 * 60l + zone_minute * 60l) * 1000l;
-    } else {
-      milli_second -= (zone_hour * 60 * 60l + zone_minute * 60l) * 1000l;
-    }
-  } else {
-#ifdef __APPLE__
-    sscanf(str, "%lld", &milli_second);
-#else
-    sscanf(str, "%" SCNd64, &milli_second);
-#endif
-  }
-}
 
 std::string Date::to_string() const { return std::to_string(milli_second); }
 
diff --git a/flex/utils/property/types.h b/flex/utils/property/types.h
index eda3841e5d11..aaac320a767f 100644
--- a/flex/utils/property/types.h
+++ b/flex/utils/property/types.h
@@ -45,9 +45,7 @@ struct Date {
   Date() = default;
   ~Date() = default;
   Date(int64_t x);
-  Date(const char* str);
 
-  void reset(const char* str);
   std::string to_string() const;
 
   int64_t milli_second;
@@ -392,18 +390,6 @@ struct AnyConverter<double> {
   }
 };
 
-void ParseRecord(const char* line, std::vector<Any>& rec);
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, int& prop);
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, Date& prop);
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst,
-                  grape::EmptyType& prop);
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, double& prop);
-
-void ParseRecordX(const char* line, int64_t& src, int64_t& dst, int64_t& prop);
-
 grape::InArchive& operator<<(grape::InArchive& in_archive, const Any& value);
 grape::OutArchive& operator>>(grape::OutArchive& out_archive, Any& value);
 


Test patch:
diff --git a/flex/tests/hqps/hqps_cypher_test.sh b/flex/tests/hqps/hqps_cypher_test.sh
index 4de7a77e0805..5696a898f350 100644
--- a/flex/tests/hqps/hqps_cypher_test.sh
+++ b/flex/tests/hqps/hqps_cypher_test.sh
@@ -76,6 +76,9 @@ kill_service(){
     info "Kill Service success"
 }
 
+# kill service when exit
+trap kill_service EXIT
+
 create_ir_conf(){
     rm ${HQPS_IR_CONF} || true
     echo "engine.type: hiactor" >> ${HQPS_IR_CONF}
diff --git a/flex/tests/rt_mutable_graph/test_graph_loading.cc b/flex/tests/rt_mutable_graph/test_graph_loading.cc
index 5c3d0ed36609..e0280a836b03 100644
--- a/flex/tests/rt_mutable_graph/test_graph_loading.cc
+++ b/flex/tests/rt_mutable_graph/test_graph_loading.cc
@@ -57,7 +57,7 @@ int main(int argc, char** argv) {
   }
 
   for (auto i = 0; i < graph.schema().edge_label_num(); ++i) {
-    LOG(INFO) << "vertex label " << i
+    LOG(INFO) << "edge label " << i
               << " name: " << graph.schema().get_edge_label_name(i);
   }