Repository: alibaba/GraphScope (language: python)

Issue:
feat(interactive): Support non-blocking graph dataloading 
Currently graph dataloading is a blocking operation. By wrapping it as a job, we can support non-block graph dataloading.
[BUG] Interactive: Primary key id should be int64 or string
**Describe the bug**

**Create a graph with `DT_DOUBLE` as primary key type**
Status code return 200(successfully created graph) but with `Primary key id should be int64 or string` error message during creating graph process.

![截屏2024-04-08 10 49 32](https://github.com/alibaba/GraphScope/assets/23491543/df8a370b-99ad-4e6a-a8ff-766325f73ed5)



Discussion / hints:

Fix patch:
diff --git a/docs/flex/interactive/development/admin_service.md b/docs/flex/interactive/development/admin_service.md
index 21e8faf31beb..ed9eb114cea8 100644
--- a/docs/flex/interactive/development/admin_service.md
+++ b/docs/flex/interactive/development/admin_service.md
@@ -9,23 +9,26 @@ Welcome to the GraphScope Interactive Admin Service documentation. This guide is
 The table below provides an overview of the available APIs:
 
 
-| API name        | Method and URL                        | Explanation                                                        |
-|-----------------|---------------------------------------|--------------------------------------------------------------------|
-| ListGraphs      | GET /v1/graph                             | Get all graphs in current interactive service, the schema for each graph is returned. |
-| GetGraphSchema  | GET /v1/graph/{graph}/schema          | Get the schema for the specified graph.                            |
-| CreateGraph     | POST /v1/graph                             | Create an empty graph with the specified schema.                    |
-| DeleteGraph     | DELETE /v1/graph/{graph}                 | Delete the specified graph.                                        |
-| ImportGraph     | POST /v1/graph/{graph}/dataloading     | Import data to graph.                                              |
-| CreateProcedure | POST /v1/graph/{graph}/procedure      | Create a new stored procedure bound to a graph.                    |
-| ShowProcedures  | GET /v1/graph/{graph}/procedure      | Get all procedures bound to the specified graph.                   |
-| GetProcedure    | GET /v1/graph/{graph}/procedure/{proc_name} | Get the metadata of the procedure.                                 |
-| DeleteProcedure | DELETE /v1/graph/{graph}/procedure/{proc_name} | Delete the specified procedure.                                   |
-| UpdateProcedure | PUT /v1/graph/{graph}/procedure/{proc_name} | Update some metadata for the specified procedure, i.e. update description, enable/disable. |
-| StartService   | POST /v1/service/start                     | Start the query service on the graph specified in request body.          |
+| API name        | Method and URL                                 | Explanation                                                        |
+|-----------------|------------------------------------------------|--------------------------------------------------------------------|
+| ListGraphs      | GET /v1/graph                                  | Get all graphs in current interactive service, the schema for each graph is returned. |
+| GetGraphSchema  | GET /v1/graph/{graph}/schema                   | Get the schema for the specified graph.                            |
+| CreateGraph     | POST /v1/graph                                 | Create an empty graph with the specified schema.                   |
+| DeleteGraph     | DELETE /v1/graph/{graph}                       | Delete the specified graph.                                        |
+| ImportGraph     | POST /v1/graph/{graph}/dataloading             | Import data to graph.                                               |
+| CreateProcedure | POST /v1/graph/{graph}/procedure               | Create a new stored procedure bound to a graph.                    |
+| ShowProcedures  | GET /v1/graph/{graph}/procedure                | Get all procedures bound to the specified graph.                   |
+| GetProcedure    | GET /v1/graph/{graph}/procedure/{proc_name}    | Get the metadata of the procedure.                                 |
+| DeleteProcedure | DELETE /v1/graph/{graph}/procedure/{proc_name} | Delete the specified procedure.                                    |
+| UpdateProcedure | PUT /v1/graph/{graph}/procedure/{proc_name}    | Update some metadata for the specified procedure, i.e. update description, enable/disable. |
+| StartService    | POST /v1/service/start                         | Start the service on the graph specified in request body.          |
+| RestartService    | POST /v1/service/restart                         | Restart the service on the current running graph.         |
 | StopService    | GET /v1/service/stop                     | Stop the query service                                            |
-| ServiceStatus  | GET /v1/service/status                    | Get current service status.                                        |
-| SystemMetrics     | GET /v1/node/status                       | Get the system metrics of current host/pod, i.e. CPU usage, memory usages.                    |
-
+| ServiceStatus   | GET /v1/service/status                         | Get current service status.                                        |
+| SystemMetrics   | GET /v1/node/status                            | Get the system metrics of current host/pod, i.e. CPU usage, memory usages.                    |
+| GetJobById      | GET /v1/job/{job_id}                           | Get the metadata for the specified job.               |
+| GetAllJobs      | GET /v1/job/                                   | Get all jobs's metadata.                | 
+| CancelJob       | DELETE /v1/job/{job_id}                        | Cancel the job with the specified job_id.                           |
 
 ## Detailed API Documentation
 
@@ -318,13 +321,13 @@ Delete a graph by name, including schema, indices and stored procedures.
 
 #### HTTP Request
 - **Method**: DELETE
-- **Endpoint**: `/v1/graph/{graph_name}`
+- **Endpoint**: `/v1/graph/{graph_id}`
 - **Content-type**: `application/json`
 
 
 #### Curl Command Example
 ```bash
-curl -X DELETE  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_name}"
+curl -X DELETE  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_id}"
 ```
 
 #### Expected Response
@@ -347,12 +350,12 @@ Get the schema for the specified graph.
 
 #### HTTP Request
 - **Method**: GET
-- **Endpoint**: `/v1/graph/{graph_name}`
+- **Endpoint**: `/v1/graph/{graph_id}`
 - **Content-type**: `application/json`
 
 #### Curl Command Example
 ```bash
-curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_name}"
+curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_id}"
 ```
 
 
@@ -361,113 +364,110 @@ curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_
 - **Body**:
 ```json
 {
-    "name": "modern",
-    "schema": {
-        "vertex_types": [
-            {
-                "type_id": 0,
-                "type_name": "person",
-                "properties": [
-                    {
-                        "property_id": 0,
-                        "property_name": "id",
-                        "property_type": {
-                            "primitive_type": "DT_SIGNED_INT64"
-                        }
-                    },
-                    {
-                        "property_id": 1,
-                        "property_name": "name",
-                        "property_type": {
-                            "primitive_type": "DT_STRING"
-                        }
-                    },
-                    {
-                        "property_id": 2,
-                        "property_name": "age",
-                        "property_type": {
-                            "primitive_type": "DT_SIGNED_INT32"
-                        }
-                    }
-                ],
-                "primary_keys": [
-                    "id"
-                ]
-            },
-            {
-                "type_id": 1,
-                "type_name": "software",
-                "properties": [
-                    {
-                        "property_id": 0,
-                        "property_name": "id",
-                        "property_type": {
-                            "primitive_type": "DT_SIGNED_INT64"
-                        }
-                    },
-                    {
-                        "property_id": 1,
-                        "property_name": "name",
-                        "property_type": {
-                            "primitive_type": "DT_STRING"
-                        }
-                    },
-                    {
-                        "property_id": 2,
-                        "property_name": "lang",
-                        "property_type": {
-                            "primitive_type": "DT_STRING"
-                        }
-                    }
-                ],
-                "primary_keys": [
-                    "id"
-                ]
-            }
-        ],
-        "edge_types": [
-            {
-                "type_id": 0,
-                "type_name": "knows",
-                "vertex_type_pair_relations": [
-                    {
-                        "source_vertex": "person",
-                        "destination_vertex": "person",
-                        "relation": "MANY_TO_MANY",
-                    }
-                ],
-                "properties": [
-                    {
-                        "property_id": 0,
-                        "property_name": "weight",
-                        "property_type": {
-                            "primitive_type": "DT_DOUBLE"
-                        }
-                    }
-                ]
-            },
-            {
-                "type_id": 1,
-                "type_name": "created",
-                "vertex_type_pair_relations": [
-                    {
-                        "source_vertex": "person",
-                        "destination_vertex": "software",
-                        "relation": "ONE_TO_MANY",
-                    }
-                ],
-                "properties": [
-                    {
-                        "property_id": 0,
-                        "property_name": "weight",
-                        "property_type": {
-                            "primitive_type": "DT_DOUBLE"
-                        }
-                    }
-                ]
-            }
-        ]
-    }
+  "vertex_types": [
+      {
+          "type_id": 0,
+          "type_name": "person",
+          "properties": [
+              {
+                  "property_id": 0,
+                  "property_name": "id",
+                  "property_type": {
+                      "primitive_type": "DT_SIGNED_INT64"
+                  }
+              },
+              {
+                  "property_id": 1,
+                  "property_name": "name",
+                  "property_type": {
+                      "primitive_type": "DT_STRING"
+                  }
+              },
+              {
+                  "property_id": 2,
+                  "property_name": "age",
+                  "property_type": {
+                      "primitive_type": "DT_SIGNED_INT32"
+                  }
+              }
+          ],
+          "primary_keys": [
+              "id"
+          ]
+      },
+      {
+          "type_id": 1,
+          "type_name": "software",
+          "properties": [
+              {
+                  "property_id": 0,
+                  "property_name": "id",
+                  "property_type": {
+                      "primitive_type": "DT_SIGNED_INT64"
+                  }
+              },
+              {
+                  "property_id": 1,
+                  "property_name": "name",
+                  "property_type": {
+                      "primitive_type": "DT_STRING"
+                  }
+              },
+              {
+                  "property_id": 2,
+                  "property_name": "lang",
+                  "property_type": {
+                      "primitive_type": "DT_STRING"
+                  }
+              }
+          ],
+          "primary_keys": [
+              "id"
+          ]
+      }
+  ],
+  "edge_types": [
+      {
+          "type_id": 0,
+          "type_name": "knows",
+          "vertex_type_pair_relations": [
+              {
+                  "source_vertex": "person",
+                  "destination_vertex": "person",
+                  "relation": "MANY_TO_MANY",
+              }
+          ],
+          "properties": [
+              {
+                  "property_id": 0,
+                  "property_name": "weight",
+                  "property_type": {
+                      "primitive_type": "DT_DOUBLE"
+                  }
+              }
+          ]
+      },
+      {
+          "type_id": 1,
+          "type_name": "created",
+          "vertex_type_pair_relations": [
+              {
+                  "source_vertex": "person",
+                  "destination_vertex": "software",
+                  "relation": "ONE_TO_MANY",
+              }
+          ],
+          "properties": [
+              {
+                  "property_id": 0,
+                  "property_name": "weight",
+                  "property_type": {
+                      "primitive_type": "DT_DOUBLE"
+                  }
+              }
+          ]
+      }
+  ]
 }
 ```
 
@@ -480,12 +480,14 @@ curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_
 
 #### Description
 
-Import data to empty graph.
+Import data to empty graph. This is a non-blocking API as it will directly return the job id after the bulk loading job is created.
+Client can use the returned job_id to [query the status of job](#getjobbyid-jobmanagement-category) or [cancel a runing job](#canceljobbyid-jobmanagement-category).
+
 
 #### HTTP Request
 
 - **Method**: POST
-- **Endpoint**: `/v1/graph/{graph_name}/dataloading`
+- **Endpoint**: `/v1/graph/{graph_id}/dataloading`
 - **Content-type**: `application/json`
 - **Body**:
 ```json
@@ -493,8 +495,8 @@ Import data to empty graph.
   "graph": "graph",
   "loading_config": {
     "data_source": {
-      "scheme": "file"
-	  "location": {path_to_file},
+      "scheme": "file",
+	    "location": "path_to_file",
     },
     "format": {
       "metadata": {
@@ -669,7 +671,7 @@ Import data to empty graph.
 
 #### Curl Command Example
 ```bash
-curl -X POST -H "Content-Type: application/json" -d @path/to/json "http://[host]/v1/graph/{graph_name}/dataloading"
+curl -X POST -H "Content-Type: application/json" -d @path/to/json "http://[host]/v1/graph/{graph_id}/dataloading"
 ```
 
 #### Expected Response
@@ -695,7 +697,7 @@ Create a new stored procedure.
 
 #### HTTP Request
 - **Method**: POST
-- **Endpoint**: `/v1/graph/{graph_name}/procedure`
+- **Endpoint**: `/v1/graph/{graph_id}/procedure`
 - **Content-type**: `application/json`
 - **Body**:
 ```json
@@ -711,7 +713,7 @@ Create a new stored procedure.
 
 #### Curl Command Example
 ```bash
-curl -X POST -H "Content-Type: application/json" -d @/path/to/json "http://[host]/v1/graph/{graph_name}/procedure"
+curl -X POST -H "Content-Type: application/json" -d @/path/to/json "http://[host]/v1/graph/{graph_id}/procedure"
 ```
 
 
@@ -737,12 +739,12 @@ List all procedures bound to a graph.
 
 #### HTTP Request
 - **Method**: GET
-- **Endpoint**: `/v1/graph/{graph_name}/procedure`
+- **Endpoint**: `/v1/graph/{graph_id}/procedure`
 - **Content-type**: `application/json`
 
 #### Curl Command Example
 ```bash
-curl -X GET -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_name}/procedure"
+curl -X GET -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_id}/procedure"
 ```
 
 #### Expected Response
@@ -795,12 +797,12 @@ Get a single procedure's metadata.
 
 #### HTTP Request
 - **Method**: GET
-- **Endpoint**: `/v1/graph/{graph_name}/procedure/{procedure_name}`
+- **Endpoint**: `/v1/graph/{graph_id}/procedure/{procedure_id}`
 - **Content-type**: `application/json`
 
 #### Curl Command Example
 ```bash
-curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_name}/procedure/{procedure_name}"
+curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_id}/procedure/{procedure_id}"
 ```
 
 
@@ -853,7 +855,7 @@ Update a procedure's metadata, enable/disable status, description. The procedure
 
 #### HTTP Request
 - **Method**: PUT
-- **Endpoint**: `/v1/graph/{graph_name}/procedure/{procedure_name}`
+- **Endpoint**: `/v1/graph/{graph_id}/procedure/{procedure_id}`
 - **Content-type**: `application/json`
 - **Body**:
 ```json
@@ -865,7 +867,7 @@ Update a procedure's metadata, enable/disable status, description. The procedure
 
 #### Curl Command Example
 ```bash
-curl -X PUT  -H "Content-Type: application/json" -d @/path/to/json "http://[host]//v1/graph/{graph_name}/procedure/{procedure_name}"
+curl -X PUT  -H "Content-Type: application/json" -d @/path/to/json "http://[host]//v1/graph/{graph_id}/procedure/{procedure_id}"
 ```
 
 #### Expected Response
@@ -916,13 +918,13 @@ Delete a procedure bound to the graph.
 #### HTTP Request
 
 - **Method**: DELETE
-- **Endpoint**: `/v1/graph/{graph_name}/procedure/{procedure_name}`
+- **Endpoint**: `/v1/graph/{graph_id}/procedure/{procedure_id}`
 - **Content-type**: `application/json`
 
 
 #### Curl Command Example
 ```bash
-curl -X DELETE -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_name}/procedure/{procedure_name}"
+curl -X DELETE -H "Content-Type: application/json" "http://[host]/v1/graph/{graph_id}/procedure/{procedure_id}"
 ```
 
 #### Expected Response
@@ -943,7 +945,7 @@ curl -X DELETE -H "Content-Type: application/json" "http://[host]/v1/graph/{grap
 
 #### Description 
 
-Start the query service on a graph. The `graph_name` param can be empty, indicating restarting on current running graph.
+Start the query service on a graph. The `graph_id` param can be empty, indicating restarting on current running graph.
 
 1. After the AdminService receives this request, the current actor scope for query actors will be cancelled.
 2. During the scope cancellation process of the query actors or after scope cancellation is completed, all requests sent to the query_service will fail and be rejected. 
@@ -965,7 +967,7 @@ The response of the http request will be like
 - **Body**:
 ```json
 {
-	"graph_name": "modern_graph"
+	"graph_id": "modern_graph"
 }
 ```
 
@@ -987,31 +989,67 @@ curl -X POST -H "Content-Type: application/json" "http://[host]/v1/service/start
 - `200 OK`: Request successful.
 - `500 Internal Error`: Server internal Error.
 
-### StopService (ServiceManagement Category)
 
-#### Description 
+### RestartService (ServiceManagement Category)
 
-Stop the current running query service. 
+#### Description
+
+Restart the graph query service on current running graph.
 
 #### HTTP Request
 - **Method**: POST
-- **Endpoint**: `/v1/service/stop`
+- **Endpoint**: `/v1/service/restart`
 - **Content-type**: `application/json`
+- **Body**: EmptyBody.
 
 #### Curl Command Example
 ```bash
-curl -X POST -H "Content-Type: application/json" "http://[host]/v1/service/stop"
+curl -X POST "http://[host]/v1/service/restart"
 ```
 
 #### Expected Response
 - **Format**: application/json
 - **Body**:
 ```json
+"Successfully started service"
+```
+
+#### Status Codes
+- `200 OK`: Request successful.
+- `500 Internal Error`: Server internal Error.
+
+
+### StopService (ServiceManagement Category)
+
+#### Description
+
+Stop the graph query service. Note that after service is stopped, the query server is still running, you still can submit queries to the endpoint, 
+but you will receive the following error message to each request:
+```json
 {
-  "message": "message"
+  "code": 500,
+  "message" : "Unable to send message, the target actor has been canceled!"
 }
 ```
 
+#### HTTP Request
+- **Method**: POST
+- **Endpoint**: `/v1/service/stop`
+- **Content-type**: `application/json`
+- **Body**: EmptyBody.
+
+#### Curl Command Example
+```bash
+curl -X POST "http://[host]/v1/service/stop"
+```
+
+#### Expected Response
+- **Format**: application/json
+- **Body**:
+```text
+Successfully stop service
+```
+
 #### Status Codes
 - `200 OK`: Request successful.
 - `500 Internal Error`: Server internal Error.
@@ -1032,7 +1070,7 @@ Get the status of current service.
 - **Body**:
 ```json
 {
-  "graph_name": "graph_name",
+  "graph_id": "graph_id",
   "query_port": 6,
   "status": "running"
 }
@@ -1042,7 +1080,6 @@ Get the status of current service.
 - `200 OK`: Request successful.
 - `500 Internal Error`: Server internal Error.
 
-
 ### SystemMetrics (NodeMetrics Category)
 
 #### Description
@@ -1069,6 +1106,124 @@ curl -X GET  -H "Content-Type: application/json" "http://[host]/v1/node/status"
 }
 ```
 
+#### Status Codes
+- `200 OK`: Request successful.
+- `500 Internal Error`: Server internal Error.
+
+
+###  GetJobById (JobManagement Category)
+
+#### Description
+
+Get the metadata of a job with a unique job id.
+The job id is received when you [launch a bulk loading job](#importgraph-graphmanagement-category)
+
+#### HTTP Request
+- **Method**: GET
+- **Endpoint**: `/v1/job/{job_id}`
+- **Content-type**: `application/json`
+- **Body** : EmptyBody
+
+#### Curl Command Example
+```bash
+curl -X GET "http://[host]/v1/job/{job_id}"
+```
+
+#### Expected Response
+- **Format**: application/json
+- **Body**:
+```json
+{
+  "detail": {
+    "graph_id": "graph_id"
+  },
+  "job_id": "job_{graph_id}_{create_timestamp}_{process_id}",
+  "log": "line1\nlin2...\n",
+  "start_time": 1706786404768, // create_timestamp
+  "status": "CANCELLED/RUNNING/SUCCESS/FAILED",
+  "type": "bulk_loading"
+}
+```
+
+#### Status Codes
+- `200 OK`: Request successful.
+- `500 Internal Error`: Server internal Error.
+- `404 Not Found`: Job not found
+
+
+###  GetAllJobs (JobManagement category)
+
+#### Description
+
+Get the metadata of all running/cancelled/success/failed jobs.
+
+#### HTTP Request
+- **Method**: GET
+- **Endpoint**: `/v1/job/`
+- **Content-type**: `application/json`
+- **Body** : EmptyBody
+
+#### Curl Command Example
+```bash
+curl -X GET "http://[host]/v1/job/"
+```
+
+#### Expected Response
+- **Format**: application/json
+- **Body**:
+```json
+[
+  {
+  "detail": {
+    "graph_id": "graph_id"
+  },
+  "job_id": "job_{graph_id}_{create_timestamp}_{process_id}",
+  "log": "line1\nlin2...\n",
+  "start_time": 1706786404768, // create_timestamp
+  "status": "CANCELLED/RUNNING/SUCCESS/FAILED",
+  "type": "bulk_loading"
+  }
+]
+```
+
+#### Status Codes
+- `200 OK`: Request successful.
+- `500 Internal Error`: Server internal Error.
+
+
+###  CancelJobById (JobManagement category)
+
+#### Description
+
+Cancel a job according to the give job_id.
+
+#### HTTP Request
+- **Method**: DELETE
+- **Endpoint**: `/v1/job/{job_id}`
+- **Content-type**: `application/json`
+- **Body** : EmptyBody
+
+#### Curl Command Example
+```bash
+curl -X DELETE "http://[host]/v1/job/{job_id}"
+```
+
+#### Expected Response
+- **Format**: application/json
+- **Body**:
+
+
+Fail:
+```text
+Job is not running, can not cancel: {job_id}
+```
+
+Success
+```
+Successfully cancelled job: {job_id}
+```
+
+
 #### Status Codes
 - `200 OK`: Request successful.
 - `500 Internal Error`: Server internal Error.
@@ -1111,4 +1266,4 @@ The mapping between statusCode and http code is shown in the following table.
 | gs::StatusCode::IOError             | 500         |
 | gs::StatusCode::NotFound            | 404         |
 | gs::StatusCode::QueryFailed         | 500         |
-| default                             | 500         |
\ No newline at end of file
+| default                             | 500         |
diff --git a/flex/CMakeLists.txt b/flex/CMakeLists.txt
index ba9e1ff60ca4..a4a8ded53e4c 100644
--- a/flex/CMakeLists.txt
+++ b/flex/CMakeLists.txt
@@ -14,7 +14,6 @@ option(BUILD_HQPS "Whether to build HighQPS Engine" ON)
 option(BUILD_TEST "Whether to build test" ON)
 option(BUILD_DOC "Whether to build doc" OFF)
 option(BUILD_ODPS_FRAGMENT_LOADER "Whether to build odps fragment loader" OFF)
-
 option(USE_PTHASH "Whether to use pthash" OFF)
 
 #print options
@@ -22,7 +21,6 @@ message(STATUS "Build HighQPS Engine: ${BUILD_HQPS}")
 message(STATUS "Build test: ${BUILD_TEST}")
 message(STATUS "Build doc: ${BUILD_DOC}")
 message(STATUS "Build odps fragment loader: ${BUILD_ODPS_FRAGMENT_LOADER}")
-
 message(STATUS "Use pthash indexer : ${USE_PTHASH}")
 
 # ------------------------------------------------------------------------------
@@ -121,6 +119,7 @@ endif ()
 find_package(Boost REQUIRED COMPONENTS system filesystem
              # required by folly
              context program_options regex thread date_time)
+add_definitions("-DBOOST_BIND_GLOBAL_PLACEHOLDERS")
 
 #find arrow----------------------------------------------------------------------
 include("cmake/FindArrow.cmake")
@@ -159,6 +158,7 @@ if (BUILD_DOC)
     endif(DOXYGEN_FOUND)
 endif()
 
+
 # Check whether ${CMAKE_SOURCE_DIR}/third_party/single_include/nlohmann/json.hpp exists
 if (NOT EXISTS ${CMAKE_SOURCE_DIR}/third_party/nlohmann-json/single_include/nlohmann/json.hpp)
     message(FATAL_ERROR "${CMAKE_SOURCE_DIR}/third_party/nlohmann-json/single_include/nlohmann/json.hpp not found, "
diff --git a/flex/bin/interactive_server.cc b/flex/bin/interactive_server.cc
index a5896c1fc836..2982496a6d81 100644
--- a/flex/bin/interactive_server.cc
+++ b/flex/bin/interactive_server.cc
@@ -120,8 +120,6 @@ void openDefaultGraph(const std::string workspace, int32_t thread_num,
   }
   LOG(INFO) << "Successfully init graph db for default graph: "
             << default_graph;
-
-  server::WorkDirManipulator::SetRunningGraph(default_graph);
 }
 
 }  // namespace gs
diff --git a/flex/bin/load_plan_and_gen.sh b/flex/bin/load_plan_and_gen.sh
index 9cf71a27ff6c..30d3f683be22 100755
--- a/flex/bin/load_plan_and_gen.sh
+++ b/flex/bin/load_plan_and_gen.sh
@@ -268,7 +268,7 @@ compile_hqps_so() {
   cp ${CMAKE_TEMPLATE_PATH} ${cur_dir}/CMakeLists.txt
   # run cmake and make in output path.
   pushd ${cur_dir}
-  cmd="cmake . -DQUERY_NAME=${query_name} -DFLEX_INCLUDE_PREFIX=${FLEX_INCLUDE_PREFIX}"
+  cmd="cmake . -DQUERY_NAME=${procedure_name} -DFLEX_INCLUDE_PREFIX=${FLEX_INCLUDE_PREFIX}"
   # if CMAKE_CXX_COMPILER is set, use it.
   if [ ! -z ${CMAKE_CXX_COMPILER} ]; then
     cmd="${cmd} -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER}"
diff --git a/flex/engines/http_server/CMakeLists.txt b/flex/engines/http_server/CMakeLists.txt
index ab843eb02b22..07cd7034d813 100644
--- a/flex/engines/http_server/CMakeLists.txt
+++ b/flex/engines/http_server/CMakeLists.txt
@@ -49,5 +49,6 @@ if (Hiactor_FOUND)
   if (OPENTELEMETRY_CPP_FOUND)
     target_link_libraries(flex_server otel)
   endif()
+  target_link_libraries(flex_server flex_metadata_store)
   install_without_export_flex_target(flex_server)
 endif ()
diff --git a/flex/engines/http_server/actor/admin_actor.act.cc b/flex/engines/http_server/actor/admin_actor.act.cc
index 8c4121ab6cd6..75eb588ffdde 100644
--- a/flex/engines/http_server/actor/admin_actor.act.cc
+++ b/flex/engines/http_server/actor/admin_actor.act.cc
@@ -13,11 +13,14 @@
  * limitations under the License.
  */
 
+#include <filesystem>
+
 #include "flex/engines/http_server/actor/admin_actor.act.h"
 
 #include "flex/engines/graph_db/database/graph_db.h"
 #include "flex/engines/graph_db/database/graph_db_session.h"
 #include "flex/engines/http_server/codegen_proxy.h"
+#include "flex/engines/http_server/service/hqps_service.h"
 #include "flex/engines/http_server/workdir_manipulator.h"
 #include "flex/utils/service_utils.h"
 #include "nlohmann/json.hpp"
@@ -26,6 +29,182 @@
 
 namespace server {
 
+void add_runnable_info(gs::PluginMeta& plugin_meta) {
+  const auto& graph_db = gs::GraphDB::get();
+  const auto& schema = graph_db.schema();
+  const auto& plugins_map = schema.GetPlugins();
+  auto plugin_iter = plugins_map.find(plugin_meta.id);
+  if (plugin_iter != plugins_map.end()) {
+    plugin_meta.runnable = true;
+  } else {
+    plugin_meta.runnable = false;
+  }
+}
+
+gs::Result<gs::JobId> invoke_loading_graph(
+    std::shared_ptr<gs::IGraphMetaStore> metadata_store,
+    const std::string& graph_id, const YAML::Node& loading_config,
+    int32_t loading_thread_num) {
+  // First try to load to a tmp directory.
+  // If the loading process itself is atomic(The tmp data dir will be cleaned
+  // if process terminated unexpectedly, and the original data dir is
+  // recovered)
+  auto tmp_indices_dir = WorkDirManipulator::GetTempIndicesDir(graph_id);
+  WorkDirManipulator::CleanTempIndicesDir(graph_id);
+  auto loading_res = server::WorkDirManipulator::LoadGraph(
+      graph_id, loading_config, loading_thread_num, tmp_indices_dir,
+      metadata_store);
+  if (!loading_res.ok()) {
+    WorkDirManipulator::CleanTempIndicesDir(graph_id);
+    return loading_res.status();
+  }
+  auto job_id = loading_res.value();
+  return gs::Result<gs::JobId>(job_id);
+}
+
+seastar::future<seastar::sstring> invoke_creating_procedure(
+    std::shared_ptr<gs::IGraphMetaStore> metadata_store,
+    const std::string& graph_id, const std::string& plugin_creation_parameter) {
+  auto& hqps_service = HQPSService::get();
+  // First create a plugin meta to get the plugin id, then do the real
+  // creation.
+  nlohmann::json json;
+  try {
+    LOG(INFO) << "parsing: " << plugin_creation_parameter;
+    json = nlohmann::json::parse(plugin_creation_parameter);
+  } catch (const std::exception& e) {
+    return seastar::make_exception_future<seastar::sstring>(
+        "Fail to parse parameter as json: " + plugin_creation_parameter);
+  }
+  if (json.contains("name")) {
+    // Currently we need id== name
+    json["id"] = json["name"];
+  }
+  json["graph_id"] = graph_id;
+  json["creation_time"] = gs::GetCurrentTimeStamp();
+  json["update_time"] = json["creation_time"];
+  auto procedure_meta_request = gs::CreatePluginMetaRequest::FromJson(json);
+
+  LOG(INFO) << "parse create plugin meta:" << procedure_meta_request.ToString();
+  auto insert_res = metadata_store->CreatePluginMeta(procedure_meta_request);
+  if (!insert_res.ok()) {
+    return seastar::make_exception_future<seastar::sstring>(
+        std::runtime_error(insert_res.status().error_message()));
+  }
+  auto plugin_id = insert_res.value();
+
+  return server::WorkDirManipulator::CreateProcedure(
+             graph_id, plugin_id, json,
+             hqps_service.get_service_config().engine_config_path)
+      .then_wrapped([graph_id = graph_id, old_plugin_id = plugin_id,
+                     json = json, metadata_store = metadata_store](auto&& f) {
+        std::string proc_id;
+        try {
+          proc_id = f.get0();
+          // proc_yaml path should already checked to exists.
+          if (proc_id.empty()) {
+            metadata_store->DeletePluginMeta(graph_id, old_plugin_id);
+            return seastar::make_exception_future<seastar::sstring>(
+                std::runtime_error("Fail to create plugin: " + proc_id));
+          }
+          if (proc_id != old_plugin_id) {
+            metadata_store->DeletePluginMeta(graph_id, old_plugin_id);
+            return seastar::make_exception_future<seastar::sstring>(
+                std::runtime_error(
+                    std::string(
+                        "the generated plugin id is not same as the old one:") +
+                    proc_id + " " + old_plugin_id));
+          }
+          VLOG(10) << "Successfully create plugin and meta: " << proc_id
+                   << ", now update "
+                      "the plugin meta and update the graph meta: "
+                   << graph_id;
+
+          //  Then insert the plugin meta.
+          auto procedure_meta_from_file =
+              WorkDirManipulator::GetProcedureByGraphAndProcedureName(graph_id,
+                                                                      proc_id);
+          if (!procedure_meta_from_file.ok()) {
+            VLOG(10) << "Fail to insert plugin meta: "
+                     << procedure_meta_from_file.status().error_message();
+            metadata_store->DeletePluginMeta(graph_id, old_plugin_id);
+            WorkDirManipulator::DeleteProcedure(graph_id, proc_id);
+            return seastar::make_exception_future<seastar::sstring>(
+                std::runtime_error(
+                    procedure_meta_from_file.status().error_message() + ", " +
+                    proc_id.c_str()));
+          }
+          seastar::sstring procedure_meta_str =
+              procedure_meta_from_file.value();
+          VLOG(10) << "got procedure meta: " << procedure_meta_str;
+          // When updating procedure meta, we should not change the name. since
+          // neo4j use name as the key.
+          auto internal_plugin_update =
+              gs::UpdatePluginMetaRequest::FromJson(procedure_meta_str);
+          // the field enable should be parsed from json
+          if (json.contains("enable")) {
+            internal_plugin_update.enable = json["enable"].get<bool>();
+          }
+          auto str = internal_plugin_update.ToString();
+          VLOG(10) << "internal plugin update: " << str;
+          auto update_res = metadata_store->UpdatePluginMeta(
+              graph_id, proc_id, internal_plugin_update);
+          VLOG(10) << "update_res: " << update_res.status().ok();
+          if (!update_res.ok()) {
+            metadata_store->DeletePluginMeta(graph_id, old_plugin_id);
+            WorkDirManipulator::DeleteProcedure(graph_id, proc_id);
+            return seastar::make_exception_future<seastar::sstring>(
+                std::runtime_error(update_res.status().error_message()));
+          }
+          VLOG(10) << "Successfully created procedure: " << proc_id;
+          std::string response = "{\"procedure_id\":\"" + proc_id + "\"}";
+          return seastar::make_ready_future<seastar::sstring>(response);
+        } catch (std::exception& e) {
+          LOG(ERROR) << "Fail to create plugin: " << e.what();
+          metadata_store->DeletePluginMeta(graph_id, old_plugin_id);
+          WorkDirManipulator::DeleteProcedure(graph_id, old_plugin_id);
+          return seastar::make_exception_future<seastar::sstring>(
+              std::runtime_error("Fail to create plugin: "));
+        }
+      });
+}
+
+gs::Status invoke_delete_plugin_meta(
+    std::shared_ptr<gs::IGraphMetaStore> metadata_store,
+    const std::string& graph_id, const std::string& procedure_id) {
+  // First delete the plugin meta.
+  auto delete_meta_res =
+      metadata_store->DeletePluginMeta(graph_id, procedure_id);
+  if (!delete_meta_res.ok()) {
+    return delete_meta_res.status();
+  }
+  // Then delete the plugin libxx.so and xxx.yaml on disk
+  auto delete_res =
+      server::WorkDirManipulator::DeleteProcedure(graph_id, procedure_id);
+  if (!delete_res.ok()) {
+    return delete_res.status();
+  }
+  return gs::Status::OK();
+}
+
+// util functions
+
+std::string to_json_str(const std::vector<gs::PluginMeta>& plugin_metas) {
+  nlohmann::json res;
+  for (auto& plugin_meta : plugin_metas) {
+    res.push_back(nlohmann::json::parse(plugin_meta.ToJson()));
+  }
+  return res.empty() ? "{}" : res.dump();
+}
+
+std::string to_json_str(const std::vector<gs::JobMeta>& job_metas) {
+  nlohmann::json res;
+  for (auto& job_meta : job_metas) {
+    res.push_back(nlohmann::json::parse(job_meta.ToJson(true)));
+  }
+  return res.empty() ? "{}" : res.dump();
+}
+
 admin_actor::~admin_actor() {
   // finalization
   // ...
@@ -37,6 +216,9 @@ admin_actor::admin_actor(hiactor::actor_base* exec_ctx,
   set_max_concurrency(1);  // set max concurrency for task reentrancy (stateful)
   // initialization
   // ...
+  auto& hqps_service = HQPSService::get();
+  // meta_data_ should be thread safe.
+  metadata_store_ = hqps_service.get_metadata_store();
 }
 
 // Create a new Graph with the passed graph config.
@@ -55,37 +237,85 @@ seastar::future<admin_query_result> admin_actor::run_create_graph(
     LOG(ERROR) << "Fail to parse json: " << e.what();
     return seastar::make_ready_future<admin_query_result>(
         gs::Result<seastar::sstring>(
-            gs::StatusCode::InvalidSchema,
-            "Fail to parse json: " + std::string(e.what())));
+            gs::Status(gs::StatusCode::InvalidSchema,
+                       "Fail to parse json: " + std::string(e.what()))));
   } catch (...) {
     LOG(ERROR) << "Fail to parse json: " << query_param.content;
     return seastar::make_ready_future<admin_query_result>(
-        gs::Result<seastar::sstring>(gs::StatusCode::InvalidSchema,
-                                     "Fail to parse json: "));
+        gs::Result<seastar::sstring>(
+            gs::Status(gs::StatusCode::InvalidSchema, "Fail to parse json: ")));
   }
 
-  auto result = server::WorkDirManipulator::CreateGraph(yaml);
-  return seastar::make_ready_future<admin_query_result>(std::move(result));
+  auto parse_schema_res = gs::Schema::LoadFromYamlNode(yaml);
+  if (!parse_schema_res.ok()) {
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(parse_schema_res.status()));
+  }
+
+  auto result = metadata_store_->CreateGraphMeta(
+      gs::CreateGraphMetaRequest::FromJson(query_param.content));
+  // we also need to store a graph.yaml on disk, for other services to read.
+  if (result.ok()) {
+    auto dump_res = WorkDirManipulator::DumpGraphSchema(result.value(), yaml);
+    if (!dump_res.ok()) {
+      LOG(ERROR) << "Fail to dump graph schema: "
+                 << dump_res.status().error_message();
+      // If dump schema fails, we should delete the graph meta.
+      metadata_store_->DeleteGraphMeta(result.value());
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(dump_res.status()));
+    } else {
+      VLOG(10) << "Successfully created graph";
+      std::string response = "{\"graph_id\":\"" + result.value() + "\"}";
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(std::move(response)));
+    }
+  } else {
+    LOG(ERROR) << "Fail to create graph: " << result.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(result.status()));
+  }
 }
 
 // get graph schema
 // query_param is the graph name
 seastar::future<admin_query_result> admin_actor::run_get_graph_schema(
     query_param&& query_param) {
-  LOG(INFO) << "Get Graph schema for graph: " << query_param.content;
+  LOG(INFO) << "Get Graph schema for graph_id: " << query_param.content;
+  auto schema_res = metadata_store_->GetGraphMeta(query_param.content);
 
-  auto schema_result =
-      server::WorkDirManipulator::GetGraphSchemaString(query_param.content);
-  return seastar::make_ready_future<admin_query_result>(
-      std::move(schema_result));
+  if (schema_res.ok()) {
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(std::move(schema_res.value().schema)));
+  } else {
+    LOG(ERROR) << "Fail to get graph schema: "
+               << schema_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(schema_res.status()));
+  }
 }
 
 // list all graphs
 seastar::future<admin_query_result> admin_actor::run_list_graphs(
     query_param&& query_param) {
   LOG(INFO) << "List all graphs.";
-  auto list_result = server::WorkDirManipulator::ListGraphs();
-  return seastar::make_ready_future<admin_query_result>(std::move(list_result));
+  // auto list_result = server::WorkDirManipulator::ListGraphs();
+  auto all_graph_meta_res = metadata_store_->GetAllGraphMeta();
+  if (!all_graph_meta_res.ok()) {
+    LOG(ERROR) << "Fail to list graphs: "
+               << all_graph_meta_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(all_graph_meta_res.status()));
+  } else {
+    VLOG(10) << "Successfully list graphs";
+    // collect all 'schema' field into a json stirng
+    nlohmann::json res;
+    for (auto& graph_meta : all_graph_meta_res.value()) {
+      res.push_back(nlohmann::json::parse(graph_meta.ToJson()));
+    }
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(res.dump()));
+  }
 }
 
 // delete one graph
@@ -93,98 +323,214 @@ seastar::future<admin_query_result> admin_actor::run_delete_graph(
     query_param&& query_param) {
   LOG(INFO) << "Delete graph: " << query_param.content;
 
-  auto delete_res =
-      server::WorkDirManipulator::DeleteGraph(query_param.content);
+  auto get_res = metadata_store_->GetGraphMeta(query_param.content);
+  if (!get_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << query_param.content;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_res.status()));
+  }
+  // can not delete a builtin graph
+  if (get_res.value().is_builtin) {
+    LOG(ERROR) << "Can not delete a builtin graph: " << query_param.content;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::IllegalOperation,
+            "Can not delete a builtin graph: " + query_param.content)));
+  }
 
-  return seastar::make_ready_future<admin_query_result>(std::move(delete_res));
+  auto delete_res = metadata_store_->DeleteGraphMeta(query_param.content);
+
+  if (delete_res.ok()) {
+    // delete the disk data
+    auto delete_plugins_res =
+        metadata_store_->DeletePluginMetaByGraphId(query_param.content);
+    if (!delete_plugins_res.ok()) {
+      LOG(ERROR) << "Fail to delete graph's plugins: "
+                 << delete_plugins_res.status().error_message();
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(delete_plugins_res.status()));
+    }
+    WorkDirManipulator::DeleteGraph(query_param.content);
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>("Successfully delete graph: " +
+                                     query_param.content));
+  } else {
+    LOG(ERROR) << "Fail to delete graph: "
+               << delete_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(delete_res.status()));
+  }
 }
 
 // load the graph.
 seastar::future<admin_query_result> admin_actor::run_graph_loading(
     graph_management_param&& query_param) {
-  // query_param contains two parameter, first for graph name, second for graph
-  // config
+  // query_param contains two parameter, first for graph name, second for
+  // graph config
   auto content = query_param.content;
-  auto& graph_name = content.first;
-  VLOG(1) << "Parse json payload for graph: " << graph_name;
-  auto& graph_config = content.second;
+  auto& graph_id = content.first;
+  VLOG(1) << "Parse json payload for graph: " << graph_id;
+  auto& loading_config = content.second;
 
   YAML::Node yaml;
   try {
     // parse json from query_param.content
-    nlohmann::json json = nlohmann::json::parse(graph_config);
+    nlohmann::json json = nlohmann::json::parse(loading_config);
     std::stringstream json_ss;
-    json_ss << graph_config;
+    json_ss << loading_config;
     yaml = YAML::Load(json_ss);
   } catch (std::exception& e) {
     LOG(ERROR) << "Fail to parse json: " << e.what();
     return seastar::make_ready_future<admin_query_result>(
         gs::Result<seastar::sstring>(
-            gs::StatusCode::InvalidImportFile,
-            "Fail to parse json: " + std::string(e.what())));
+            gs::Status(gs::StatusCode::InvalidImportFile,
+                       "Fail to parse json: " + std::string(e.what()))));
   } catch (...) {
-    LOG(ERROR) << "Fail to parse json: ";
+    LOG(ERROR) << "Fail to parse json: " << loading_config;
     return seastar::make_ready_future<admin_query_result>(
-        gs::Result<seastar::sstring>(gs::StatusCode::InvalidImportFile,
-                                     "Fail to parse json: "));
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::InvalidImportFile, "Fail to parse json: ")));
   }
 
   int32_t loading_thread_num = 1;
   if (yaml["loading_thread_num"]) {
     loading_thread_num = yaml["loading_thread_num"].as<int32_t>();
   }
+  // First check graph exists
+  auto graph_meta_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!graph_meta_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(graph_meta_res.status()));
+  }
 
-  auto graph_loading_res = server::WorkDirManipulator::LoadGraph(
-      graph_name, yaml, loading_thread_num);
+  // try to lock the graph indices dir
+  auto lock_res = metadata_store_->LockGraphIndices(graph_id);
+  if (!lock_res.ok() || !lock_res.value()) {
+    LOG(ERROR) << "Fail to lock graph indices dir: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::AlreadyLocked,
+            "Fail to acquire lock for graph indices dir: " + graph_id +
+                ", maybe the graph is already running")));
+  }
+
+  std::string graph_id_str = graph_id.c_str();
+  auto job_id_res = invoke_loading_graph(metadata_store_, graph_id_str, yaml,
+                                         loading_thread_num);
+  if (!job_id_res.ok()) {
+    LOG(ERROR) << "Fail to run graph loading : "
+               << job_id_res.status().error_message();
+    metadata_store_->UnlockGraphIndices(graph_id);
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(job_id_res.status()));
+  }
+  seastar::sstring res = "{\"job_id\":\"" + job_id_res.value() + "\"}";
   return seastar::make_ready_future<admin_query_result>(
-      std::move(graph_loading_res));
+      gs::Result<seastar::sstring>(std::move(res)));
 }
 
 // Get all procedure with graph_name and procedure_name
 seastar::future<admin_query_result>
 admin_actor::get_procedure_by_procedure_name(
     procedure_query_param&& query_param) {
-  auto& graph_name = query_param.content.first;
-  auto& procedure_name = query_param.content.second;
-  LOG(INFO) << "Get procedure: " << procedure_name
-            << " for graph: " << graph_name;
+  auto& graph_id = query_param.content.first;
+  auto& procedure_id = query_param.content.second;
+
+  auto get_graph_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!get_graph_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_graph_res.status()));
+  }
+
+  LOG(INFO) << "Get procedure: " << procedure_id << " for graph: " << graph_id;
   auto get_procedure_res =
-      server::WorkDirManipulator::GetProcedureByGraphAndProcedureName(
-          graph_name, procedure_name);
+      metadata_store_->GetPluginMeta(graph_id, procedure_id);
 
-  return seastar::make_ready_future<admin_query_result>(
-      std::move(get_procedure_res));
+  if (get_procedure_res.ok()) {
+    VLOG(10) << "Successfully get procedure procedures";
+    auto& proc_meta = get_procedure_res.value();
+    add_runnable_info(proc_meta);
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(proc_meta.ToJson()));
+  } else {
+    LOG(ERROR) << "Fail to get procedure for graph: " << graph_id
+               << " and procedure: " << procedure_id << ", error message: "
+               << get_procedure_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_procedure_res.status()));
+  }
 }
 
 // Get all procedures of one graph.
 seastar::future<admin_query_result> admin_actor::get_procedures_by_graph_name(
     query_param&& query_param) {
-  auto& graph_name = query_param.content;
-  auto get_all_procedure_res =
-      server::WorkDirManipulator::GetProceduresByGraphName(graph_name);
+  auto& graph_id = query_param.content;
+  // first check graph exists
+  auto graph_meta_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!graph_meta_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(graph_meta_res.status()));
+  }
 
-  return seastar::make_ready_future<admin_query_result>(
-      std::move(get_all_procedure_res));
+  auto get_all_procedure_res = metadata_store_->GetAllPluginMeta(graph_id);
+  if (get_all_procedure_res.ok()) {
+    VLOG(10) << "Successfully get all procedures: "
+             << get_all_procedure_res.value().size();
+    auto& all_plugin_metas = get_all_procedure_res.value();
+    for (auto& plugin_meta : all_plugin_metas) {
+      add_runnable_info(plugin_meta);
+    }
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(to_json_str(all_plugin_metas)));
+  } else {
+    LOG(ERROR) << "Fail to get all procedures: "
+               << get_all_procedure_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_all_procedure_res.status()));
+  }
 }
 
 seastar::future<admin_query_result> admin_actor::create_procedure(
     create_procedure_query_param&& query_param) {
-  auto& graph_name = query_param.content.first;
+  auto& graph_id = query_param.content.first;
   auto& parameter = query_param.content.second;
-  auto& hqps_service = HQPSService::get();
-  return server::WorkDirManipulator::CreateProcedure(
-             graph_name, parameter,
-             hqps_service.get_service_config().engine_config_path)
-      .then_wrapped([](auto&& f) {
+
+  auto graph_meta_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!graph_meta_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(graph_meta_res.status()));
+  }
+
+  auto lock_res = metadata_store_->LockGraphPlugins(graph_id);
+  if (!lock_res.ok() || !lock_res.value()) {
+    LOG(ERROR) << "Fail to lock graph plugin dir: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::AlreadyLocked,
+            "Fail to acquire lock for graph plugin dir: " + graph_id +
+                ", try again later")));
+  }
+
+  return invoke_creating_procedure(metadata_store_, graph_id, parameter)
+      .then_wrapped([this, graph_id = graph_id](auto&& f) {
+        auto unlock_res = metadata_store_->UnlockGraphPlugins(graph_id);
+        if (!unlock_res.ok()) {
+          LOG(ERROR) << "Fail to unlock graph plugin dir: " << graph_id;
+        }
         try {
           auto res = f.get();
           return seastar::make_ready_future<admin_query_result>(
-              admin_query_result{std::move(res)});
+              gs::Result<seastar::sstring>(std::move(res)));
         } catch (std::exception& e) {
           LOG(ERROR) << "Fail to create procedure: " << e.what();
-          return seastar::make_exception_future<admin_query_result>(
-              std::runtime_error("Fail to create procedure: " +
-                                 std::string(e.what())));
+          return seastar::make_ready_future<admin_query_result>(
+              gs::Result<seastar::sstring>(gs::Status(
+                  gs::StatusCode::InternalError,
+                  "Fail to create procedure: " + std::string(e.what()))));
         }
       });
 }
@@ -192,25 +538,120 @@ seastar::future<admin_query_result> admin_actor::create_procedure(
 // Delete a procedure by graph name and procedure name
 seastar::future<admin_query_result> admin_actor::delete_procedure(
     create_procedure_query_param&& query_param) {
-  auto& graph_name = query_param.content.first;
-  auto& procedure_name = query_param.content.second;
-  auto delete_procedure_res =
-      server::WorkDirManipulator::DeleteProcedure(graph_name, procedure_name);
+  auto& graph_id = query_param.content.first;
+  auto& procedure_id = query_param.content.second;
+
+  auto graph_meta_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!graph_meta_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(graph_meta_res.status()));
+  }
+
+  auto get_procedure_res =
+      metadata_store_->GetPluginMeta(graph_id, procedure_id);
+
+  if (!get_procedure_res.ok()) {
+    LOG(ERROR) << "Procedure " << procedure_id
+               << " not exists on graph: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(
+            gs::Status(gs::StatusCode::NotFound,
+                       "Procedure " + procedure_id +
+                           " not exists on graph: " + graph_id)));
+  }
+
+  auto lock_res = metadata_store_->LockGraphPlugins(graph_id);
+  if (!lock_res.ok() || !lock_res.value()) {
+    LOG(ERROR) << "Fail to lock graph plugin dir: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::AlreadyLocked,
+            "Fail to acquire lock for graph plugin dir: " + graph_id +
+                ", try again later")));
+  }
+
+  auto delete_res =
+      invoke_delete_plugin_meta(metadata_store_, graph_id, procedure_id);
+  auto unlock_res = metadata_store_->UnlockGraphPlugins(graph_id);
+  if (!unlock_res.ok()) {
+    LOG(ERROR) << "Fail to unlock graph plugin dir: " << graph_id;
+  }
+  if (!delete_res.ok()) {
+    LOG(ERROR) << "Fail to run delete procedure: "
+               << delete_res.error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(delete_res));
+  }
+
+  VLOG(10) << "Successfully delete procedure: " << procedure_id;
   return seastar::make_ready_future<admin_query_result>(
-      std::move(delete_procedure_res));
+      gs::Result<seastar::sstring>("Successfully delete procedure: " +
+                                   procedure_id));
 }
 
 // update a procedure by graph name and procedure name
 seastar::future<admin_query_result> admin_actor::update_procedure(
     update_procedure_query_param&& query_param) {
-  auto& graph_name = std::get<0>(query_param.content);
-  auto& procedure_name = std::get<1>(query_param.content);
-  auto& parameter = std::get<2>(query_param.content);
-  auto update_procedure_res = server::WorkDirManipulator::UpdateProcedure(
-      graph_name, procedure_name, parameter);
+  auto& graph_id = std::get<0>(query_param.content);
+  auto& procedure_id = std::get<1>(query_param.content);
+  auto& update_request_json = std::get<2>(query_param.content);
 
-  return seastar::make_ready_future<admin_query_result>(
-      std::move(update_procedure_res));
+  auto graph_meta_res = metadata_store_->GetGraphMeta(graph_id);
+  if (!graph_meta_res.ok()) {
+    LOG(ERROR) << "Graph not exists: " << graph_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(graph_meta_res.status()));
+  }
+
+  auto get_procedure_res =
+      metadata_store_->GetPluginMeta(graph_id, procedure_id);
+
+  if (!get_procedure_res.ok()) {
+    LOG(ERROR) << "Procedure not exists: " << procedure_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_procedure_res.status()));
+  }
+  VLOG(10) << "update request json: " << update_request_json;
+  auto req = gs::UpdatePluginMetaRequest::FromJson(update_request_json);
+  VLOG(10) << "Update plugin req: " << req.ToString();
+  // If updatePluginMetaRequest contains field params, returns, library, and
+  // option, we warning and return.
+  if (req.params.has_value() || req.returns.has_value() ||
+      req.library.has_value() || req.option.has_value()) {
+    LOG(ERROR) << "UpdatePluginMetaRequest contains field params, returns, "
+                  "library, or option, which should not be updated.";
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::IllegalOperation,
+            "UpdatePluginMetaRequest contains field params, returns, library, "
+            "and option, which should not be updated.")));
+  }
+
+  if (req.name.has_value()) {
+    LOG(ERROR) << "UpdatePluginMetaRequest contains field 'name', which should "
+                  "not be updated.";
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(
+            gs::Status(gs::StatusCode::IllegalOperation,
+                       "UpdatePluginMetaRequest contains field "
+                       "'name', which should not be updated.")));
+  }
+
+  auto update_res =
+      metadata_store_->UpdatePluginMeta(graph_id, procedure_id, req);
+
+  if (update_res.ok()) {
+    VLOG(10) << "Successfully update procedure: " << procedure_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>("Successfully update procedure: " +
+                                     procedure_id));
+  } else {
+    LOG(ERROR) << "Fail to create procedure: "
+               << update_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(update_res.status()));
+  }
 }
 
 // Start service on a graph first means stop all current running actors, then
@@ -220,39 +661,123 @@ seastar::future<admin_query_result> admin_actor::start_service(
   // parse query_param.content as json and get graph_name
   auto& content = query_param.content;
   std::string graph_name;
+
+  auto cur_running_graph_res = metadata_store_->GetRunningGraph();
+  if (!cur_running_graph_res.ok()) {
+    LOG(ERROR) << "Fail to get running graph: "
+               << cur_running_graph_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(cur_running_graph_res.status()));
+  }
+  auto cur_running_graph = cur_running_graph_res.value();
+  LOG(INFO) << "Current running graph: " << cur_running_graph;
   try {
     if (!content.empty()) {
       nlohmann::json json = nlohmann::json::parse(content);
-      if (json.contains("graph_name")) {
-        graph_name = json["graph_name"].get<std::string>();
+      if (json.contains("graph_id")) {
+        graph_name = json["graph_id"].get<std::string>();
       }
     } else {
-      graph_name = server::WorkDirManipulator::GetRunningGraph();
+      graph_name = cur_running_graph;
       LOG(WARNING)
           << "Request payload is empty, will restart on current graph: "
-          << server::WorkDirManipulator::GetRunningGraph();
+          << graph_name;
     }
     LOG(WARNING) << "Starting service with graph: " << graph_name;
   } catch (std::exception& e) {
     LOG(ERROR) << "Fail to Start service: ";
     return seastar::make_ready_future<admin_query_result>(
         gs::Result<seastar::sstring>(
-            gs::StatusCode::InvalidSchema,
-            "Fail to parse json: " + std::string(e.what())));
+            gs::Status(gs::StatusCode::InvalidSchema,
+                       "Fail to parse json: " + std::string(e.what()))));
+  }
+
+  auto get_graph_res = metadata_store_->GetGraphMeta(graph_name);
+  if (!get_graph_res.ok()) {
+    LOG(ERROR) << "Fail to get graph meta: "
+               << get_graph_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_graph_res.status()));
+  }
+
+  auto get_lock_res = metadata_store_->GetGraphIndicesLocked(graph_name);
+  if (!get_lock_res.ok()) {
+    LOG(ERROR) << "Failed to get lock for graph: " << graph_name;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_lock_res.status()));
+  }
+  auto prev_lock = get_lock_res.value();
+  if (prev_lock) {
+    if (cur_running_graph == graph_name) {
+      LOG(INFO) << "Service already running on graph: " << graph_name;
+    } else {
+      LOG(ERROR) << "The graph is locked but not running: " << graph_name
+                 << ", maybe a data loading job is running on this graph";
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(gs::Status(
+              gs::StatusCode::AlreadyLocked,
+              "The graph is locked but not running: " + graph_name +
+                  ", maybe a data loading job is running on this graph")));
+    }
+  } else {
+    LOG(INFO) << "The graph is not locked: " << graph_name;
+    auto acquire_lock_res = metadata_store_->LockGraphIndices(graph_name);
+    if (!acquire_lock_res.ok() || !acquire_lock_res.value()) {
+      LOG(ERROR) << "Fail to lock graph: " << graph_name;
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(
+              gs::Status(gs::StatusCode::AlreadyLocked,
+                         "Fail to acquire lock for graph: " + graph_name +
+                             ", try again later")));
+    }
+    LOG(INFO) << "Successfully locked graph: " << graph_name;
   }
 
-  auto schema_result = server::WorkDirManipulator::GetGraphSchema(graph_name);
-  if (!schema_result.ok()) {
+  // Dump the latest schema to file, which include all enabled plugins.
+  auto plugins_res = metadata_store_->GetAllPluginMeta(graph_name);
+  if (!plugins_res.ok()) {
+    LOG(ERROR) << "Fail to get all plugins: "
+               << plugins_res.status().error_message();
+    if (!prev_lock) {
+      // If the graph is not locked before, and we fail at some
+      // steps after locking, we should unlock it.
+      metadata_store_->UnlockGraphIndices(graph_name);
+    }
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(plugins_res.status()));
+  }
+  // With all enabled plugins and graph schema, dump to a new schema file.
+  auto dump_res = WorkDirManipulator::DumpGraphSchema(get_graph_res.value(),
+                                                      plugins_res.value());
+  if (!dump_res.ok()) {
+    LOG(ERROR) << "Fail to dump graph schema: "
+               << dump_res.status().error_message();
+    if (!prev_lock) {
+      metadata_store_->UnlockGraphIndices(graph_name);
+    }
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(dump_res.status()));
+  }
+
+  auto schema_res = server::WorkDirManipulator::GetGraphSchema(graph_name);
+  if (!schema_res.ok()) {
     LOG(ERROR) << "Fail to get graph schema: "
-               << schema_result.status().error_message() << ", " << graph_name;
+               << schema_res.status().error_message() << ", " << graph_name;
+    if (!prev_lock) {
+      metadata_store_->UnlockGraphIndices(graph_name);
+    }
     return seastar::make_ready_future<admin_query_result>(
-        gs::Result<seastar::sstring>(schema_result.status()));
+        gs::Result<seastar::sstring>(schema_res.status()));
   }
-  auto& schema_value = schema_result.value();
+  auto& schema_value = schema_res.value();
   auto data_dir = server::WorkDirManipulator::GetDataDirectory(graph_name);
   if (!data_dir.ok()) {
     LOG(ERROR) << "Fail to get data directory: "
                << data_dir.status().error_message();
+    if (!prev_lock) {  // If the graph is not locked before, and we fail at some
+                       // steps after locking, we should unlock it.
+      metadata_store_->UnlockGraphIndices(graph_name);
+    }
     return seastar::make_ready_future<admin_query_result>(
         gs::Result<seastar::sstring>(data_dir.status()));
   }
@@ -261,7 +786,8 @@ seastar::future<admin_query_result> admin_actor::start_service(
   // First Stop query_handler's actors.
 
   auto& hqps_service = HQPSService::get();
-  return hqps_service.stop_query_actors().then([this, graph_name, schema_value,
+  return hqps_service.stop_query_actors().then([this, prev_lock, graph_name,
+                                                schema_value, cur_running_graph,
                                                 data_dir_value, &hqps_service] {
     LOG(INFO) << "Successfully stopped query handler";
 
@@ -276,12 +802,38 @@ seastar::future<admin_query_result> admin_actor::start_service(
       if (!db.Open(schema_value, data_dir_value, thread_num).ok()) {
         LOG(ERROR) << "Fail to load graph from data directory: "
                    << data_dir_value;
+        if (!prev_lock) {  // If the graph is not locked before, and we fail at
+                           // some
+                           // steps after locking, we should unlock it.
+          metadata_store_->UnlockGraphIndices(graph_name);
+        }
         return seastar::make_ready_future<admin_query_result>(
-            gs::Result<seastar::sstring>(
+            gs::Result<seastar::sstring>(gs::Status(
                 gs::StatusCode::InternalError,
-                "Fail to open graph from data directory: " + data_dir_value));
+                "Fail to load graph from data directory: " + data_dir_value)));
+      }
+      // unlock the previous graph
+      if (graph_name != cur_running_graph) {
+        auto unlock_res =
+            metadata_store_->UnlockGraphIndices(cur_running_graph);
+        if (!unlock_res.ok()) {
+          LOG(ERROR) << "Fail to unlock graph: " << cur_running_graph;
+          if (!prev_lock) {
+            metadata_store_->UnlockGraphIndices(graph_name);
+          }
+          return seastar::make_ready_future<admin_query_result>(
+              gs::Result<seastar::sstring>(unlock_res.status()));
+        }
+      }
+      auto set_res = metadata_store_->SetRunningGraph(graph_name);
+      if (!set_res.ok()) {
+        LOG(ERROR) << "Fail to set running graph: " << graph_name;
+        if (!prev_lock) {
+          metadata_store_->UnlockGraphIndices(graph_name);
+        }
+        return seastar::make_ready_future<admin_query_result>(
+            gs::Result<seastar::sstring>(set_res.status()));
       }
-      server::WorkDirManipulator::SetRunningGraph(graph_name);
     }
     hqps_service.start_query_actors();  // start on a new scope.
     LOG(INFO) << "Successfully restart query actors";
@@ -290,8 +842,9 @@ seastar::future<admin_query_result> admin_actor::start_service(
         server::WorkDirManipulator::GetGraphSchemaPath(graph_name);
     if (!hqps_service.start_compiler_subprocess(schema_path)) {
       LOG(ERROR) << "Fail to start compiler";
-      return seastar::make_exception_future<admin_query_result>(
-          seastar::sstring("Fail to start compiler"));
+      return seastar::make_ready_future<admin_query_result>(
+          gs::Result<seastar::sstring>(gs::Status(gs::StatusCode::InternalError,
+                                                  "Fail to start compiler")));
     }
     LOG(INFO) << "Successfully started service with graph: " << graph_name;
     return seastar::make_ready_future<admin_query_result>(
@@ -314,8 +867,8 @@ seastar::future<admin_query_result> admin_actor::stop_service(
     } else {
       LOG(ERROR) << "Fail to stop compiler";
       return seastar::make_ready_future<admin_query_result>(
-          gs::Result<seastar::sstring>(gs::StatusCode::InternalError,
-                                       "Fail to stop compiler"));
+          gs::Result<seastar::sstring>(gs::Status(gs::StatusCode::InternalError,
+                                                  "Fail to stop compiler")));
     }
   });
 }
@@ -325,11 +878,16 @@ seastar::future<admin_query_result> admin_actor::service_status(
     query_param&& query_param) {
   auto& hqps_service = HQPSService::get();
   auto query_port = hqps_service.get_query_port();
+  auto running_graph_res = metadata_store_->GetRunningGraph();
   nlohmann::json res;
   if (query_port != 0) {
-    res["status"] = "running";
+    res["status"] = hqps_service.is_actors_running() ? "Running" : "Stopped";
     res["query_port"] = query_port;
-    res["graph_name"] = server::WorkDirManipulator::GetRunningGraph();
+    if (running_graph_res.ok()) {
+      res["graph_id"] = running_graph_res.value();
+    } else {
+      res["graph_id"] = "UNKNOWN";
+    }
   } else {
     LOG(INFO) << "Query service has not been inited!";
     res["status"] = "Query service has not been inited!";
@@ -365,4 +923,99 @@ seastar::future<admin_query_result> admin_actor::node_status(
       gs::Result<seastar::sstring>(json.dump()));
 }
 
+///////////////////////// Job related /////////////////////////
+seastar::future<admin_query_result> admin_actor::get_job(
+    query_param&& query_param) {
+  auto& job_id = query_param.content;
+  auto job_meta_res = metadata_store_->GetJobMeta(job_id);
+  if (job_meta_res.ok()) {
+    VLOG(10) << "Successfully get job: " << job_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(job_meta_res.value().ToJson()));
+  } else {
+    LOG(ERROR) << "Fail to get job: " << job_id
+               << ", error message: " << job_meta_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(job_meta_res.status()));
+  }
+}
+
+seastar::future<admin_query_result> admin_actor::list_jobs(
+    query_param&& query_param) {
+  auto list_res = metadata_store_->GetAllJobMeta();
+  if (list_res.ok()) {
+    VLOG(10) << "Successfully list jobs";
+    auto list_job_metas_str = to_json_str(list_res.value());
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(std::move(list_job_metas_str)));
+  } else {
+    LOG(ERROR) << "Fail to list jobs: " << list_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(list_res.status());
+  }
+}
+
+// cancel job
+seastar::future<admin_query_result> admin_actor::cancel_job(
+    query_param&& query_param) {
+  auto& job_id = query_param.content;
+  auto get_job_meta_res = metadata_store_->GetJobMeta(job_id);
+  if (!get_job_meta_res.ok()) {
+    LOG(ERROR) << "Job not exists: " << job_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(get_job_meta_res.status()));
+  }
+  auto& job_meta = get_job_meta_res.value();
+  if (job_meta.process_id <= 0) {
+    LOG(ERROR) << "Invalid process id: " << job_meta.process_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(
+            gs::StatusCode::InternalError,
+            "Invalid process id: " + std::to_string(job_meta.process_id)));
+  }
+  // if job is already cancelled, return directly.
+  if (job_meta.status == gs::JobStatus::kCancelled ||
+      job_meta.status == gs::JobStatus::kFailed ||
+      job_meta.status == gs::JobStatus::kSuccess) {
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(
+            gs::Status(gs::StatusCode::IllegalOperation,
+                       "Job already " + std::to_string(job_meta.status) + ": " +
+                           job_id.c_str())));
+  }
+  if (job_meta.status == gs::JobStatus::kUnknown) {
+    VLOG(10) << "Job status is unknown, try cancelling";
+  }
+
+  boost::process::child::child_handle child(job_meta.process_id);
+  std::error_code ec;
+  boost::process::detail::api::terminate(child, ec);
+
+  VLOG(10) << "Killing process: " << job_meta.process_id
+           << ", res: " << ec.message();
+  if (ec.value() != 0) {
+    LOG(ERROR) << "Fail to kill process: " << job_meta.process_id
+               << ", error message: " << ec.message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(gs::Status(
+            gs::StatusCode::InternalError,
+            "Fail to kill process: " + std::to_string(job_meta.process_id) +
+                ", error message: " + ec.message())));
+  }
+  // Now update job meta to cancelled.
+  auto update_job_meta_request = gs::UpdateJobMetaRequest::NewCancel();
+  auto cancel_meta_res =
+      metadata_store_->UpdateJobMeta(job_id, update_job_meta_request);
+
+  if (cancel_meta_res.ok()) {
+    VLOG(10) << "Successfully cancel job: " << job_id;
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>("Successfully cancel job: " + job_id));
+  } else {
+    LOG(ERROR) << "Fail to cancel job: " << job_id << ", error message: "
+               << cancel_meta_res.status().error_message();
+    return seastar::make_ready_future<admin_query_result>(
+        gs::Result<seastar::sstring>(cancel_meta_res.status()));
+  }
+}
+
 }  // namespace server
\ No newline at end of file
diff --git a/flex/engines/http_server/actor/admin_actor.act.h b/flex/engines/http_server/actor/admin_actor.act.h
index daaa9cbf2266..c33233482b25 100644
--- a/flex/engines/http_server/actor/admin_actor.act.h
+++ b/flex/engines/http_server/actor/admin_actor.act.h
@@ -17,13 +17,16 @@
 #define ENGINES_HTTP_SERVER_ACTOR_ADMIN_ACT_H_
 
 #include "flex/engines/http_server/types.h"
-#include "flex/engines/http_server/service/hqps_service.h"
 #include "flex/engines/graph_db/database/graph_db.h"
-#include <mutex>
+
+#include "flex/storages/metadata/graph_meta_store.h"
 
 #include <hiactor/core/actor-template.hh>
 #include <hiactor/util/data_type.hh>
 
+#include <memory>
+#include <mutex>
+
 namespace server {
 
 class ANNOTATION(actor:impl) admin_actor : public hiactor::actor {
@@ -59,12 +62,19 @@ class ANNOTATION(actor:impl) admin_actor : public hiactor::actor {
 
   seastar::future<admin_query_result> ANNOTATION(actor:method) node_status(query_param&& param);
 
+  seastar::future<admin_query_result> ANNOTATION(actor:method) get_job(query_param&& param);
+
+  seastar::future<admin_query_result> ANNOTATION(actor:method) list_jobs(query_param&& param);
+
+  seastar::future<admin_query_result> ANNOTATION(actor:method) cancel_job(query_param&& param);
+
   // DECLARE_RUN_QUERIES;
   /// Declare `do_work` func here, no need to implement.
   ACTOR_DO_WORK()
 
  private:
   std::mutex mtx_;
+  std::shared_ptr<gs::IGraphMetaStore> metadata_store_;
 };
 
 }  // namespace server
diff --git a/flex/engines/http_server/codegen_proxy.cc b/flex/engines/http_server/codegen_proxy.cc
index 3ce930cc8580..91cdf0cf357b 100644
--- a/flex/engines/http_server/codegen_proxy.cc
+++ b/flex/engines/http_server/codegen_proxy.cc
@@ -13,6 +13,7 @@
  * limitations under the License.
  */
 #include "flex/engines/http_server/codegen_proxy.h"
+#include "flex/engines/http_server/service/hqps_service.h"
 #include "flex/engines/http_server/workdir_manipulator.h"
 
 namespace server {
@@ -67,10 +68,25 @@ seastar::future<std::pair<int32_t, std::string>> CodegenProxy::DoGen(
              [this, next_job_id] { return !check_job_running(next_job_id); });
   }
 
-  auto cur_graph_schema_path = default_graph_schema_path_.empty()
-                                   ? WorkDirManipulator::GetGraphSchemaPath(
-                                         WorkDirManipulator::GetRunningGraph())
-                                   : default_graph_schema_path_;
+  auto cur_graph_schema_path = default_graph_schema_path_;
+  if (cur_graph_schema_path.empty()) {
+    auto& hqps_service = server::HQPSService::get();
+    if (hqps_service.get_metadata_store()) {
+      auto running_graph_res =
+          hqps_service.get_metadata_store()->GetRunningGraph();
+      if (!running_graph_res.ok()) {
+        return seastar::make_exception_future<std::pair<int32_t, std::string>>(
+            std::runtime_error("Get running graph failed"));
+      }
+      cur_graph_schema_path =
+          WorkDirManipulator::GetGraphSchemaPath(running_graph_res.value());
+    } else {
+      LOG(ERROR) << "Graph schema path is empty";
+      return seastar::make_exception_future<std::pair<int32_t, std::string>>(
+          std::runtime_error("Graph schema path is empty"));
+    }
+  }
+
   if (cur_graph_schema_path.empty()) {
     LOG(ERROR) << "Graph schema path is empty";
     return seastar::make_exception_future<std::pair<int32_t, std::string>>(
@@ -172,8 +188,8 @@ seastar::future<int> CodegenProxy::CallCodegenCmd(
     const std::string& engine_config, const std::string& procedure_desc) {
   // TODO: different suffix for different platform
   std::string cmd = codegen_bin + " -e=hqps " + " -i=" + plan_path +
-                    " -o=" + output_dir + " --procedure_name=" + query_name +
-                    " -w=" + work_dir + " --ir_conf=" + engine_config +
+                    " -o=" + output_dir + " --procedure_name=\"" + query_name +
+                    "\" -w=" + work_dir + " --ir_conf=" + engine_config +
                     " --graph_schema_path=" + graph_schema_path;
   if (!procedure_desc.empty()) {
     cmd += " --procedure_desc=\'" + procedure_desc + "\'";
diff --git a/flex/engines/http_server/handler/admin_http_handler.cc b/flex/engines/http_server/handler/admin_http_handler.cc
index 8a044d5a651d..8d66360c9327 100644
--- a/flex/engines/http_server/handler/admin_http_handler.cc
+++ b/flex/engines/http_server/handler/admin_http_handler.cc
@@ -28,6 +28,17 @@
 
 namespace server {
 
+std::string trim_slash(const std::string& origin) {
+  std::string res = origin;
+  if (res.front() == '/') {
+    res.erase(res.begin());
+  }
+  if (res.back() == '/') {
+    res.pop_back();
+  }
+  return res;
+}
+
 /**
  * Handle all request for graph management.
  */
@@ -58,13 +69,14 @@ class admin_http_graph_handler_impl : public seastar::httpd::handler_base {
     if (method == "POST") {
       if (path.find("dataloading") != seastar::sstring::npos) {
         LOG(INFO) << "Route to loading graph";
-        if (!req->param.exists("graph_name")) {
-          return new_bad_request_reply(std::move(rep),
-                                       "expect field 'graph_name' in request");
+        if (!req->param.exists("graph_id")) {
+          return seastar::make_exception_future<
+              std::unique_ptr<seastar::httpd::reply>>(
+              std::runtime_error("graph_id not exists"));
         } else {
-          auto graph_name = req->param.at("graph_name");
-          LOG(INFO) << "Graph name: " << graph_name;
-          auto pair = std::make_pair(graph_name, std::move(req->content));
+          auto graph_id = trim_slash(req->param.at("graph_id"));
+          LOG(INFO) << "Graph id: " << graph_id;
+          auto pair = std::make_pair(graph_id, std::move(req->content));
           return admin_actor_refs_[dst_executor]
               .run_graph_loading(graph_management_param{std::move(pair)})
               .then_wrapped(
@@ -86,11 +98,11 @@ class admin_http_graph_handler_impl : public seastar::httpd::handler_base {
                 });
       }
     } else if (method == "GET") {
-      if (req->param.exists("graph_name") &&
+      if (req->param.exists("graph_id") &&
           path.find("schema") != seastar::sstring::npos) {
-        auto graph_name = req->param.at("graph_name");
+        auto graph_id = trim_slash(req->param.at("graph_id"));
         return admin_actor_refs_[dst_executor]
-            .run_get_graph_schema(query_param{std::move(graph_name)})
+            .run_get_graph_schema(query_param{std::move(graph_id)})
             .then_wrapped(
                 [rep = std::move(rep)](
                     seastar::future<admin_query_result>&& fut) mutable {
@@ -108,14 +120,14 @@ class admin_http_graph_handler_impl : public seastar::httpd::handler_base {
                 });
       }
     } else if (method == "DELETE") {
-      if (!req->param.exists("graph_name")) {
+      if (!req->param.exists("graph_id")) {
         return seastar::make_exception_future<
             std::unique_ptr<seastar::httpd::reply>>(
-            std::runtime_error("graph_name not given"));
+            std::runtime_error("graph_id not given"));
       }
-      auto graph_name = req->param.at("graph_name");
+      auto graph_id = trim_slash(req->param.at("graph_id"));
       return admin_actor_refs_[dst_executor]
-          .run_delete_graph(query_param{std::move(graph_name)})
+          .run_delete_graph(query_param{std::move(graph_id)})
           .then_wrapped([rep = std::move(rep)](
                             seastar::future<admin_query_result>&& fut) mutable {
             return return_reply_with_result(std::move(rep), std::move(fut));
@@ -157,29 +169,20 @@ class admin_http_procedure_handler_impl : public seastar::httpd::handler_base {
 
     executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
     LOG(INFO) << "Handling path:" << path << ", method: " << req->_method;
-    // LOG(INFO) << "Graph_name:" << req->param.at("graph_name");
     if (req->_method == "GET") {
-      // get graph_name param
-      if (!req->param.exists("graph_name")) {
+      // get graph_id param
+      if (!req->param.exists("graph_id")) {
         return seastar::make_exception_future<
             std::unique_ptr<seastar::httpd::reply>>(
-            std::runtime_error("graph_name not exists"));
+            std::runtime_error("graph_id not exists"));
       }
-      auto graph_name = req->param.at("graph_name");
-      // remove / from the graph_name
-      graph_name.erase(std::remove(graph_name.begin(), graph_name.end(), '/'),
-                       graph_name.end());
-      if (req->param.exists("procedure_name")) {
+      auto graph_id = trim_slash(req->param.at("graph_id"));
+      if (req->param.exists("procedure_id")) {
         // Get the procedures
-        auto procedure_name = req->param.at("procedure_name");
-        // remove / from the procedure_name
-        procedure_name.erase(
-            std::remove(procedure_name.begin(), procedure_name.end(), '/'),
-            procedure_name.end());
-
-        LOG(INFO) << "Get procedure for: " << graph_name << ", "
-                  << procedure_name;
-        auto pair = std::make_pair(graph_name, procedure_name);
+        auto procedure_id = trim_slash(req->param.at("procedure_id"));
+
+        LOG(INFO) << "Get procedure for: " << graph_id << ", " << procedure_id;
+        auto pair = std::make_pair(graph_id, procedure_id);
         return admin_actor_refs_[dst_executor]
             .get_procedure_by_procedure_name(
                 procedure_query_param{std::move(pair)})
@@ -191,9 +194,9 @@ class admin_http_procedure_handler_impl : public seastar::httpd::handler_base {
                 });
       } else {
         // get all procedures.
-        LOG(INFO) << "Get all procedures for: " << graph_name;
+        LOG(INFO) << "Get all procedures for: " << graph_id;
         return admin_actor_refs_[dst_executor]
-            .get_procedures_by_graph_name(query_param{std::move(graph_name)})
+            .get_procedures_by_graph_name(query_param{std::move(graph_id)})
             .then_wrapped(
                 [rep = std::move(rep)](
                     seastar::future<admin_query_result>&& fut) mutable {
@@ -202,66 +205,52 @@ class admin_http_procedure_handler_impl : public seastar::httpd::handler_base {
                 });
       }
     } else if (req->_method == "POST") {
-      if (!req->param.exists("graph_name")) {
+      if (!req->param.exists("graph_id")) {
         return seastar::make_exception_future<
             std::unique_ptr<seastar::httpd::reply>>(
-            std::runtime_error("graph_name not given"));
+            std::runtime_error("graph_id not given"));
       }
-      auto graph_name = req->param.at("graph_name");
-      // remove / from the graph_name
-      graph_name.erase(std::remove(graph_name.begin(), graph_name.end(), '/'),
-                       graph_name.end());
-      LOG(INFO) << "Creating procedure for: " << graph_name;
+      auto graph_id = trim_slash(req->param.at("graph_id"));
+      LOG(INFO) << "Creating procedure for: " << graph_id;
       return admin_actor_refs_[dst_executor]
           .create_procedure(create_procedure_query_param{
-              std::make_pair(graph_name, std::move(req->content))})
+              std::make_pair(graph_id, std::move(req->content))})
           .then_wrapped([rep = std::move(rep)](
                             seastar::future<admin_query_result>&& fut) mutable {
             return return_reply_with_result(std::move(rep), std::move(fut));
           });
     } else if (req->_method == "DELETE") {
-      // delete must give graph_name and procedure_name
-      if (!req->param.exists("graph_name") ||
-          !req->param.exists("procedure_name")) {
+      // delete must give graph_id and procedure_id
+      if (!req->param.exists("graph_id") ||
+          !req->param.exists("procedure_id")) {
         return seastar::make_exception_future<
             std::unique_ptr<seastar::httpd::reply>>(
-            std::runtime_error("graph_name or procedure_name not given: "));
+            std::runtime_error("graph_id or procedure_id not given: "));
       }
-      auto graph_name = req->param.at("graph_name");
-      graph_name.erase(std::remove(graph_name.begin(), graph_name.end(), '/'),
-                       graph_name.end());
-      auto procedure_name = req->param.at("procedure_name");
-      procedure_name.erase(
-          std::remove(procedure_name.begin(), procedure_name.end(), '/'),
-          procedure_name.end());
-      LOG(INFO) << "Deleting procedure for: " << graph_name << ", "
-                << procedure_name;
+      auto graph_id = trim_slash(req->param.at("graph_id"));
+      auto procedure_id = trim_slash(req->param.at("procedure_id"));
+      LOG(INFO) << "Deleting procedure for: " << graph_id << ", "
+                << procedure_id;
       return admin_actor_refs_[dst_executor]
           .delete_procedure(
-              procedure_query_param{std::make_pair(graph_name, procedure_name)})
+              procedure_query_param{std::make_pair(graph_id, procedure_id)})
           .then_wrapped([rep = std::move(rep)](
                             seastar::future<admin_query_result>&& fut) mutable {
             return return_reply_with_result(std::move(rep), std::move(fut));
           });
     } else if (req->_method == "PUT") {
-      if (!req->param.exists("graph_name") ||
-          !req->param.exists("procedure_name")) {
+      if (!req->param.exists("graph_id") ||
+          !req->param.exists("procedure_id")) {
         return seastar::make_exception_future<
             std::unique_ptr<seastar::httpd::reply>>(
-            std::runtime_error("graph_name or procedure_name not given: "));
+            std::runtime_error("graph_id or procedure_id not given: "));
       }
-      auto graph_name = req->param.at("graph_name");
-      graph_name.erase(std::remove(graph_name.begin(), graph_name.end(), '/'),
-                       graph_name.end());
-      auto procedure_name = req->param.at("procedure_name");
-      procedure_name.erase(
-          std::remove(procedure_name.begin(), procedure_name.end(), '/'),
-          procedure_name.end());
-      LOG(INFO) << "Update procedure for: " << graph_name << ", "
-                << procedure_name;
+      auto graph_id = trim_slash(req->param.at("graph_id"));
+      auto procedure_id = trim_slash(req->param.at("procedure_id"));
+      LOG(INFO) << "Update procedure for: " << graph_id << ", " << procedure_id;
       return admin_actor_refs_[dst_executor]
           .update_procedure(update_procedure_query_param{
-              std::make_tuple(graph_name, procedure_name, req->content)})
+              std::make_tuple(graph_id, procedure_id, req->content)})
           .then_wrapped([rep = std::move(rep)](
                             seastar::future<admin_query_result>&& fut) mutable {
             return return_reply_with_result(std::move(rep), std::move(fut));
@@ -311,11 +300,8 @@ class admin_http_service_handler_impl : public seastar::httpd::handler_base {
             std::unique_ptr<seastar::httpd::reply>>(
             std::runtime_error("action is expected for /v1/service/"));
       }
-      auto action = req->param.at("action");
+      auto action = trim_slash(req->param.at("action"));
       LOG(INFO) << "POST with action: " << action;
-      // Remove / from the action
-      action.erase(std::remove(action.begin(), action.end(), '/'),
-                   action.end());
 
       if (action == "start" || action == "restart") {
         return admin_actor_refs_[dst_executor]
@@ -403,6 +389,82 @@ class admin_http_node_handler_impl : public seastar::httpd::handler_base {
   std::vector<admin_actor_ref> admin_actor_refs_;
 };
 
+class admin_http_job_handler_impl : public seastar::httpd::handler_base {
+ public:
+  admin_http_job_handler_impl(uint32_t group_id, uint32_t shard_concurrency)
+      : shard_concurrency_(shard_concurrency), executor_idx_(0) {
+    admin_actor_refs_.reserve(shard_concurrency_);
+    hiactor::scope_builder builder;
+    builder.set_shard(hiactor::local_shard_id())
+        .enter_sub_scope(hiactor::scope<executor_group>(0))
+        .enter_sub_scope(hiactor::scope<hiactor::actor_group>(group_id));
+    for (unsigned i = 0; i < shard_concurrency_; ++i) {
+      admin_actor_refs_.emplace_back(builder.build_ref<admin_actor_ref>(i));
+    }
+  }
+  ~admin_http_job_handler_impl() override = default;
+
+  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
+      const seastar::sstring& path,
+      std::unique_ptr<seastar::httpd::request> req,
+      std::unique_ptr<seastar::httpd::reply> rep) override {
+    auto dst_executor = executor_idx_;
+
+    executor_idx_ = (executor_idx_ + 1) % shard_concurrency_;
+    auto& method = req->_method;
+    if (method == "GET") {
+      if (req->param.exists("job_id")) {
+        auto job_id = trim_slash(req->param.at("job_id"));
+        return admin_actor_refs_[dst_executor]
+            .get_job(query_param{std::move(job_id)})
+            .then_wrapped(
+                [rep = std::move(rep)](
+                    seastar::future<admin_query_result>&& fut) mutable {
+                  return return_reply_with_result(std::move(rep),
+                                                  std::move(fut));
+                });
+      } else {
+        return admin_actor_refs_[dst_executor]
+            .list_jobs(query_param{std::move(req->content)})
+            .then_wrapped(
+                [rep = std::move(rep)](
+                    seastar::future<admin_query_result>&& fut) mutable {
+                  return return_reply_with_result(std::move(rep),
+                                                  std::move(fut));
+                });
+      }
+    } else if (method == "DELETE") {
+      if (!req->param.exists("job_id")) {
+        rep->set_status(seastar::httpd::reply::status_type::bad_request);
+        rep->write_body("application/json",
+                        seastar::sstring("expect field 'job_id' in request"));
+        rep->done();
+        return seastar::make_ready_future<
+            std::unique_ptr<seastar::httpd::reply>>(std::move(rep));
+      }
+      auto job_id = trim_slash(req->param.at("job_id"));
+      return admin_actor_refs_[dst_executor]
+          .cancel_job(query_param{std::move(job_id)})
+          .then_wrapped([rep = std::move(rep)](
+                            seastar::future<admin_query_result>&& fut) mutable {
+            return return_reply_with_result(std::move(rep), std::move(fut));
+          });
+    } else {
+      rep->set_status(seastar::httpd::reply::status_type::bad_request);
+      rep->write_body("application/json",
+                      seastar::sstring("Unsupported method: ") + method);
+      rep->done();
+      return seastar::make_ready_future<std::unique_ptr<seastar::httpd::reply>>(
+          std::move(rep));
+    }
+  }
+
+ private:
+  const uint32_t shard_concurrency_;
+  uint32_t executor_idx_;
+  std::vector<admin_actor_ref> admin_actor_refs_;
+};
+
 admin_http_handler::admin_http_handler(uint16_t http_port)
     : http_port_(http_port) {}
 
@@ -442,11 +504,14 @@ seastar::future<> admin_http_handler::set_routes() {
     auto node_handler = new admin_http_node_handler_impl(
         interactive_admin_group_id, shard_admin_node_concurrency);
 
+    auto job_handler = new admin_http_job_handler_impl(
+        interactive_admin_group_id, shard_admin_job_concurrency);
+
     ////Procedure management ///
     {
       auto match_rule = new seastar::httpd::match_rule(procedures_handler);
       match_rule->add_str("/v1/graph")
-          .add_param("graph_name")
+          .add_param("graph_id")
           .add_str("/procedure");
       // Get All procedures
       r.add(match_rule, seastar::httpd::operation_type::GET);
@@ -457,9 +522,9 @@ seastar::future<> admin_http_handler::set_routes() {
       // Each procedure's handling
       auto match_rule = new seastar::httpd::match_rule(procedures_handler);
       match_rule->add_str("/v1/graph")
-          .add_param("graph_name")
+          .add_param("graph_id")
           .add_str("/procedure")
-          .add_param("procedure_name");
+          .add_param("procedure_id");
       // Get a procedure
       r.add(match_rule, seastar::httpd::operation_type::GET);
       // Delete a procedure
@@ -477,21 +542,19 @@ seastar::future<> admin_http_handler::set_routes() {
 
     // Delete a graph
     r.add(seastar::httpd::operation_type::DELETE,
-          seastar::httpd::url("/v1/graph").remainder("graph_name"),
+          seastar::httpd::url("/v1/graph").remainder("graph_id"),
           admin_graph_handler);
 
     {  // load data to graph
       auto match_rule = new seastar::httpd::match_rule(admin_graph_handler);
       match_rule->add_str("/v1/graph")
-          .add_param("graph_name")
+          .add_param("graph_id")
           .add_str("/dataloading");
       r.add(match_rule, seastar::httpd::operation_type::POST);
     }
     {  // Get Graph Schema
       auto match_rule = new seastar::httpd::match_rule(admin_graph_handler);
-      match_rule->add_str("/v1/graph")
-          .add_param("graph_name")
-          .add_str("/schema");
+      match_rule->add_str("/v1/graph").add_param("graph_id").add_str("/schema");
       r.add(match_rule, seastar::httpd::operation_type::GET);
     }
 
@@ -513,8 +576,8 @@ seastar::future<> admin_http_handler::set_routes() {
       auto test_handler = r.get_handler(seastar::httpd::operation_type::POST,
                                         "/v1/graph/abc/dataloading", params);
       CHECK(test_handler);
-      CHECK(params.exists("graph_name"));
-      CHECK(params.at("graph_name") == "/abc") << params.at("graph_name");
+      CHECK(params.exists("graph_id"));
+      CHECK(params.at("graph_id") == "/abc") << params.at("graph_id");
     }
 
     {
@@ -522,8 +585,8 @@ seastar::future<> admin_http_handler::set_routes() {
       auto test_handler = r.get_handler(seastar::httpd::operation_type::GET,
                                         "/v1/graph/abc/schema", params);
       CHECK(test_handler);
-      CHECK(params.exists("graph_name"));
-      CHECK(params.at("graph_name") == "/abc") << params.at("graph_name");
+      CHECK(params.exists("graph_id"));
+      CHECK(params.at("graph_id") == "/abc") << params.at("graph_id");
     }
 
     {
@@ -531,8 +594,8 @@ seastar::future<> admin_http_handler::set_routes() {
       auto test_handler = r.get_handler(seastar::httpd::operation_type::GET,
                                         "/v1/graph/abc/procedure", params);
       CHECK(test_handler);
-      CHECK(params.exists("graph_name"));
-      CHECK(params.at("graph_name") == "/abc") << params.at("graph_name");
+      CHECK(params.exists("graph_id"));
+      CHECK(params.at("graph_id") == "/abc") << params.at("graph_id");
     }
 
     {
@@ -540,8 +603,8 @@ seastar::future<> admin_http_handler::set_routes() {
       auto test_handler = r.get_handler(seastar::httpd::operation_type::POST,
                                         "/v1/graph/abc/procedure", params);
       CHECK(test_handler);
-      CHECK(params.exists("graph_name"));
-      CHECK(params.at("graph_name") == "/abc") << params.at("graph_name");
+      CHECK(params.exists("graph_id"));
+      CHECK(params.at("graph_id") == "/abc") << params.at("graph_id");
     }
 
     {
@@ -550,11 +613,11 @@ seastar::future<> admin_http_handler::set_routes() {
           r.get_handler(seastar::httpd::operation_type::GET,
                         "/v1/graph/abc/procedure/proce1", params);
       CHECK(test_handler);
-      CHECK(params.exists("graph_name"));
-      CHECK(params.at("graph_name") == "/abc") << params.at("graph_name");
-      CHECK(params.exists("procedure_name"));
-      CHECK(params.at("procedure_name") == "/proce1")
-          << params.at("procedure_name");
+      CHECK(params.exists("graph_id"));
+      CHECK(params.at("graph_id") == "/abc") << params.at("graph_id");
+      CHECK(params.exists("procedure_id"));
+      CHECK(params.at("procedure_id") == "/proce1")
+          << params.at("procedure_id");
       params.clear();
       test_handler = r.get_handler(seastar::httpd::operation_type::DELETE,
                                    "/v1/graph/abc/procedure/proce1", params);
@@ -564,6 +627,19 @@ seastar::future<> admin_http_handler::set_routes() {
       CHECK(test_handler);
     }
 
+    {
+      // job request handling.
+      r.add(seastar::httpd::operation_type::GET, seastar::httpd::url("/v1/job"),
+            job_handler);
+      auto match_rule = new seastar::httpd::match_rule(job_handler);
+
+      match_rule->add_str("/v1/job").add_param("job_id");
+      r.add(match_rule, seastar::httpd::operation_type::GET);
+
+      r.add(seastar::httpd::operation_type::DELETE,
+            seastar::httpd::url("/v1/job").remainder("job_id"), job_handler);
+    }
+
     return seastar::make_ready_future<>();
   });
 }
diff --git a/flex/engines/http_server/options.cc b/flex/engines/http_server/options.cc
index 3a2274598d9b..2f7c0441acf3 100644
--- a/flex/engines/http_server/options.cc
+++ b/flex/engines/http_server/options.cc
@@ -23,6 +23,7 @@ uint32_t shard_adhoc_concurrency = 4;
 uint32_t shard_admin_graph_concurrency = 1;
 uint32_t shard_admin_procedure_concurrency = 1;
 uint32_t shard_admin_node_concurrency = 1;
+uint32_t shard_admin_job_concurrency = 1;
 uint32_t shard_admin_service_concurrency = 1;
 
 }  // namespace server
diff --git a/flex/engines/http_server/options.h b/flex/engines/http_server/options.h
index e6afd886cf24..732e19eecd33 100644
--- a/flex/engines/http_server/options.h
+++ b/flex/engines/http_server/options.h
@@ -39,6 +39,7 @@ extern uint32_t shard_adhoc_concurrency;
 extern uint32_t shard_admin_graph_concurrency;
 extern uint32_t shard_admin_node_concurrency;
 extern uint32_t shard_admin_service_concurrency;
+extern uint32_t shard_admin_job_concurrency;
 extern uint32_t shard_admin_procedure_concurrency;
 
 }  // namespace server
diff --git a/flex/engines/http_server/service/hqps_service.cc b/flex/engines/http_server/service/hqps_service.cc
index 8815307dc66f..fd52b5ce4704 100644
--- a/flex/engines/http_server/service/hqps_service.cc
+++ b/flex/engines/http_server/service/hqps_service.cc
@@ -14,6 +14,7 @@
  */
 #include "flex/engines/http_server/service/hqps_service.h"
 #include "flex/engines/http_server/options.h"
+#include "flex/engines/http_server/workdir_manipulator.h"
 namespace server {
 
 ServiceConfig::ServiceConfig()
@@ -25,7 +26,8 @@ ServiceConfig::ServiceConfig()
       enable_thread_resource_pool(true),
       external_thread_num(2),
       start_admin_service(true),
-      start_compiler(false) {}
+      start_compiler(false),
+      metadata_store_type_(gs::MetadataStoreType::kLocalFile) {}
 
 const std::string HQPSService::DEFAULT_GRAPH_NAME = "modern_graph";
 const std::string HQPSService::DEFAULT_INTERACTIVE_HOME = "/opt/flex/";
@@ -49,9 +51,35 @@ void HQPSService::init(const ServiceConfig& config) {
   if (config.start_admin_service) {
     admin_hdl_ = std::make_unique<admin_http_handler>(config.admin_port);
   }
+
   initialized_.store(true);
   service_config_ = config;
   gs::init_cpu_usage_watch();
+  if (config.start_admin_service) {
+    metadata_store_ = gs::MetadataStoreFactory::Create(
+        config.metadata_store_type_, WorkDirManipulator::GetWorkspace());
+
+    auto res = metadata_store_->Open();
+    if (!res.ok()) {
+      std::cerr << "Failed to open metadata store: "
+                << res.status().error_message() << std::endl;
+      return;
+    }
+    LOG(INFO) << "Metadata store opened successfully.";
+    gs::GraphId default_graph_id = insert_default_graph_meta();
+    auto set_res = metadata_store_->SetRunningGraph(default_graph_id);
+    if (!set_res.ok()) {
+      LOG(FATAL) << "Failed to set running graph: "
+                 << res.status().error_message();
+      return;
+    }
+
+    auto lock_res = metadata_store_->LockGraphIndices(default_graph_id);
+    if (!lock_res.ok()) {
+      LOG(FATAL) << lock_res.status().error_message();
+      return;
+    }
+  }
   if (config.start_compiler) {
     start_compiler_subprocess();
   }
@@ -62,6 +90,10 @@ HQPSService::~HQPSService() {
     actor_sys_->terminate();
   }
   stop_compiler_subprocess();
+  clear_running_graph();
+  if (metadata_store_) {
+    metadata_store_->Close();
+  }
 }
 
 const ServiceConfig& HQPSService::get_service_config() const {
@@ -83,6 +115,10 @@ uint16_t HQPSService::get_query_port() const {
   return 0;
 }
 
+std::shared_ptr<gs::IGraphMetaStore> HQPSService::get_metadata_store() const {
+  return metadata_store_;
+}
+
 gs::Result<seastar::sstring> HQPSService::service_status() {
   if (!is_initialized()) {
     return gs::Result<seastar::sstring>(
@@ -246,4 +282,68 @@ std::string HQPSService::find_interactive_class_path() {
   return "";
 }
 
+gs::GraphId HQPSService::insert_default_graph_meta() {
+  if (!metadata_store_) {
+    LOG(FATAL) << "Metadata store has not been inited!" << std::endl;
+  }
+  // If there is no graph in the metadata store, insert the default graph.
+  auto graph_metas_res = metadata_store_->GetAllGraphMeta();
+  if (!graph_metas_res.ok()) {
+    LOG(FATAL) << "Failed to get graph metas: "
+               << graph_metas_res.status().error_message();
+  }
+  if (!graph_metas_res.value().empty()) {
+    LOG(INFO) << "There are already " << graph_metas_res.value().size()
+              << " graph metas in the metadata store.";
+
+    // return the first graph id
+    return graph_metas_res.value().begin()->id;
+  }
+
+  auto default_graph_name = this->service_config_.default_graph;
+  auto schema_str_res =
+      WorkDirManipulator::GetGraphSchemaString(default_graph_name);
+  if (!schema_str_res.ok()) {
+    LOG(FATAL) << "Failed to get graph schema string: "
+               << schema_str_res.status().error_message();
+  }
+  auto request = gs::CreateGraphMetaRequest::FromJson(schema_str_res.value());
+  request.is_builtin = true;
+  request.data_update_time = gs::GetCurrentTimeStamp();
+
+  auto res = metadata_store_->CreateGraphMeta(request);
+  if (!res.ok()) {
+    LOG(FATAL) << "Failed to insert default graph meta: "
+               << res.status().error_message();
+  }
+
+  auto dst_graph_dir = WorkDirManipulator::GetGraphDir(res.value());
+  auto src_graph_dir = WorkDirManipulator::GetGraphDir(default_graph_name);
+  if (std::filesystem::exists(dst_graph_dir)) {
+    // if the dst_graph_dir is already existed, we do nothing.
+    LOG(INFO) << "Graph dir " << dst_graph_dir << " already exists.";
+  } else {
+    // create soft link
+    std::filesystem::create_symlink(src_graph_dir, dst_graph_dir);
+    LOG(INFO) << "Create soft link from " << src_graph_dir << " to "
+              << dst_graph_dir;
+  }
+
+  LOG(INFO) << "Insert default graph meta successfully, graph_id: "
+            << res.value();
+  return res.value();
+}
+
+void HQPSService::clear_running_graph() {
+  if (!metadata_store_) {
+    std::cerr << "Metadata store has not been inited!" << std::endl;
+    return;
+  }
+  auto res = metadata_store_->ClearRunningGraph();
+  if (!res.ok()) {
+    std::cerr << "Failed to clear running graph: "
+              << res.status().error_message() << std::endl;
+    return;
+  }
+}
 }  // namespace server
diff --git a/flex/engines/http_server/service/hqps_service.h b/flex/engines/http_server/service/hqps_service.h
index 6cf7b64e0ac5..c10213a905c7 100644
--- a/flex/engines/http_server/service/hqps_service.h
+++ b/flex/engines/http_server/service/hqps_service.h
@@ -15,6 +15,7 @@
 #ifndef ENGINES_HTTP_SERVER_HQPS_SERVICE_H_
 #define ENGINES_HTTP_SERVER_HQPS_SERVICE_H_
 
+#include <memory>
 #include <string>
 
 #include "flex/engines/graph_db/database/graph_db.h"
@@ -22,6 +23,8 @@
 #include "flex/engines/http_server/handler/admin_http_handler.h"
 #include "flex/engines/http_server/handler/hqps_http_handler.h"
 #include "flex/engines/http_server/workdir_manipulator.h"
+#include "flex/storages/metadata/graph_meta_store.h"
+#include "flex/storages/metadata/metadata_store_factory.h"
 #include "flex/utils/result.h"
 #include "flex/utils/service_utils.h"
 
@@ -48,6 +51,7 @@ struct ServiceConfig {
   bool start_admin_service;  // Whether to start the admin service or only
                              // start the query service.
   bool start_compiler;
+  gs::MetadataStoreType metadata_store_type_;
 
   // Those has not default value
   std::string default_graph;
@@ -75,6 +79,8 @@ class HQPSService {
 
   uint16_t get_query_port() const;
 
+  std::shared_ptr<gs::IGraphMetaStore> get_metadata_store() const;
+
   gs::Result<seastar::sstring> service_status();
 
   void run_and_wait_for_exit();
@@ -99,6 +105,10 @@ class HQPSService {
   HQPSService() = default;
 
   std::string find_interactive_class_path();
+  // Insert graph meta into metadata store.
+  gs::GraphId insert_default_graph_meta();
+  void open_default_graph();
+  void clear_running_graph();
 
  private:
   std::unique_ptr<actor_system> actor_sys_;
@@ -110,6 +120,8 @@ class HQPSService {
 
   ServiceConfig service_config_;
   boost::process::child compiler_process_;
+  // handler for metadata store
+  std::shared_ptr<gs::IGraphMetaStore> metadata_store_;
 };
 
 }  // namespace server
@@ -143,6 +155,22 @@ struct convert<server::ServiceConfig> {
         LOG(INFO) << "shard_num not found, use default value "
                   << service_config.shard_num;
       }
+
+      auto metadata_store_node = engine_node["metadata_store"];
+      if (metadata_store_node) {
+        auto metadata_store_type = metadata_store_node["type"];
+        if (metadata_store_type) {
+          auto metadata_store_type_str = metadata_store_type.as<std::string>();
+          if (metadata_store_type_str == "file") {
+            service_config.metadata_store_type_ =
+                gs::MetadataStoreType::kLocalFile;
+          } else {
+            LOG(ERROR) << "Unsupported metadata store type: "
+                       << metadata_store_type_str;
+            return false;
+          }
+        }
+      }
     } else {
       LOG(ERROR) << "Fail to find compute_engine configuration";
       return false;
diff --git a/flex/engines/http_server/workdir_manipulator.cc b/flex/engines/http_server/workdir_manipulator.cc
index 4e3aa8366cc0..ecfa789d08c5 100644
--- a/flex/engines/http_server/workdir_manipulator.cc
+++ b/flex/engines/http_server/workdir_manipulator.cc
@@ -32,36 +32,11 @@ void WorkDirManipulator::SetWorkspace(const std::string& path) {
   workspace = path;
 }
 
-void WorkDirManipulator::SetRunningGraph(const std::string& name) {
-  // clear the old RUNNING_GRAPH_FILE_NAME, and write the new one.
-  auto running_graph_file = workspace + "/" + RUNNING_GRAPH_FILE_NAME;
-  try {
-    std::ofstream ofs(running_graph_file,
-                      std::ofstream::out | std::ofstream::trunc);
-    ofs << name;
-    ofs.close();
-    LOG(INFO) << "Successfully set running graph: " << name;
-  } catch (const std::exception& e) {
-    LOG(ERROR) << "Fail to set running graph: " << name
-               << ", error: " << e.what();
-  }
-}
-
-std::string WorkDirManipulator::GetRunningGraph() {
-  auto running_graph_file = workspace + "/" + RUNNING_GRAPH_FILE_NAME;
-  std::ifstream ifs(running_graph_file);
-  if (!ifs.is_open()) {
-    LOG(ERROR) << "Fail to open running graph file: " << running_graph_file;
-    return "";
-  }
-  std::string line;
-  std::getline(ifs, line);
-  return line;
-}
+std::string WorkDirManipulator::GetWorkspace() { return workspace; }
 
 // GraphName can be specified in the config file or in the argument.
-gs::Result<seastar::sstring> WorkDirManipulator::CreateGraph(
-    const YAML::Node& yaml_config) {
+gs::Result<seastar::sstring> WorkDirManipulator::DumpGraphSchema(
+    const gs::GraphId& graph_id, const YAML::Node& yaml_config) {
   // First check graph exits
   if (!yaml_config["name"]) {
     return gs::Result<seastar::sstring>(
@@ -69,12 +44,11 @@ gs::Result<seastar::sstring> WorkDirManipulator::CreateGraph(
                    "Graph name is not specified"),
         seastar::sstring("Graph name is not specified"));
   }
-  auto graph_name = yaml_config["name"].as<std::string>();
 
-  if (is_graph_exist(graph_name)) {
+  if (is_graph_exist(graph_id)) {
     return gs::Result<seastar::sstring>(
         gs::Status(gs::StatusCode::AlreadyExists, "Graph already exists"),
-        seastar::sstring("graph " + graph_name + " already exists"));
+        seastar::sstring("graph " + graph_id + " already exists"));
   }
 
   // First check whether yaml is valid
@@ -85,19 +59,72 @@ gs::Result<seastar::sstring> WorkDirManipulator::CreateGraph(
   }
   auto& schema = schema_result.value();
   // dump schema to file.
-  auto dump_res = dump_graph_schema(yaml_config, graph_name);
+  auto dump_res = dump_graph_schema(yaml_config, graph_id);
   if (!dump_res.ok()) {
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::PermissionError,
         "Fail to dump graph schema: " + dump_res.status().error_message()));
   }
-  VLOG(10) << "Successfully dump graph schema to file: " << graph_name << ", "
-           << GetGraphSchemaPath(graph_name);
+  VLOG(10) << "Successfully dump graph schema to file: " << graph_id << ", "
+           << GetGraphSchemaPath(graph_id);
 
   return gs::Result<seastar::sstring>(
       seastar::sstring("successfully created graph "));
 }
 
+gs::Result<bool> WorkDirManipulator::DumpGraphSchema(
+    const gs::GraphMeta& graph_meta,
+    const std::vector<gs::PluginMeta>& plugin_metas) {
+  auto graph_id = graph_meta.id;
+  if (!is_graph_exist(graph_id)) {
+    return gs::Result<bool>(
+        gs::Status(gs::StatusCode::NotExists, "Graph not exists: " + graph_id),
+        false);
+  }
+  auto graph_schema = graph_meta.ToJson();
+  YAML::Node yaml_node;
+  try {
+    yaml_node = YAML::Load(graph_schema);
+  } catch (const std::exception& e) {
+    return gs::Result<bool>(
+        gs::Status(gs::StatusCode::InternalError,
+                   "Fail to parse graph schema: " + graph_schema +
+                       ", error: " + e.what()),
+        false);
+  }
+  // add plugins which are enabled on the graph.
+  if (!yaml_node["stored_procedures"]) {
+    yaml_node["stored_procedures"] = YAML::Node();
+    // create a new node
+    yaml_node["stored_procedures"]["enable_lists"] = YAML::Node();
+  }
+  auto procedures = yaml_node["stored_procedures"]["enable_lists"];
+  // clear the old enabled list
+  for (const auto& plugin_meta : plugin_metas) {
+    if (plugin_meta.enable) {
+      VLOG(10) << "Add enabled plugin: " << plugin_meta.id;
+      procedures.push_back(plugin_meta.id);
+    }
+  }
+
+  auto schema_res = gs::Schema::LoadFromYamlNode(yaml_node);
+  if (!schema_res.ok()) {
+    return gs::Result<bool>(schema_res.status(), false);
+  }
+  LOG(INFO) << "Finish parsing new schema";
+  // dump schema to file.
+  auto dump_res = dump_graph_schema(yaml_node, graph_id);
+  if (!dump_res.ok()) {
+    return gs::Result<bool>(gs::Status(gs::StatusCode::PermissionError,
+                                       "Fail to dump graph schema: " +
+                                           dump_res.status().error_message()),
+                            false);
+  }
+  VLOG(10) << "Successfully dump graph schema to file: " << graph_id << ", "
+           << GetGraphSchemaPath(graph_id);
+  return gs::Result<bool>(true);
+}
+
 gs::Result<seastar::sstring> WorkDirManipulator::GetGraphSchemaString(
     const std::string& graph_name) {
   if (!is_graph_exist(graph_name)) {
@@ -167,9 +194,7 @@ gs::Result<seastar::sstring> WorkDirManipulator::GetDataDirectory(
   }
   auto data_dir = GetGraphIndicesDir(graph_name);
   if (!std::filesystem::exists(data_dir)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::NotExists,
-        "Graph data directory is expected, but not exists: " + data_dir));
+    std::filesystem::create_directory(data_dir);
   }
   return gs::Result<seastar::sstring>(data_dir);
 }
@@ -208,27 +233,6 @@ gs::Result<seastar::sstring> WorkDirManipulator::ListGraphs() {
 
 gs::Result<seastar::sstring> WorkDirManipulator::DeleteGraph(
     const std::string& graph_name) {
-  if (!is_graph_exist(graph_name)) {
-    return gs::Result<seastar::sstring>(
-        gs::Status(gs::StatusCode::NotExists,
-                   "Graph not exists: " + graph_name),
-        seastar::sstring("graph " + graph_name + " not exists"));
-  }
-  if (is_graph_running(graph_name)) {
-    return gs::Result<seastar::sstring>(
-        gs::Status(gs::StatusCode::IllegalOperation,
-                   "Can not remove a running " + graph_name),
-        seastar::sstring("graph " + graph_name +
-                         " is running, can not be removed"));
-  }
-  if (is_graph_locked(graph_name)) {
-    return gs::Result<seastar::sstring>(
-        gs::Status(gs::StatusCode::IllegalOperation,
-                   "Can not remove graph " + graph_name +
-                       ", since data loading ongoing"),
-        seastar::sstring("Can not remove graph " + graph_name +
-                         ", since data loading ongoing"));
-  }
   // remove the graph directory
   try {
     auto graph_path = get_graph_dir(graph_name);
@@ -245,34 +249,13 @@ gs::Result<seastar::sstring> WorkDirManipulator::DeleteGraph(
 
 gs::Result<seastar::sstring> WorkDirManipulator::LoadGraph(
     const std::string& graph_name, const YAML::Node& yaml_node,
-    int32_t loading_thread_num) {
+    int32_t loading_thread_num, const std::string& dst_indices_dir,
+    std::shared_ptr<gs::IGraphMetaStore> metadata_store) {
   // First check whether graph exists
   if (!is_graph_exist(graph_name)) {
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::NotExists, "Graph not exists: " + graph_name));
   }
-  if (is_graph_locked(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::IllegalOperation,
-        "Graph is locked: " + graph_name +
-            ", either service is running on graph, or graph is loading"));
-  }
-  // Then check graph is already loaded
-  if (is_graph_loaded(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::IllegalOperation,
-        "Graph is already loaded, can not be loaded twice: " + graph_name));
-  }
-  // check is graph locked
-  if (is_graph_running(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::IllegalOperation,
-        "Graph is already running, can not be loaded: " + graph_name));
-  }
-  if (!try_lock_graph(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::IllegalOperation, "Fail to lock graph: " + graph_name));
-  }
 
   // No need to check whether graph exists, because it is checked in LoadGraph
   // First load schema
@@ -304,22 +287,16 @@ gs::Result<seastar::sstring> WorkDirManipulator::LoadGraph(
   auto loading_config = loading_config_res.value();
   std::string temp_file_name = graph_name + "_bulk_loading_config.yaml";
   auto temp_file_path = TMP_DIR + "/" + temp_file_name;
-  auto dump_res = dump_yaml_to_file(yaml_node, temp_file_path);
-  if (!dump_res.ok()) {
-    return gs::Result<seastar::sstring>(
-        gs::Status(gs::StatusCode::InternalError,
-                   "Fail to dump loading config to file: " + temp_file_path +
-                       ", error: " + dump_res.status().error_message()));
-  }
+  RETURN_IF_NOT_OK(dump_yaml_to_file(yaml_node, temp_file_path));
 
-  auto res = LoadGraph(temp_file_path, graph_name, loading_thread_num);
-  if (!res.ok()) {
-    return gs::Result<seastar::sstring>(res.status());
+  auto loading_config_json_str_res = gs::get_json_string_from_yaml(yaml_node);
+  if (!loading_config_json_str_res.ok()) {
+    return loading_config_json_str_res.status();
   }
-  // unlock graph
-  unlock_graph(graph_name);
 
-  return gs::Result<seastar::sstring>(res.status(), res.value());
+  return load_graph_impl(temp_file_path, graph_name, loading_thread_num,
+                         dst_indices_dir, loading_config_json_str_res.value(),
+                         metadata_store);
 }
 
 gs::Result<seastar::sstring> WorkDirManipulator::GetProceduresByGraphName(
@@ -367,13 +344,13 @@ gs::Result<seastar::sstring> WorkDirManipulator::GetProceduresByGraphName(
 
 gs::Result<seastar::sstring>
 WorkDirManipulator::GetProcedureByGraphAndProcedureName(
-    const std::string& graph_name, const std::string& procedure_name) {
-  if (!is_graph_exist(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::NotExists, "Graph not exists: " + graph_name));
+    const std::string& graph_id, const std::string& procedure_id) {
+  if (!is_graph_exist(graph_id)) {
+    return gs::Result<seastar::sstring>(
+        gs::Status(gs::StatusCode::NotExists, "Graph not exists: " + graph_id));
   }
   // get graph schema file, and get procedure lists.
-  auto schema_file = GetGraphSchemaPath(graph_name);
+  auto schema_file = GetGraphSchemaPath(graph_id);
   if (!std::filesystem::exists(schema_file)) {
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::NotExists,
@@ -388,13 +365,13 @@ WorkDirManipulator::GetProcedureByGraphAndProcedureName(
         "Fail to load graph schema: " + schema_file + ", error: " + e.what()));
   }
   // get yaml file in plugin directory.
-  auto plugin_dir = get_graph_plugin_dir(graph_name);
+  auto plugin_dir = get_graph_plugin_dir(graph_id);
   if (!std::filesystem::exists(plugin_dir)) {
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::NotExists,
         "Graph plugin directory is expected, but not exists: " + plugin_dir));
   }
-  auto plugin_file = plugin_dir + "/" + procedure_name + ".yaml";
+  auto plugin_file = plugin_dir + "/" + procedure_id + ".yaml";
   if (!std::filesystem::exists(plugin_file)) {
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::NotExists, "plugin not found " + plugin_file));
@@ -408,38 +385,14 @@ WorkDirManipulator::GetProcedureByGraphAndProcedureName(
         gs::StatusCode::InternalError,
         "Fail to load graph plugin: " + plugin_file + ", error: " + e.what()));
   }
-  plugin_node["enabled"] = false;
-
-  if (schema_node["stored_procedures"]) {
-    auto procedure_node = schema_node["stored_procedures"];
-    if (procedure_node["enable_lists"]) {
-      auto procedures = procedure_node["enable_lists"];
-      if (procedures.IsSequence()) {
-        std::vector<std::string> procedure_list;
-        for (const auto& procedure : procedures) {
-          procedure_list.push_back(procedure.as<std::string>());
-        }
-        if (std::find(procedure_list.begin(), procedure_list.end(),
-                      procedure_name) != procedure_list.end()) {
-          // add enabled: true to the plugin yaml.
-          plugin_node["enabled"] = true;
-        }
-      }
-    } else {
-      LOG(INFO) << "No enabled procedures found: " << graph_name
-                << ", schema file: " << schema_file;
-    }
-  }
-  // yaml_list to string
-  YAML::Emitter emitter;
-  emitter << plugin_node;
-  auto str = emitter.c_str();
-  return gs::Result<seastar::sstring>(std::move(str));
+  return gs::Result<seastar::sstring>(
+      gs::get_json_string_from_yaml(plugin_node).value());
 }
 
 seastar::future<seastar::sstring> WorkDirManipulator::CreateProcedure(
-    const std::string& graph_name, const std::string& parameter,
-    const std::string& engine_config_path) {
+    const std::string& graph_name, const std::string& plugin_id,
+    const nlohmann::json& json, const std::string& engine_config_path) {
+  LOG(INFO) << "Create procedure: " << plugin_id << " on graph: " << graph_name;
   if (!is_graph_exist(graph_name)) {
     return seastar::make_ready_future<seastar::sstring>("Graph not exists: " +
                                                         graph_name);
@@ -455,73 +408,23 @@ seastar::future<seastar::sstring> WorkDirManipulator::CreateProcedure(
     }
   }
   // load parameter as json, and do some check
-  nlohmann::json json;
-  try {
-    json = nlohmann::json::parse(parameter);
-  } catch (const std::exception& e) {
-    return seastar::make_exception_future<seastar::sstring>(
-        "Fail to parse parameter as json: " + parameter);
-  }
   // check required fields is give.
   auto res = create_procedure_sanity_check(json);
   if (!res.ok()) {
     return seastar::make_exception_future<seastar::sstring>(
         res.status().error_message());
   }
+
   LOG(INFO) << "Pass sanity check for procedure: "
             << json["name"].get<std::string>();
   // get procedure name
-  auto procedure_name = json["name"].get<std::string>();
   // check whether procedure already exists.
-  auto plugin_file = plugin_dir + "/" + procedure_name + ".yaml";
+  auto plugin_file = plugin_dir + "/" + plugin_id + ".yaml";
   if (std::filesystem::exists(plugin_file)) {
     return seastar::make_exception_future<seastar::sstring>(
-        "Procedure already exists: " + procedure_name);
+        "Procedure already exists: " + plugin_id);
   }
-  return generate_procedure(json, engine_config_path)
-      .then_wrapped([json](auto&& fut) {
-        try {
-          auto res = fut.get();
-          bool enable = true;  // default enable.
-          if (json.contains("enable")) {
-            if (json["enable"].is_boolean()) {
-              enable = json["enable"].get<bool>();
-            } else if (json["enable"].is_string()) {
-              auto enable_str = json["enable"].get<std::string>();
-              if (enable_str == "true" || enable_str == "True" ||
-                  enable_str == "TRUE") {
-                enable = true;
-              } else {
-                enable = false;
-              }
-            } else {
-              return seastar::make_ready_future<seastar::sstring>(
-                  "Fail to parse enable field: " + json["enable"].dump());
-            }
-          }
-          LOG(INFO) << "Enable: " << std::to_string(enable);
-
-          // If create procedure success, update graph schema (dump to file)
-          // and add to plugin list. this is critical, and should be
-          // transactional.
-          if (enable) {
-            LOG(INFO)
-                << "Procedure is enabled, add to graph schema and plugin list.";
-            return add_procedure_to_graph(json, res);
-          } else {
-            // Not enabled, do nothing.
-            LOG(INFO) << "Procedure is not enabled, do nothing.";
-          }
-
-          return seastar::make_ready_future<seastar::sstring>(
-              seastar::sstring("Successfully create procedure"));
-        } catch (const std::exception& e) {
-          return seastar::make_ready_future<seastar::sstring>(
-              "Fail to generate procedure: " + std::string(e.what()));
-        }
-        return seastar::make_ready_future<seastar::sstring>(
-            "Fail to generate procedure");
-      });
+  return generate_procedure(graph_name, plugin_id, json, engine_config_path);
 }
 
 gs::Result<seastar::sstring> WorkDirManipulator::DeleteProcedure(
@@ -532,59 +435,6 @@ gs::Result<seastar::sstring> WorkDirManipulator::DeleteProcedure(
     return gs::Result<seastar::sstring>(gs::Status(
         gs::StatusCode::NotExists, "Graph not exists: " + graph_name));
   }
-  // delete from graph schema
-  auto schema_file = GetGraphSchemaPath(graph_name);
-  if (!std::filesystem::exists(schema_file)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::NotExists,
-        "Graph schema file is expected, but not exists: " + schema_file));
-  }
-  YAML::Node schema_node;
-  try {
-    schema_node = YAML::LoadFile(schema_file);
-  } catch (const std::exception& e) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::InternalError,
-        "Fail to load graph schema: " + schema_file + ", error: " + e.what()));
-  }
-
-  if (schema_node["stored_procedures"]) {
-    auto procedure_node = schema_node["stored_procedures"];
-    if (procedure_node["enable_lists"]) {
-      auto procedures = procedure_node["enable_lists"];
-      if (procedures.IsSequence()) {
-        std::vector<std::string> procedure_list;
-        for (const auto& procedure : procedures) {
-          procedure_list.push_back(procedure.as<std::string>());
-        }
-        auto it = std::find(procedure_list.begin(), procedure_list.end(),
-                            procedure_name);
-        if (it != procedure_list.end()) {
-          procedure_list.erase(it);
-          procedures = procedure_list;
-          VLOG(1) << "Successfully removed " << procedure_name
-                  << " from procedure list" << gs::to_string(procedure_list);
-          procedure_node["enable_lists"] = procedures;
-          // dump to file.
-          auto dump_res = dump_yaml_to_file(schema_node, schema_file);
-          if (dump_res.ok()) {
-            LOG(INFO) << "Dump graph schema to file: " << schema_file;
-          } else {
-            return gs::Result<seastar::sstring>(gs::Status(
-                gs::StatusCode::InternalError,
-                "Fail to dump graph schema: " + schema_file +
-                    ", error: " + dump_res.status().error_message()));
-          }
-        }
-      } else {
-        VLOG(10) << "No enabled procedures found: " << graph_name
-                 << ", schema file: " << schema_file;
-      }
-    } else {
-      LOG(INFO) << "No enabled procedures found: " << graph_name
-                << ", schema file: " << schema_file;
-    }
-  }
   // remove the plugin file and dynamic lib
   auto plugin_dir = get_graph_plugin_dir(graph_name);
   if (!std::filesystem::exists(plugin_dir)) {
@@ -624,10 +474,6 @@ gs::Result<seastar::sstring> WorkDirManipulator::DeleteProcedure(
 gs::Result<seastar::sstring> WorkDirManipulator::UpdateProcedure(
     const std::string& graph_name, const std::string& procedure_name,
     const std::string& parameters) {
-  if (!is_graph_exist(graph_name)) {
-    return gs::Result<seastar::sstring>(gs::Status(
-        gs::StatusCode::NotExists, "Graph not exists: " + graph_name));
-  }
   // check procedure exits.
   auto plugin_dir = get_graph_plugin_dir(graph_name);
   if (!std::filesystem::exists(plugin_dir)) {
@@ -732,6 +578,10 @@ std::string WorkDirManipulator::GetGraphSchemaPath(
   return get_graph_dir(graph_name) + "/" + GRAPH_SCHEMA_FILE_NAME;
 }
 
+std::string WorkDirManipulator::GetGraphDir(const std::string& graph_name) {
+  return get_graph_dir(graph_name);
+}
+
 std::string WorkDirManipulator::get_graph_lock_file(
     const std::string& graph_name) {
   return get_graph_dir(graph_name) + "/" + LOCK_FILE;
@@ -766,15 +616,14 @@ std::string WorkDirManipulator::GetCompilerLogFile() {
   return log_path;
 }
 
-std::string WorkDirManipulator::get_graph_indices_file(
-    const std::string& graph_name) {
-  return get_graph_dir(graph_name) + GRAPH_INDICES_DIR_NAME + "/" +
-         GRAPH_INDICES_FILE_NAME;
-}
-
-std::string WorkDirManipulator::get_graph_plugin_dir(
-    const std::string& graph_name) {
-  return get_graph_dir(graph_name) + "/" + GRAPH_PLUGIN_DIR_NAME;
+std::string WorkDirManipulator::CommitTempIndices(const std::string& graph_id) {
+  auto temp_indices_dir = GetTempIndicesDir(graph_id);
+  auto indices_dir = GetGraphIndicesDir(graph_id);
+  if (std::filesystem::exists(indices_dir)) {
+    std::filesystem::remove_all(indices_dir);
+  }
+  std::filesystem::rename(temp_indices_dir, indices_dir);
+  return indices_dir;
 }
 
 // graph_name can be a path, first try as it is absolute path, or
@@ -791,37 +640,29 @@ bool WorkDirManipulator::is_graph_exist(const std::string& graph_name) {
   return std::filesystem::exists(graph_path);
 }
 
-bool WorkDirManipulator::is_graph_loaded(const std::string& graph_name) {
-  return std::filesystem::exists(get_graph_indices_file(graph_name));
-}
-
-bool WorkDirManipulator::is_graph_running(const std::string& graph_name) {
-  return GetRunningGraph() == graph_name;
+std::string WorkDirManipulator::GetTempIndicesDir(
+    const std::string& graph_name) {
+  return get_graph_dir(graph_name) + "/" + GRAPH_TEMP_INDICES_DIR_NAME;
 }
 
-bool WorkDirManipulator::is_graph_locked(const std::string& graph_name) {
-  auto lock_file = get_graph_lock_file(graph_name);
-  return std::filesystem::exists(lock_file);
+std::string WorkDirManipulator::CleanTempIndicesDir(
+    const std::string& graph_name) {
+  auto temp_indices_dir = GetTempIndicesDir(graph_name);
+  if (std::filesystem::exists(temp_indices_dir)) {
+    std::filesystem::remove_all(temp_indices_dir);
+  }
+  return temp_indices_dir;
 }
 
-bool WorkDirManipulator::try_lock_graph(const std::string& graph_name) {
-  auto lock_file = get_graph_lock_file(graph_name);
-  if (std::filesystem::exists(lock_file)) {
-    return false;
-  }
-  std::ofstream fout(lock_file);
-  if (!fout.is_open()) {
-    return false;
-  }
-  fout.close();
-  return true;
+std::string WorkDirManipulator::get_graph_indices_file(
+    const std::string& graph_name) {
+  return get_graph_dir(graph_name) + GRAPH_INDICES_DIR_NAME + "/" +
+         GRAPH_INDICES_FILE_NAME;
 }
 
-void WorkDirManipulator::unlock_graph(const std::string& graph_name) {
-  auto lock_file = get_graph_lock_file(graph_name);
-  if (std::filesystem::exists(lock_file)) {
-    std::filesystem::remove(lock_file);
-  }
+std::string WorkDirManipulator::get_graph_plugin_dir(
+    const std::string& graph_name) {
+  return get_graph_dir(graph_name) + "/" + GRAPH_PLUGIN_DIR_NAME;
 }
 
 bool WorkDirManipulator::ensure_graph_dir_exists(
@@ -845,34 +686,123 @@ gs::Result<std::string> WorkDirManipulator::dump_graph_schema(
   if (!fout.is_open()) {
     return {gs::Status(gs::StatusCode::PermissionError, "Fail to open file")};
   }
-  fout << yaml_config;
+  std::string yaml_str;
+  ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(
+      yaml_str, gs::get_yaml_string_from_yaml_node(yaml_config));
+  fout << yaml_str;
   fout.close();
   VLOG(10) << "Successfully dump graph schema to file: " << graph_path;
   return gs::Result<std::string>(gs::Status::OK());
 }
 
-gs::Result<std::string> WorkDirManipulator::LoadGraph(
-    const std::string& config_file_path, const std::string& graph_name,
-    int32_t loading_thread_num) {
-  // TODO: call GRAPH_LOADER_BIN.
-  auto schema_file = GetGraphSchemaPath(graph_name);
-  auto cur_indices_dir = GetGraphIndicesDir(graph_name);
-  // system call to GRAPH_LOADER_BIN schema_file, loading_config,
-  // cur_indices_dir
-  std::string cmd_string = GRAPH_LOADER_BIN + " -g " + schema_file + " -l " +
-                           config_file_path + " -d " + cur_indices_dir + " " +
-                           std::to_string(loading_thread_num);
-  LOG(INFO) << "Call graph_loader: " << cmd_string;
-  auto res = std::system(cmd_string.c_str());
-  if (res != 0) {
-    return gs::Result<std::string>(
-        gs::Status(gs::StatusCode::InternalError,
-                   "Fail to load graph: " + graph_name +
-                       ", error code: " + std::to_string(res)));
+std::string WorkDirManipulator::get_tmp_bulk_loading_job_log_path(
+    const std::string& graph_name) {
+  // file_name = graph_name + current_time + ".log";
+  auto current_time = std::chrono::system_clock::now();
+  auto current_time_str = std::chrono::duration_cast<std::chrono::milliseconds>(
+                              current_time.time_since_epoch())
+                              .count();
+  auto file_name = TMP_DIR + "/" + graph_name + "_" +
+                   std::to_string(current_time_str) + ".log";
+  return file_name;
+}
+
+gs::Result<seastar::sstring> WorkDirManipulator::load_graph_impl(
+    const std::string& config_file_path, const std::string& graph_id,
+    int32_t loading_thread_num, const std::string& dst_indices_dir,
+    const std::string& loading_config_json_str,
+    std::shared_ptr<gs::IGraphMetaStore> metadata_store) {
+  auto schema_file = GetGraphSchemaPath(graph_id);
+  auto final_indices_dir = GetGraphIndicesDir(graph_id);
+  auto bulk_loading_job_log = get_tmp_bulk_loading_job_log_path(graph_id);
+  VLOG(10) << "Bulk loading job log: " << bulk_loading_job_log;
+  std::stringstream ss;
+  ss << GRAPH_LOADER_BIN << " -g " << schema_file << " -l " << config_file_path
+     << " -d " << dst_indices_dir << " -p "
+     << std::to_string(loading_thread_num);
+  auto cmd_string = ss.str();
+  VLOG(10) << "Call graph_loader: " << cmd_string;
+
+  gs::JobId job_id;
+  auto fut =
+      hiactor::thread_resource_pool::submit_work(
+          [&job_id, copied_graph_id = graph_id, cmd_string_copied = cmd_string,
+           tmp_indices_dir_copied = dst_indices_dir,
+           final_indices_dir_copied = final_indices_dir,
+           bulk_loading_job_log_copied = bulk_loading_job_log,
+           loading_config_json_str_copied = loading_config_json_str,
+           metadata_store = metadata_store]() mutable {
+            boost::process::child child_handle(
+                cmd_string_copied,
+                boost::process::std_out > bulk_loading_job_log_copied,
+                boost::process::std_err > bulk_loading_job_log_copied);
+            int32_t pid = child_handle.id();
+
+            auto create_job_req = gs::CreateJobMetaRequest::NewRunning(
+                copied_graph_id, pid, bulk_loading_job_log_copied,
+                "BULK_LOADING");
+            auto create_job_res = metadata_store->CreateJobMeta(create_job_req);
+            if (!create_job_res.ok()) {
+              LOG(ERROR) << "Fail to create job meta for graph: "
+                         << copied_graph_id;
+              return gs::Result<seastar::sstring>(create_job_res.status());
+            }
+            job_id = create_job_res.value();
+            LOG(INFO) << "Successfully created job: " << job_id;
+            auto internal_job_id = job_id;
+            LOG(INFO) << "Waiting exiting...";
+
+            child_handle.wait();
+            auto res = child_handle.exit_code();
+            VLOG(10) << "Graph loader finished, job_id: " << internal_job_id
+                     << ", res: " << res;
+
+            LOG(INFO) << "Updating graph meta";
+            auto exit_request = gs::UpdateJobMetaRequest::NewFinished(res);
+            auto update_exit_res =
+                metadata_store->UpdateJobMeta(internal_job_id, exit_request);
+            if (!update_exit_res.ok()) {
+              LOG(ERROR) << "Fail to update job status to finished, job_id: "
+                         << internal_job_id;
+              return gs::Result<seastar::sstring>(update_exit_res.status());
+            }
+
+            gs::UpdateGraphMetaRequest update_graph_meta_req(
+                gs::GetCurrentTimeStamp(), loading_config_json_str_copied);
+            // Note that this call is also transactional
+            auto update_graph_meta_res = metadata_store->UpdateGraphMeta(
+                copied_graph_id, update_graph_meta_req);
+
+            if (!update_graph_meta_res.ok()) {
+              LOG(INFO) << "Fail to update graph meta for graph: "
+                        << copied_graph_id;
+              WorkDirManipulator::CleanTempIndicesDir(copied_graph_id);
+              return gs::Result<seastar::sstring>(
+                  update_graph_meta_res.status());
+            }
+
+            LOG(INFO) << "Committing temp indices for graph: "
+                      << copied_graph_id;
+            WorkDirManipulator::CommitTempIndices(copied_graph_id);
+            return gs::Result<seastar::sstring>(
+                "Finish Loading and commit temp "
+                "indices");
+          })
+          .then_wrapped([copied_graph_id = graph_id,
+                         metadata_store = metadata_store](auto&& f) {
+            // the destructor of lock_file will unlock the graph.
+            // the destructor of decrementer will decrement the job count.
+            metadata_store->UnlockGraphIndices(copied_graph_id);
+            return gs::Result<seastar::sstring>("Finish unlock graph");
+          });
+
+  while (job_id.empty()) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }
 
-  return gs::Result<std::string>(
-      gs::Status::OK(), "Successfully load data to graph: " + graph_name);
+  LOG(INFO) << "Successfully created job: " << job_id;
+
+  return gs::Result<seastar::sstring>(job_id);
 }
 
 gs::Result<seastar::sstring> WorkDirManipulator::create_procedure_sanity_check(
@@ -903,6 +833,7 @@ gs::Result<seastar::sstring> WorkDirManipulator::create_procedure_sanity_check(
 }
 
 seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
+    const std::string& graph_id, const std::string& plugin_id,
     const nlohmann::json& json, const std::string& engine_config_path) {
   VLOG(10) << "Generate procedure: " << json.dump();
   auto codegen_bin = gs::find_codegen_bin();
@@ -914,9 +845,9 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
   }
   // dump json["query"] to file.
   auto query = json["query"].get<std::string>();
-  auto name = json["name"].get<std::string>();
+  // auto name = json["name"].get<std::string>();
   auto type = json["type"].get<std::string>();
-  auto bounded_graph = json["bound_graph"].get<std::string>();
+  std::string query_name = plugin_id;
   std::string procedure_desc;
   if (json.contains("description")) {
     procedure_desc = json["description"].get<std::string>();
@@ -925,9 +856,9 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
   }
   std::string query_file;
   if (type == "cypher" || type == "CYPHER") {
-    query_file = temp_codegen_directory + "/" + name + ".cypher";
+    query_file = temp_codegen_directory + "/" + plugin_id + ".cypher";
   } else if (type == "CPP" || type == "cpp") {
-    query_file = temp_codegen_directory + "/" + name + ".cpp";
+    query_file = temp_codegen_directory + "/" + plugin_id + ".cpp";
   } else {
     return seastar::make_exception_future<seastar::sstring>(
         "Procedure type is not supported: " + type);
@@ -947,26 +878,26 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
                            ", error: " + std::string(e.what())));
   }
 
-  if (!is_graph_exist(bounded_graph)) {
+  if (!is_graph_exist(graph_id)) {
     return seastar::make_exception_future<seastar::sstring>(
-        std::runtime_error("Graph not exists: " + bounded_graph));
+        std::runtime_error("Graph not exists: " + graph_id));
   }
-  auto output_dir = get_graph_plugin_dir(bounded_graph);
+  auto output_dir = get_graph_plugin_dir(graph_id);
   if (!std::filesystem::exists(output_dir)) {
     std::filesystem::create_directory(output_dir);
   }
-  auto schema_path = GetGraphSchemaPath(bounded_graph);
+  auto schema_path = GetGraphSchemaPath(graph_id);
 
   return CodegenProxy::CallCodegenCmd(
-             codegen_bin, query_file, name, temp_codegen_directory, output_dir,
-             schema_path, engine_config_path, procedure_desc)
-      .then_wrapped([name, output_dir](auto&& f) {
+             codegen_bin, query_file, query_name, temp_codegen_directory,
+             output_dir, schema_path, engine_config_path, procedure_desc)
+      .then_wrapped([plugin_id = plugin_id, output_dir](auto&& f) {
         try {
           auto res = f.get();
           std::string so_file;
           {
             std::stringstream ss;
-            ss << output_dir << "/lib" << name << ".so";
+            ss << output_dir << "/lib" << plugin_id << ".so";
             so_file = ss.str();
           }
           VLOG(10) << "Check so file: " << so_file;
@@ -980,7 +911,7 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
           std::string yaml_file;
           {
             std::stringstream ss;
-            ss << output_dir << "/" << name << ".yaml";
+            ss << output_dir << "/" << plugin_id << ".yaml";
             yaml_file = ss.str();
           }
           LOG(INFO) << "Check yaml file: " << yaml_file;
@@ -991,7 +922,7 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
                     yaml_file));
           }
           return seastar::make_ready_future<seastar::sstring>(
-              seastar::sstring{yaml_file});
+              seastar::sstring(plugin_id));
         } catch (const std::exception& e) {
           LOG(ERROR) << "Fail to generate procedure, error: " << e.what();
           return seastar::make_exception_future<seastar::sstring>(
@@ -1005,75 +936,6 @@ seastar::future<seastar::sstring> WorkDirManipulator::generate_procedure(
       });
 }
 
-seastar::future<seastar::sstring> WorkDirManipulator::add_procedure_to_graph(
-    const nlohmann::json& json, const std::string& proc_yaml_config_file) {
-  try {
-    YAML::Node proc_config_node;
-    proc_config_node = YAML::LoadFile(proc_yaml_config_file);
-  } catch (const std::exception& e) {
-    return seastar::make_exception_future<seastar::sstring>(gs::Status(
-        gs::StatusCode::InternalError,
-        "Fail to load procedure config file: " + proc_yaml_config_file +
-            ", error: " + e.what()));
-  }
-  // get graph_name from json
-  auto graph_name = json["bound_graph"].get<std::string>();
-  auto proc_name = json["name"].get<std::string>();
-  if (proc_name.empty()) {
-    return seastar::make_exception_future<seastar::sstring>(
-        "Procedure name is empty, can not add to graph: " + graph_name);
-  }
-  // get graph schema file
-  auto graph_schema_file = GetGraphSchemaPath(graph_name);
-  // load graph schema
-  YAML::Node schema_node;
-  try {
-    schema_node = YAML::LoadFile(graph_schema_file);
-  } catch (const std::exception& e) {
-    return seastar::make_exception_future<seastar::sstring>(
-        "Fail to load graph schema: " + graph_schema_file +
-        ", error: " + e.what());
-  }
-  // get plugin list
-  if (!schema_node) {
-    return seastar::make_exception_future<seastar::sstring>(
-        "Graph schema is empty, can not add procedure to graph: " + graph_name);
-  }
-  if (!schema_node["stored_procedures"]) {
-    schema_node["stored_procedures"] = YAML::Node(YAML::NodeType::Map);
-  }
-  auto stored_procedures = schema_node["stored_procedures"];
-  if (!stored_procedures["enable_lists"]) {
-    stored_procedures["enable_lists"] = YAML::Node(YAML::NodeType::Sequence);
-  }
-  auto enable_lists = stored_procedures["enable_lists"];
-  // check whether procedure is already in the list, if so, then we raise
-  // error: procedure already exists.
-  for (const auto& item : enable_lists) {
-    if (item.as<std::string>() == proc_name) {
-      return seastar::make_exception_future<seastar::sstring>(
-          "Procedure already exists in graph: " + graph_name);
-    }
-  }
-  enable_lists.push_back(proc_name);
-  // dump schema to file
-  try {
-    std::ofstream fout(graph_schema_file);
-    if (!fout.is_open()) {
-      return seastar::make_exception_future<seastar::sstring>(
-          "Fail to open graph schema file: " + graph_schema_file);
-    }
-    fout << schema_node;
-    fout.close();
-  } catch (const std::exception& e) {
-    return seastar::make_exception_future<seastar::sstring>(
-        "Fail to dump graph schema to file: " + graph_schema_file +
-        ", error: " + e.what());
-  }
-  return seastar::make_ready_future<seastar::sstring>(
-      seastar::sstring("Successfully create procedure"));
-}
-
 gs::Result<seastar::sstring> WorkDirManipulator::get_all_procedure_yamls(
     const std::string& graph_name,
     const std::vector<std::string>& procedure_names) {
@@ -1329,6 +1191,8 @@ const std::string WorkDirManipulator::GRAPH_SCHEMA_FILE_NAME = "graph.yaml";
 const std::string WorkDirManipulator::GRAPH_INDICES_FILE_NAME =
     "init_snapshot.bin";
 const std::string WorkDirManipulator::GRAPH_INDICES_DIR_NAME = "indices";
+const std::string WorkDirManipulator::GRAPH_TEMP_INDICES_DIR_NAME =
+    "temp_indices";
 const std::string WorkDirManipulator::GRAPH_PLUGIN_DIR_NAME = "plugins";
 const std::string WorkDirManipulator::CONF_ENGINE_CONFIG_FILE_NAME =
     "engine_config.yaml";
diff --git a/flex/engines/http_server/workdir_manipulator.h b/flex/engines/http_server/workdir_manipulator.h
index 4627c513eef5..5c3212aef036 100644
--- a/flex/engines/http_server/workdir_manipulator.h
+++ b/flex/engines/http_server/workdir_manipulator.h
@@ -16,20 +16,24 @@
 #ifndef ENGINES_HTTP_SERVER_WORKDIR_MANIPULATOR_H_
 #define ENGINES_HTTP_SERVER_WORKDIR_MANIPULATOR_H_
 
-#include <yaml-cpp/yaml.h>
-#include <boost/property_tree/json_parser.hpp>
-#include <seastar/core/future.hh>
-#include <seastar/core/sstring.hh>
+#include <filesystem>
 #include <string>
+
 #include "flex/engines/graph_db/database/graph_db.h"
 #include "flex/engines/http_server/types.h"
+#include "flex/storages/metadata/graph_meta_store.h"
 #include "flex/storages/rt_mutable_graph/loading_config.h"
 #include "flex/storages/rt_mutable_graph/schema.h"
 #include "flex/utils/result.h"
 #include "flex/utils/service_utils.h"
 #include "flex/utils/yaml_utils.h"
 
-#include <filesystem>
+#include <seastar/core/future.hh>
+#include <seastar/core/sstring.hh>
+
+#include <yaml-cpp/yaml.h>
+#include <boost/process.hpp>
+#include <boost/property_tree/json_parser.hpp>
 #include "nlohmann/json.hpp"
 
 namespace server {
@@ -48,6 +52,9 @@ class WorkDirManipulator {
   static const std::string GRAPH_SCHEMA_FILE_NAME;
   static const std::string GRAPH_INDICES_FILE_NAME;
   static const std::string GRAPH_INDICES_DIR_NAME;
+  // The temp directory for the graph, used to store the indices during the
+  // loading process. Cleaned after the graph is loaded.
+  static const std::string GRAPH_TEMP_INDICES_DIR_NAME;
   static const std::string GRAPH_PLUGIN_DIR_NAME;
   static const std::string CONF_ENGINE_CONFIG_FILE_NAME;
   static const std::string RUNNING_GRAPH_FILE_NAME;
@@ -56,15 +63,25 @@ class WorkDirManipulator {
 
   static void SetWorkspace(const std::string& workspace_path);
 
-  static void SetRunningGraph(const std::string& graph_name);
-
-  static std::string GetRunningGraph();
+  static std::string GetWorkspace();
 
   /**
    * @brief Create a graph with a given name and config.
-   * @param boost_ptree The config of the graph.
+   * @param yaml_node The config of the graph.
    */
-  static gs::Result<seastar::sstring> CreateGraph(const YAML::Node& yaml_node);
+  static gs::Result<seastar::sstring> DumpGraphSchema(
+      const gs::GraphId& graph_id, const YAML::Node& yaml_node);
+
+  /**
+   * @brief Dump a new version of the graph schema on disk.
+   * @param graph_meta The graph meta.
+   * @param plugin_metas The plugin metas.
+   * @return A boolean result.
+   * @note This method will dump the graph schema to the disk.
+   */
+  static gs::Result<bool> DumpGraphSchema(
+      const gs::GraphMeta& graph_meta,
+      const std::vector<gs::PluginMeta>& plugin_metas);
 
   /**
    * @brief Get a graph with a given name.
@@ -97,19 +114,10 @@ class WorkDirManipulator {
    * @param yaml_node The config of the graph.
    * @param loading_thread_num The number of threads to load the graph.
    */
-  static gs::Result<seastar::sstring> LoadGraph(const std::string& graph_name,
-                                                const YAML::Node& yaml_node,
-                                                int32_t loading_thread_num);
-
-  /**
-   * @brief Load a graph with a given name and config.
-   * @param yaml_config_file The config file of the graph.
-   * @param yaml_node The config of the graph.
-   * @param loading_thread_num The number of threads to load the graph.
-   */
-  static gs::Result<std::string> LoadGraph(const std::string& yaml_config_file,
-                                           const std::string& graph_name,
-                                           int32_t thread_num);
+  static gs::Result<seastar::sstring> LoadGraph(
+      const std::string& graph_name, const YAML::Node& yaml_node,
+      int32_t loading_thread_num, const std::string& dst_indices_dir,
+      std::shared_ptr<gs::IGraphMetaStore> metadata_store);
 
   /**
    * @brief Get all procedures bound to the graph.
@@ -128,8 +136,8 @@ class WorkDirManipulator {
       const std::string& graph_name, const std::string& procedure_name);
 
   static seastar::future<seastar::sstring> CreateProcedure(
-      const std::string& graph_name, const std::string& parameter,
-      const std::string& engine_config_path);
+      const std::string& graph_name, const std::string& plugin_id,
+      const nlohmann::json& json, const std::string& engine_config_path);
 
   static gs::Result<seastar::sstring> DeleteProcedure(
       const std::string& graph_name, const std::string& procedure_name);
@@ -143,13 +151,36 @@ class WorkDirManipulator {
 
   static std::string GetGraphSchemaPath(const std::string& graph_name);
 
+  static std::string GetGraphDir(const std::string& graph_name);
+
   static std::string GetGraphIndicesDir(const std::string& graph_name);
 
   static std::string GetLogDir();
 
   static std::string GetCompilerLogFile();
+  // Return a unique temp dir for the graph.
+  static std::string GetTempIndicesDir(const std::string& graph_name);
+
+  static std::string CleanTempIndicesDir(const std::string& graph_name);
+
+  // Move the temp indices to the graph indices dir.
+  static std::string CommitTempIndices(const std::string& graph_name);
 
  private:
+  static std::string get_tmp_bulk_loading_job_log_path(
+      const std::string& graph_name);
+  /**
+   * @brief Load a graph with a given name and config.
+   * @param yaml_config_file The config file of the graph.
+   * @param yaml_node The config of the graph.
+   * @param loading_thread_num The number of threads to load the graph.
+   */
+  static gs::Result<seastar::sstring> load_graph_impl(
+      const std::string& yaml_config_file, const std::string& graph_name,
+      int32_t thread_num, const std::string& dst_indices_dir,
+      const std::string& loading_config_json_str,
+      std::shared_ptr<gs::IGraphMetaStore> metadata_store);
+
   static gs::Result<seastar::sstring> create_procedure_sanity_check(
       const nlohmann::json& json);
 
@@ -163,16 +194,6 @@ class WorkDirManipulator {
 
   static bool is_graph_exist(const std::string& graph_name);
 
-  static bool is_graph_loaded(const std::string& graph_name);
-
-  static bool is_graph_running(const std::string& graph_name);
-
-  static bool is_graph_locked(const std::string& graph_name);
-
-  static bool try_lock_graph(const std::string& graph_name);
-
-  static void unlock_graph(const std::string& graph_name);
-
   static bool ensure_graph_dir_exists(const std::string& graph_name);
 
   static gs::Result<std::string> dump_graph_schema(
@@ -180,11 +201,9 @@ class WorkDirManipulator {
 
   // Generate the procedure, return the generated yaml config.
   static seastar::future<seastar::sstring> generate_procedure(
+      const std::string& graph_id, const std::string& plugin_id,
       const nlohmann::json& json, const std::string& engine_config_path);
 
-  static seastar::future<seastar::sstring> add_procedure_to_graph(
-      const nlohmann::json& json, const std::string& proc_yaml_config);
-
   // Get all the procedure yaml configs in plugins directory, add additional
   // enabled:false to each config.
   static gs::Result<seastar::sstring> get_all_procedure_yamls(
diff --git a/flex/scripts/install_dependencies.sh b/flex/scripts/install_dependencies.sh
index 30ed57816e40..869e7f39b50f 100755
--- a/flex/scripts/install_dependencies.sh
+++ b/flex/scripts/install_dependencies.sh
@@ -11,7 +11,7 @@ if [ $# -eq 1 ]; then
 fi
 echo "parallelism: $parallelism"
 
-sudo apt install -y \
+sudo apt-get update && sudo apt install -y \
       ninja-build ragel libhwloc-dev libnuma-dev libpciaccess-dev vim wget curl \
       git g++ libunwind-dev libgoogle-glog-dev cmake libopenmpi-dev default-jdk libcrypto++-dev \
       libboost-all-dev libxml2-dev protobuf-compiler libprotobuf-dev libncurses5-dev libcurl4-openssl-dev
diff --git a/flex/storages/CMakeLists.txt b/flex/storages/CMakeLists.txt
index e91cb8ca895a..f4bfd1e2cef6 100644
--- a/flex/storages/CMakeLists.txt
+++ b/flex/storages/CMakeLists.txt
@@ -1,3 +1,4 @@
 
 add_subdirectory(rt_mutable_graph)
 add_subdirectory(immutable_graph)
+add_subdirectory(metadata)
diff --git a/flex/storages/metadata/CMakeLists.txt b/flex/storages/metadata/CMakeLists.txt
new file mode 100644
index 000000000000..5a146c480e45
--- /dev/null
+++ b/flex/storages/metadata/CMakeLists.txt
@@ -0,0 +1,10 @@
+
+file(GLOB_RECURSE METADATA_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
+
+add_library(flex_metadata_store SHARED ${METADATA_SRC_FILES})
+target_link_libraries(flex_metadata_store ${YAML_CPP_LIBRARIES} flex_utils)
+install_flex_target(flex_metadata_store)
+
+
+
+
diff --git a/flex/storages/metadata/default_graph_meta_store.cc b/flex/storages/metadata/default_graph_meta_store.cc
new file mode 100644
index 000000000000..eacbc4446b31
--- /dev/null
+++ b/flex/storages/metadata/default_graph_meta_store.cc
@@ -0,0 +1,458 @@
+/** 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/metadata/default_graph_meta_store.h"
+
+namespace gs {
+
+DefaultGraphMetaStore::DefaultGraphMetaStore(
+    std::unique_ptr<IMetaStore> base_store)
+    : base_store_(std::move(base_store)) {
+  // Clear previous context, in case of dirty data.
+  ClearRunningGraph();
+  clear_locks();
+}
+
+DefaultGraphMetaStore::~DefaultGraphMetaStore() {
+  if (base_store_ != nullptr) {
+    base_store_->Close();
+  }
+  auto res = Close();
+  if (!res.ok()) {
+    LOG(ERROR) << "Fail to close DefaultGraphMetaStore: "
+               << res.status().error_message();
+  }
+}
+
+Result<bool> DefaultGraphMetaStore::Open() { return base_store_->Open(); }
+
+Result<bool> DefaultGraphMetaStore::Close() {
+  RETURN_IF_NOT_OK(ClearRunningGraph());
+  RETURN_IF_NOT_OK(clear_locks());
+  return base_store_->Close();
+}
+
+Result<GraphId> DefaultGraphMetaStore::CreateGraphMeta(
+    const CreateGraphMetaRequest& request) {
+  GraphId graph_id;
+  ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(
+      graph_id, base_store_->CreateMeta(GRAPH_META, request.ToString()));
+  return Result<GraphId>(graph_id);
+}
+
+Result<GraphMeta> DefaultGraphMetaStore::GetGraphMeta(const GraphId& graph_id) {
+  auto res = base_store_->GetMeta(GRAPH_META, graph_id);
+  if (!res.ok()) {
+    return Result<GraphMeta>(
+        Status(res.status().error_code(), "Graph not exits"));
+  }
+  std::string meta_str = res.move_value();
+  auto meta = GraphMeta::FromJson(meta_str);
+  meta.id = graph_id;
+  return Result<GraphMeta>(meta);
+}
+
+Result<std::vector<GraphMeta>> DefaultGraphMetaStore::GetAllGraphMeta() {
+  auto res = base_store_->GetAllMeta(GRAPH_META);
+  if (!res.ok()) {
+    return Result<std::vector<GraphMeta>>(res.status());
+  }
+  std::vector<GraphMeta> metas;
+  for (auto& pair : res.move_value()) {
+    auto meta = GraphMeta::FromJson(pair.second);
+    meta.id = pair.first;
+    metas.push_back(meta);
+  }
+  return Result<std::vector<GraphMeta>>(metas);
+}
+
+Result<bool> DefaultGraphMetaStore::DeleteGraphMeta(const GraphId& graph_id) {
+  return base_store_->DeleteMeta(GRAPH_META, graph_id);
+}
+
+Result<bool> DefaultGraphMetaStore::UpdateGraphMeta(
+    const GraphId& graph_id, const UpdateGraphMetaRequest& request) {
+  return base_store_->UpdateMeta(
+      GRAPH_META, graph_id, [graph_id, &request](const std::string& old_meta) {
+        nlohmann::json json;
+        try {
+          json = nlohmann::json::parse(old_meta);
+        } catch (const std::exception& e) {
+          LOG(ERROR) << "Fail to parse old graph meta:" << e.what();
+          return Result<std::string>(Status(StatusCode::InternalError,
+                                            "Fail to parse old graph meta"));
+        }
+        auto graph_meta = GraphMeta::FromJson(json);
+        if (request.graph_name.has_value()) {
+          graph_meta.name = request.graph_name.value();
+        }
+        if (request.description.has_value()) {
+          graph_meta.description = request.description.value();
+        }
+        if (request.data_update_time.has_value()) {
+          graph_meta.data_update_time = request.data_update_time.value();
+        }
+        if (request.data_import_config.has_value()) {
+          graph_meta.data_import_config = request.data_import_config.value();
+        }
+        return Result<std::string>(graph_meta.ToJson());
+      });
+}
+
+Result<PluginId> DefaultGraphMetaStore::CreatePluginMeta(
+    const CreatePluginMetaRequest& request) {
+  PluginId plugin_id;
+  if (request.id.has_value()) {
+    ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(
+        plugin_id, base_store_->CreateMeta(PLUGIN_META, request.id.value(),
+                                           request.ToString()));
+  } else {
+    ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(
+        plugin_id, base_store_->CreateMeta(PLUGIN_META, request.ToString()));
+  }
+  return Result<PluginId>(plugin_id);
+}
+
+Result<PluginMeta> DefaultGraphMetaStore::GetPluginMeta(
+    const GraphId& graph_id, const PluginId& plugin_id) {
+  auto res = base_store_->GetMeta(PLUGIN_META, plugin_id);
+  if (!res.ok()) {
+    return Result<PluginMeta>(res.status());
+  }
+  std::string meta_str = res.move_value();
+  auto meta = PluginMeta::FromJson(meta_str);
+  meta.id = plugin_id;
+  return Result<PluginMeta>(meta);
+}
+
+Result<std::vector<PluginMeta>> DefaultGraphMetaStore::GetAllPluginMeta(
+    const GraphId& graph_id) {
+  auto res = base_store_->GetAllMeta(PLUGIN_META);
+  if (!res.ok()) {
+    return Result<std::vector<PluginMeta>>(res.status());
+  }
+  std::vector<PluginMeta> metas;
+  VLOG(10) << "Found plugin metas: " << res.move_value().size();
+  for (auto& pair : res.move_value()) {
+    auto plugin_meta = PluginMeta::FromJson(pair.second);
+    // the key is id.
+    plugin_meta.id = pair.first;
+    if (plugin_meta.graph_id == graph_id) {
+      metas.push_back(plugin_meta);
+    }
+  }
+  VLOG(10) << "Found plugin metas belong to graph " << graph_id << ": "
+           << metas.size();
+  return Result<std::vector<PluginMeta>>(metas);
+}
+
+Result<bool> DefaultGraphMetaStore::DeletePluginMeta(
+    const GraphId& graph_id, const PluginId& plugin_id) {
+  return base_store_->DeleteMeta(PLUGIN_META, plugin_id);
+}
+
+Result<bool> DefaultGraphMetaStore::DeletePluginMetaByGraphId(
+    const GraphId& graph_id) {
+  // get all plugin meta, and get the plugin_ids which belong to graph graph_id
+  auto res = base_store_->GetAllMeta(PLUGIN_META);
+  if (!res.ok()) {
+    return Result<bool>(res.status());
+  }
+  std::vector<PluginId> plugin_ids;
+  for (auto& meta_str : res.value()) {
+    auto plugin_meta = PluginMeta::FromJson(meta_str);
+    if (plugin_meta.graph_id == graph_id) {
+      plugin_ids.push_back(plugin_meta.id);
+    }
+  }
+  VLOG(10) << "Found plugin_ids: " << plugin_ids.size();
+  for (auto& plugin_id : plugin_ids) {
+    RETURN_IF_NOT_OK(base_store_->DeleteMeta(PLUGIN_META, plugin_id));
+  }
+  return Result<bool>(true);
+}
+
+Result<bool> DefaultGraphMetaStore::UpdatePluginMeta(
+    const GraphId& graph_id, const PluginId& plugin_id,
+    const UpdatePluginMetaRequest& update_request) {
+  return base_store_->UpdateMeta(
+      PLUGIN_META, plugin_id,
+      [graph_id, plugin_id, &update_request](const std::string& old_meta) {
+        nlohmann::json json;
+        try {
+          json = nlohmann::json::parse(old_meta);
+        } catch (const std::exception& e) {
+          LOG(ERROR) << "Fail to parse old plugin meta:" << e.what();
+          return Result<std::string>(Status(StatusCode::InternalError,
+                                            "Fail to parse old plugin meta"));
+        }
+        auto plugin_meta = PluginMeta::FromJson(json);
+        if (plugin_meta.graph_id != graph_id) {
+          return Result<std::string>(Status(gs::StatusCode::InternalError,
+                                            "Plugin not belongs to the graph"));
+        }
+        if (update_request.graph_id.has_value()) {
+          if (update_request.graph_id.value() != graph_id) {
+            return Result<std::string>(
+                Status(gs::StatusCode::IllegalOperation,
+                       "The plugin_id in update payload is not "
+                       "the same with original"));
+          }
+        }
+        if (update_request.name.has_value()) {
+          plugin_meta.name = update_request.name.value();
+        }
+        if (update_request.description.has_value()) {
+          plugin_meta.description = update_request.description.value();
+        }
+        if (update_request.params.has_value()) {
+          plugin_meta.params = update_request.params.value();
+        }
+        if (update_request.returns.has_value()) {
+          plugin_meta.returns = update_request.returns.value();
+        }
+        if (update_request.library.has_value()) {
+          plugin_meta.library = update_request.library.value();
+        }
+        if (update_request.option.has_value()) {
+          plugin_meta.option = update_request.option.value();
+        }
+        if (update_request.enable.has_value()) {
+          plugin_meta.enable = update_request.enable.value();
+        }
+        if (update_request.update_time.has_value()) {
+          plugin_meta.update_time = update_request.update_time.value();
+        }
+        return Result<std::string>(plugin_meta.ToJson());
+      });
+}
+
+Result<JobId> DefaultGraphMetaStore::CreateJobMeta(
+    const CreateJobMetaRequest& request) {
+  JobId job_id;
+  ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(
+      job_id, base_store_->CreateMeta(JOB_META, request.ToString()));
+  return Result<JobId>(job_id);
+}
+
+Result<JobMeta> DefaultGraphMetaStore::GetJobMeta(const JobId& job_id) {
+  auto res = base_store_->GetMeta(JOB_META, job_id);
+  if (!res.ok()) {
+    return Result<JobMeta>(res.status());
+  }
+  std::string meta_str = res.move_value();
+  auto job = JobMeta::FromJson(meta_str);
+  job.id = job_id;
+  return Result<JobMeta>(job);
+}
+
+Result<std::vector<JobMeta>> DefaultGraphMetaStore::GetAllJobMeta() {
+  auto res = base_store_->GetAllMeta(JOB_META);
+  if (!res.ok()) {
+    return Result<std::vector<JobMeta>>(res.status());
+  }
+  std::vector<JobMeta> metas;
+  for (auto& pair : res.move_value()) {
+    auto meta = JobMeta::FromJson(pair.second);
+    meta.id = pair.first;
+    metas.push_back(meta);
+  }
+  return Result<std::vector<JobMeta>>(metas);
+}
+
+Result<bool> DefaultGraphMetaStore::DeleteJobMeta(const JobId& job_id) {
+  return base_store_->DeleteMeta(JOB_META, job_id);
+}
+
+Result<bool> DefaultGraphMetaStore::UpdateJobMeta(
+    const JobId& job_id, const UpdateJobMetaRequest& update_request) {
+  return base_store_->UpdateMeta(
+      JOB_META, job_id, [&update_request](const std::string& old_meta) {
+        nlohmann::json json;
+        try {
+          json = nlohmann::json::parse(old_meta);
+        } catch (const std::exception& e) {
+          LOG(ERROR) << "Fail to parse old job meta:" << e.what();
+          return Result<std::string>(
+              Status(StatusCode::InternalError, "Fail to parse old job meta"));
+        }
+        auto job_meta = JobMeta::FromJson(json);
+        if (update_request.status.has_value()) {
+          job_meta.status = update_request.status.value();
+        }
+        if (update_request.end_time.has_value()) {
+          job_meta.end_time = update_request.end_time.value();
+        }
+        return Result<std::string>(job_meta.ToJson(false));
+      });
+}
+
+Result<bool> DefaultGraphMetaStore::LockGraphIndices(const GraphId& graph_id) {
+  // First try to get lock
+  auto get_lock_res = GetGraphIndicesLocked(graph_id);
+  if (!get_lock_res.ok()) {
+    return get_lock_res.status();
+  }
+  if (get_lock_res.value()) {
+    LOG(WARNING) << "graph " << graph_id << "'s indices is already locked";
+    return Result<bool>(false);
+  }
+  auto lock_res = base_store_->CreateMeta(INDICES_LOCK, graph_id, LOCKED);
+  if (!lock_res.ok()) {
+    // If the key already exists, update it.
+    return base_store_->UpdateMeta(
+        INDICES_LOCK, graph_id, [graph_id](const std::string& old_value) {
+          if (old_value == LOCKED) {
+            return old_value;
+          } else if (old_value == UNLOCKED) {
+            return std::string(LOCKED);
+          } else {
+            LOG(ERROR) << "Unknow value: " << old_value;
+            return old_value;
+          }
+        });
+  }
+  return Result<bool>(true);
+}
+
+Result<bool> DefaultGraphMetaStore::UnlockGraphIndices(
+    const GraphId& graph_id) {
+  // First try to get lock
+  auto get_lock_res = GetGraphIndicesLocked(graph_id);
+  if (!get_lock_res.ok()) {
+    return get_lock_res.status();
+  }
+  if (!get_lock_res.value()) {
+    LOG(WARNING) << "graph " << graph_id << "'s indices is already unlocked";
+    return Result<bool>(false);
+  }
+  return base_store_->UpdateMeta(
+      INDICES_LOCK, graph_id, [graph_id](const std::string& old_value) {
+        if (old_value == LOCKED) {
+          return std::string(UNLOCKED);
+        } else if (old_value == UNLOCKED) {
+          LOG(WARNING) << "graph " << graph_id
+                       << "'s indices is already unlocked";
+          return std::string(UNLOCKED);
+        } else {
+          LOG(ERROR) << "Unknow value: " << old_value;
+          return old_value;
+        }
+      });
+}
+
+Result<bool> DefaultGraphMetaStore::GetGraphIndicesLocked(
+    const GraphId& graph_id) {
+  auto res = base_store_->GetMeta(INDICES_LOCK, graph_id);
+  if (!res.ok()) {
+    return false;  // If the key not exists, return true.
+  }
+  return Result<bool>(res.value() == LOCKED);
+}
+
+Result<bool> DefaultGraphMetaStore::LockGraphPlugins(const GraphId& graph_id) {
+  // First try to get lock
+  auto get_lock_res = GetGraphPluginsLocked(graph_id);
+  if (!get_lock_res.ok()) {
+    return get_lock_res.status();
+  }
+  if (get_lock_res.value()) {
+    LOG(WARNING) << "graph " << graph_id << "'s plugins is already locked";
+    return Result<bool>(false);
+  }
+  auto res = base_store_->CreateMeta(PLUGINS_LOCK, graph_id, LOCKED);
+  if (!res.ok()) {
+    // If the key already exists, update it.
+    return base_store_->UpdateMeta(
+        PLUGINS_LOCK, graph_id, [graph_id](const std::string& old_value) {
+          if (old_value == LOCKED) {
+            return old_value;
+          } else if (old_value == UNLOCKED) {
+            return std::string(LOCKED);
+          } else {
+            LOG(ERROR) << "Unknow value: " << old_value;
+            return old_value;
+          }
+        });
+  }
+  return Result<bool>(true);
+}
+
+Result<bool> DefaultGraphMetaStore::UnlockGraphPlugins(
+    const GraphId& graph_id) {
+  // First try to get lock
+  auto get_lock_res = GetGraphPluginsLocked(graph_id);
+  if (!get_lock_res.ok()) {
+    return get_lock_res.status();
+  }
+  if (!get_lock_res.value()) {
+    LOG(WARNING) << "graph " << graph_id << "'s plugins is already unlocked";
+    return Result<bool>(false);
+  }
+
+  return base_store_->UpdateMeta(
+      PLUGINS_LOCK, graph_id, [graph_id](const std::string& old_value) {
+        if (old_value == LOCKED) {
+          return std::string(UNLOCKED);
+        } else if (old_value == UNLOCKED) {
+          LOG(WARNING) << "graph " << graph_id
+                       << "'s plugins is already unlocked";
+          return std::string(UNLOCKED);
+        } else {
+          LOG(ERROR) << "Unknow value: " << old_value;
+          return old_value;
+        }
+      });
+}
+
+Result<bool> DefaultGraphMetaStore::GetGraphPluginsLocked(
+    const GraphId& graph_id) {
+  auto res = base_store_->GetMeta(PLUGINS_LOCK, graph_id);
+  if (!res.ok()) {
+    return false;  // If the key not exists, return true.
+  }
+  return Result<bool>(res.value() == LOCKED);
+}
+
+Result<bool> DefaultGraphMetaStore::SetRunningGraph(const GraphId& graph_id) {
+  auto create_res =
+      base_store_->CreateMeta(RUNNING_GRAPH, RUNNING_GRAPH, graph_id);
+  if (!create_res.ok()) {
+    // If the key already exists, update it.
+    return base_store_->UpdateMeta(
+        RUNNING_GRAPH, RUNNING_GRAPH,
+        [graph_id](const std::string& old_value) { return graph_id; });
+  }
+  return Result<bool>(true);
+}
+
+Result<GraphId> DefaultGraphMetaStore::GetRunningGraph() {
+  auto res = base_store_->GetMeta(RUNNING_GRAPH, RUNNING_GRAPH);
+  if (!res.ok()) {
+    return Result<GraphId>(res.status());
+  }
+  return Result<GraphId>(res.value());
+}
+
+Result<bool> DefaultGraphMetaStore::ClearRunningGraph() {
+  return base_store_->DeleteMeta(RUNNING_GRAPH, RUNNING_GRAPH);
+}
+
+Result<bool> DefaultGraphMetaStore::clear_locks() {
+  RETURN_IF_NOT_OK(base_store_->DeleteAllMeta(INDICES_LOCK));
+  RETURN_IF_NOT_OK(base_store_->DeleteAllMeta(PLUGINS_LOCK));
+  return Result<bool>(true);
+}
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/storages/metadata/default_graph_meta_store.h b/flex/storages/metadata/default_graph_meta_store.h
new file mode 100644
index 000000000000..94cf5f32f679
--- /dev/null
+++ b/flex/storages/metadata/default_graph_meta_store.h
@@ -0,0 +1,113 @@
+/** 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 FLEX_STORAGES_METADATA_DEFAULT_GRAPH_META_STORE_H_
+#define FLEX_STORAGES_METADATA_DEFAULT_GRAPH_META_STORE_H_
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "flex/storages/metadata/graph_meta_store.h"
+#include "flex/storages/metadata/i_meta_store.h"
+#include "flex/utils/property/types.h"
+#include "flex/utils/result.h"
+#include "flex/utils/service_utils.h"
+
+namespace gs {
+/*
+ *The default implementation of IGraphMetaStore.
+ *Which hold a base meta store to store the metadata.
+ *The base meta store can be based on sqlite/file system or other storage.
+ */
+class DefaultGraphMetaStore : public IGraphMetaStore {
+ public:
+  static constexpr const char* GRAPH_META = "GRAPH_META";
+  static constexpr const char* PLUGIN_META = "PLUGIN_META";
+  static constexpr const char* JOB_META = "JOB_META";
+  static constexpr const char* RUNNING_GRAPH = "RUNNING_GRAPH";
+  static constexpr const char* INDICES_LOCK = "INDICES_LOCK";
+  static constexpr const char* PLUGINS_LOCK = "PLUGINS_LOCK";
+  static constexpr const char* LOCKED = "LOCKED";
+  static constexpr const char* UNLOCKED = "UNLOCKED";
+
+  DefaultGraphMetaStore(std::unique_ptr<IMetaStore> base_store);
+  ~DefaultGraphMetaStore();
+
+  Result<bool> Open() override;
+  Result<bool> Close() override;
+
+  /* Graph Meta related.
+   */
+  Result<GraphId> CreateGraphMeta(
+      const CreateGraphMetaRequest& request) override;
+  Result<GraphMeta> GetGraphMeta(const GraphId& graph_id) override;
+  Result<std::vector<GraphMeta>> GetAllGraphMeta() override;
+  // Will also delete the plugin meta related to the graph.
+  Result<bool> DeleteGraphMeta(const GraphId& graph_id) override;
+  Result<bool> UpdateGraphMeta(
+      const GraphId& graph_id,
+      const UpdateGraphMetaRequest& update_request) override;
+
+  /* Plugin Meta related.
+   */
+  Result<PluginId> CreatePluginMeta(
+      const CreatePluginMetaRequest& request) override;
+  Result<PluginMeta> GetPluginMeta(const GraphId& graph_id,
+                                   const PluginId& plugin_id) override;
+  Result<std::vector<PluginMeta>> GetAllPluginMeta(
+      const GraphId& graph_id) override;
+  Result<bool> DeletePluginMeta(const GraphId& graph_id,
+                                const PluginId& plugin_id) override;
+  Result<bool> DeletePluginMetaByGraphId(const GraphId& graph_id) override;
+  Result<bool> UpdatePluginMeta(
+      const GraphId& graph_id, const PluginId& plugin_id,
+      const UpdatePluginMetaRequest& update_request) override;
+
+  /*
+  Job related MetaData.
+  */
+  Result<JobId> CreateJobMeta(const CreateJobMetaRequest& request) override;
+  Result<JobMeta> GetJobMeta(const JobId& job_id) override;
+  Result<std::vector<JobMeta>> GetAllJobMeta() override;
+  Result<bool> DeleteJobMeta(const JobId& job_id) override;
+  Result<bool> UpdateJobMeta(
+      const JobId& job_id, const UpdateJobMetaRequest& update_request) override;
+
+  /*
+  Use a field to represent the status of the graph.
+  */
+  Result<bool> LockGraphIndices(const GraphId& graph_id) override;
+  Result<bool> UnlockGraphIndices(const GraphId& graph_id) override;
+  Result<bool> GetGraphIndicesLocked(const GraphId& graph_id) override;
+
+  // Lock the plugin directory to avoid concurrent access.
+  Result<bool> LockGraphPlugins(const GraphId& graph_id) override;
+  Result<bool> UnlockGraphPlugins(const GraphId& graph_id) override;
+  Result<bool> GetGraphPluginsLocked(const GraphId& graph_id) override;
+
+  Result<bool> SetRunningGraph(const GraphId& graph_id) override;
+  Result<GraphId> GetRunningGraph() override;
+  Result<bool> ClearRunningGraph() override;
+
+ private:
+  Result<bool> clear_locks();
+
+  std::unique_ptr<IMetaStore> base_store_;
+};
+}  // namespace gs
+
+#endif  // FLEX_STORAGES_METADATA_DEFAULT_GRAPH_META_STORE_H_
\ No newline at end of file
diff --git a/flex/storages/metadata/graph_meta_store.cc b/flex/storages/metadata/graph_meta_store.cc
new file mode 100644
index 000000000000..00f3303b0107
--- /dev/null
+++ b/flex/storages/metadata/graph_meta_store.cc
@@ -0,0 +1,694 @@
+/** 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 <filesystem>
+#include <fstream>
+#include <iostream>
+#include <string>
+
+#include "flex/storages/metadata/graph_meta_store.h"
+#include "flex/storages/rt_mutable_graph/schema.h"
+#include "flex/utils/yaml_utils.h"
+
+namespace gs {
+
+std::string read_file_to_string(const std::string& file_path) {
+  if (std::filesystem::exists(file_path)) {
+    std::ifstream fin(file_path);
+    if (fin.is_open()) {
+      std::string line;
+      std::string res;
+      while (std::getline(fin, line)) {
+        res += line + "\n";
+      }
+      fin.close();
+      return res;
+    } else {
+      LOG(ERROR) << "Fail to open file: " << file_path;
+      return "";
+    }
+  } else {
+    LOG(ERROR) << "File not exists: " << file_path;
+    return "";
+  }
+}
+
+UpdateGraphMetaRequest::UpdateGraphMetaRequest(
+    int64_t data_update_time, const std::string& data_import_config)
+    : data_update_time(data_update_time),
+      data_import_config(data_import_config) {}
+
+std::string Parameter::ToJson() const {
+  nlohmann::json json;
+  json["name"] = name;
+  json["type"] = config_parsing::PropertyTypeToString(type);
+  return json.dump();
+}
+
+std::string GraphMeta::ToJson() const {
+  nlohmann::json json;
+  json["id"] = id;
+  json["name"] = name;
+  json["is_builtin"] = is_builtin;
+  json["description"] = description;
+  json["creation_time"] = creation_time;
+  json["data_update_time"] = data_update_time;
+  if (!data_import_config.empty()) {
+    json["data_import_config"] = nlohmann::json::parse(data_import_config);
+  }
+  json["schema"] = nlohmann::json::parse(schema);
+  return json.dump();
+}
+
+GraphMeta GraphMeta::FromJson(const std::string& json_str) {
+  auto j = nlohmann::json::parse(json_str);
+  return GraphMeta::FromJson(j);
+}
+
+GraphMeta GraphMeta::FromJson(const nlohmann::json& json) {
+  GraphMeta meta;
+  if (json.contains("id")) {
+    if (json["id"].is_number()) {
+      meta.id = json["id"].get<int64_t>();
+    } else {
+      meta.id = json["id"].get<GraphId>();
+    }
+  }
+  meta.name = json["name"].get<std::string>();
+  meta.is_builtin = json["is_builtin"].get<bool>();
+  meta.description = json["description"].get<std::string>();
+  meta.creation_time = json["creation_time"].get<int64_t>();
+  meta.schema = json["schema"].dump();
+
+  if (json.contains("data_update_time")) {
+    meta.data_update_time = json["data_update_time"].get<int64_t>();
+  }
+  if (json.contains("data_import_config")) {
+    meta.data_import_config = json["data_import_config"].dump();
+  }
+  return meta;
+}
+
+PluginMeta PluginMeta::FromJson(const std::string& json_str) {
+  auto j = nlohmann::json::parse(json_str);
+  return PluginMeta::FromJson(j);
+}
+
+PluginMeta PluginMeta::FromJson(const nlohmann::json& json) {
+  PluginMeta meta;
+  if (json.contains("id")) {
+    if (json["id"].is_number()) {
+      meta.id = json["id"].get<int64_t>();
+    } else {
+      meta.id = json["id"].get<PluginId>();
+    }
+  }
+  if (json.contains("name")) {
+    meta.name = json["name"].get<std::string>();
+  }
+  if (json.contains("graph_id")) {
+    meta.graph_id = json["graph_id"].get<GraphId>();
+  }
+  if (json.contains("description")) {
+    meta.description = json["description"].get<std::string>();
+  }
+  if (json.contains("params")) {
+    meta.setParamsFromJsonString(json["params"].dump());
+  }
+  if (json.contains("returns")) {
+    meta.setReturnsFromJsonString(json["returns"].dump());
+  }
+  if (json.contains("library")) {
+    meta.library = json["library"].get<std::string>();
+  }
+  if (json.contains("option")) {
+    meta.setOptionFromJsonString(json["option"].dump());
+  }
+  if (json.contains("creation_time")) {
+    meta.creation_time = json["creation_time"].get<int64_t>();
+  }
+  if (json.contains("update_time")) {
+    meta.update_time = json["update_time"].get<int64_t>();
+  }
+  if (json.contains("enable")) {
+    meta.enable = json["enable"].get<bool>();
+  }
+  if (json.contains("runnable")) {
+    meta.runnable = json["runnable"].get<bool>();
+  }
+  return meta;
+}
+
+std::string PluginMeta::ToJson() const {
+  nlohmann::json json;
+  json["id"] = id;
+  json["name"] = name;
+  json["graph_id"] = graph_id;
+  json["description"] = description;
+  json["params"] = nlohmann::json::array();
+  for (auto& param : params) {
+    nlohmann::json p;
+    p["name"] = param.name;
+    p["type"] = config_parsing::PropertyTypeToString(param.type);
+    json["params"].push_back(p);
+  }
+  json["returns"] = nlohmann::json::array();
+  for (auto& ret : returns) {
+    nlohmann::json r;
+    r["name"] = ret.name;
+    r["type"] = config_parsing::PropertyTypeToString(ret.type);
+    json["returns"].push_back(r);
+  }
+  for (auto& opt : option) {
+    json["option"][opt.first] = opt.second;
+  }
+  json["creation_time"] = creation_time;
+  json["update_time"] = update_time;
+  json["enable"] = enable;
+  json["runnable"] = runnable;
+  return json.dump();
+}
+
+void PluginMeta::setParamsFromJsonString(const std::string& json_str) {
+  if (json_str.empty() || json_str == "[]" || json_str == "{}" ||
+      json_str == "nu") {
+    return;
+  }
+  auto j = nlohmann::json::parse(json_str);
+  if (j.is_array()) {
+    for (auto& param : j) {
+      Parameter p;
+      p.name = param["name"].get<std::string>();
+      p.type = config_parsing::StringToPropertyType(
+          param["type"].get<std::string>());
+      params.push_back(p);
+    }
+  } else {
+    LOG(ERROR) << "Invalid params string: " << json_str;
+  }
+}
+
+void PluginMeta::setReturnsFromJsonString(const std::string& json_str) {
+  auto j = nlohmann::json::parse(json_str);
+  for (auto& ret : j) {
+    Parameter p;
+    p.name = ret["name"].get<std::string>();
+    p.type =
+        config_parsing::StringToPropertyType(ret["type"].get<std::string>());
+    returns.push_back(p);
+  }
+}
+
+void PluginMeta::setOptionFromJsonString(const std::string& json_str) {
+  auto j = nlohmann::json::parse(json_str);
+  for (auto& opt : j.items()) {
+    option[opt.key()] = opt.value().get<std::string>();
+  }
+}
+
+std::string JobMeta::ToJson(bool print_log) const {
+  nlohmann::json json;
+  json["id"] = id;
+  json["graph_id"] = graph_id;
+  json["status"] = std::to_string(status);
+  json["start_time"] = start_time;
+  json["process_id"] = process_id;
+  json["end_time"] = end_time;
+  if (print_log) {
+    json["log"] = read_file_to_string(log_path);
+  } else {
+    json["log_path"] = log_path;
+  }
+  json["type"] = type;
+  return json.dump();
+}
+
+JobMeta JobMeta::FromJson(const std::string& json_str) {
+  auto j = nlohmann::json::parse(json_str);
+  return JobMeta::FromJson(j);
+}
+
+JobMeta JobMeta::FromJson(const nlohmann::json& json) {
+  JobMeta meta;
+  if (json.contains("id")) {
+    if (json["id"].is_number()) {
+      meta.id = json["id"].get<int64_t>();
+    } else {
+      meta.id = json["id"].get<JobId>();
+    }
+  }
+  if (json.contains("graph_id")) {
+    if (json["graph_id"].is_number()) {
+      meta.graph_id = json["graph_id"].get<int64_t>();
+    } else {
+      meta.graph_id = json["graph_id"].get<GraphId>();
+    }
+  }
+  if (json.contains("process_id")) {
+    meta.process_id = json["process_id"].get<int32_t>();
+  }
+  if (json.contains("start_time")) {
+    meta.start_time = json["start_time"].get<int64_t>();
+  }
+  if (json.contains("end_time")) {
+    meta.end_time = json["end_time"].get<int64_t>();
+  }
+  if (json.contains("status")) {
+    meta.status = parseFromString(json["status"].get<std::string>());
+  }
+  if (json.contains("log_path")) {
+    meta.log_path = json["log_path"].get<std::string>();
+    VLOG(10) << "log_path: " << meta.log_path;
+  }
+  if (json.contains("type")) {
+    meta.type = json["type"].get<std::string>();
+  }
+  return meta;
+}
+
+bool CreateGraphMetaRequest::GetIsBuiltin() const {
+  if (is_builtin.has_value()) {
+    return is_builtin.value();
+  }
+  return false;
+}
+
+CreateGraphMetaRequest CreateGraphMetaRequest::FromJson(
+    const std::string& json_str) {
+  CreateGraphMetaRequest request;
+  nlohmann::json json;
+  try {
+    json = nlohmann::json::parse(json_str);
+  } catch (const std::exception& e) {
+    LOG(ERROR) << "CreateGraphMetaRequest::FromJson error: " << e.what();
+    return request;
+  }
+  if (json.contains("name")) {
+    request.name = json["name"].get<std::string>();
+  }
+  if (json.contains("is_builtin")) {
+    request.is_builtin = json["is_builtin"].get<bool>();
+  }
+  if (json.contains("description")) {
+    request.description = json["description"].get<std::string>();
+  }
+  if (json.contains("schema")) {
+    request.schema = json["schema"].dump();
+  }
+  if (json.contains("data_update_time")) {
+    request.data_update_time = json["data_update_time"].get<int64_t>();
+  }
+  if (json.contains("creation_time")) {
+    request.creation_time = json["creation_time"].get<int64_t>();
+  }
+  return request;
+}
+
+std::string CreateGraphMetaRequest::ToString() const {
+  nlohmann::json json;
+  json["name"] = name;
+  if (is_builtin.has_value()) {
+    json["is_builtin"] = is_builtin.value();
+  } else {
+    json["is_builtin"] = false;
+  }
+  json["description"] = description;
+  json["schema"] = nlohmann::json::parse(schema);
+  if (data_update_time.has_value()) {
+    json["data_update_time"] = data_update_time.value();
+  }
+  json["creation_time"] = creation_time;
+  return json.dump();
+}
+
+CreatePluginMetaRequest::CreatePluginMetaRequest() : enable(true) {}
+
+std::string CreatePluginMetaRequest::paramsString() const {
+  nlohmann::json json = nlohmann::json::array();
+  for (auto& param : params) {
+    nlohmann::json param_json;
+    param_json["name"] = param.name;
+    param_json["type"] = config_parsing::PropertyTypeToString(param.type);
+    json.push_back(param_json);
+  }
+  return json.dump();
+}
+
+std::string CreatePluginMetaRequest::returnsString() const {
+  nlohmann::json json;
+  for (auto& ret : returns) {
+    nlohmann::json ret_json;
+    ret_json["name"] = ret.name;
+    ret_json["type"] = config_parsing::PropertyTypeToString(ret.type);
+    json.push_back(ret_json);
+  }
+  return json.dump();
+}
+
+std::string CreatePluginMetaRequest::optionString() const {
+  nlohmann::json json;
+  for (auto& opt : option) {
+    json[opt.first] = opt.second;
+  }
+  return json.dump();
+}
+
+std::string CreatePluginMetaRequest::ToString() const {
+  nlohmann::json json;
+  if (id.has_value()) {
+    json["id"] = id.value();
+  }
+  json["graph_id"] = graph_id;
+  json["name"] = name;
+  json["creation_time"] = creation_time;
+  json["description"] = description;
+  json["params"] = nlohmann::json::array();
+  for (auto& param : params) {
+    nlohmann::json param_json;
+    param_json["name"] = param.name;
+    param_json["type"] = config_parsing::PropertyTypeToString(param.type);
+    json["params"].push_back(param_json);
+  }
+  json["returns"] = nlohmann::json::array();
+  for (auto& ret : returns) {
+    nlohmann::json ret_json;
+    ret_json["name"] = ret.name;
+    ret_json["type"] = config_parsing::PropertyTypeToString(ret.type);
+    json["returns"].push_back(ret_json);
+  }
+
+  json["library"] = library;
+  json["option"] = nlohmann::json::object();
+  for (auto& opt : option) {
+    json["option"][opt.first] = opt.second;
+  }
+  json["enable"] = enable;
+  return json.dump();
+}
+
+CreatePluginMetaRequest CreatePluginMetaRequest::FromJson(
+    const std::string& json) {
+  auto j = nlohmann::json::parse(json);
+  return CreatePluginMetaRequest::FromJson(j);
+}
+
+CreatePluginMetaRequest CreatePluginMetaRequest::FromJson(
+    const nlohmann::json& j) {
+  // TODO: make sure this is correct
+  CreatePluginMetaRequest request;
+  if (j.contains("id")) {
+    if (j["id"].is_number()) {
+      request.id = std::to_string(j["id"].get<int64_t>());
+    } else {
+      request.id = j["id"].get<PluginId>();
+    }
+  }
+  if (j.contains("name")) {
+    request.name = j["name"].get<std::string>();
+  }
+  if (j.contains("graph_id")) {
+    if (j["id"].is_number()) {
+      request.graph_id = j["graph_id"].get<int64_t>();
+    } else {
+      request.graph_id = j["graph_id"].get<PluginId>();
+    }
+  }
+  if (j.contains("creation_time")) {
+    request.creation_time = j["creation_time"].get<int64_t>();
+  }
+  if (j.contains("description")) {
+    request.description = j["description"].get<std::string>();
+  }
+  if (j.contains("params")) {
+    for (auto& param : j["params"]) {
+      Parameter p;
+      p.name = param["name"].get<std::string>();
+      p.type = config_parsing::StringToPropertyType(
+          param["type"].get<std::string>());
+      request.params.push_back(p);
+    }
+  }
+  if (j.contains("returns")) {
+    for (auto& ret : j["returns"]) {
+      Parameter p;
+      p.name = ret["name"].get<std::string>();
+      p.type =
+          config_parsing::StringToPropertyType(ret["type"].get<std::string>());
+      request.returns.push_back(p);
+    }
+  }
+  if (j.contains("library")) {
+    request.library = j["library"].get<std::string>();
+  }
+  if (j.contains("option")) {
+    for (auto& opt : j["option"].items()) {
+      request.option[opt.key()] = opt.value().get<std::string>();
+    }
+  }
+  if (j.contains("enable")) {
+    request.enable = j["enable"].get<bool>();
+  }
+  return request;
+}
+
+UpdatePluginMetaRequest::UpdatePluginMetaRequest() : enable(true) {}
+
+UpdatePluginMetaRequest UpdatePluginMetaRequest::FromJson(
+    const std::string& json) {
+  UpdatePluginMetaRequest request;
+  try {
+    auto j = nlohmann::json::parse(json);
+    if (j.contains("name")) {
+      if (j["name"].is_number()) {
+        request.name = std::to_string(j["name"].get<int64_t>());
+      } else {
+        request.name = j["name"].get<std::string>();
+      }
+    }
+    if (j.contains("description")) {
+      request.description = j["description"].get<std::string>();
+    }
+    if (j.contains("update_time")) {
+      request.update_time = j["update_time"].get<int64_t>();
+    } else {
+      request.update_time = GetCurrentTimeStamp();
+    }
+    if (j.contains("params")) {
+      request.params.emplace();
+      for (auto& param : j["params"]) {
+        Parameter p;
+        p.name = param["name"].get<std::string>();
+        p.type = config_parsing::StringToPropertyType(
+            param["type"].get<std::string>());
+
+        request.params->emplace_back(p);
+      }
+    }
+    if (j.contains("returns")) {
+      request.returns.emplace();
+      for (auto& ret : j["returns"]) {
+        Parameter p;
+        p.name = ret["name"].get<std::string>();
+        p.type = config_parsing::StringToPropertyType(
+            ret["type"].get<std::string>());
+        request.returns->emplace_back(p);
+      }
+    }
+    if (j.contains("library")) {
+      request.library = j["library"].get<std::string>();
+    }
+    if (j.contains("option")) {
+      request.option.emplace();
+      for (auto& opt : j["option"].items()) {
+        request.option->insert({opt.key(), opt.value()});
+      }
+    }
+    if (j.contains("enable")) {
+      request.enable = j["enable"].get<bool>();
+    }
+  } catch (const std::exception& e) {
+    LOG(ERROR) << "UpdatePluginMetaRequest::FromJson error: " << e.what();
+  }
+  return request;
+}
+
+std::string UpdatePluginMetaRequest::paramsString() const {
+  nlohmann::json json;
+  if (params.has_value()) {
+    for (auto& param : params.value()) {
+      nlohmann::json param_json;
+      param_json["name"] = param.name;
+      param_json["type"] = config_parsing::PropertyTypeToString(param.type);
+      json.push_back(param_json);
+    }
+  }
+  return json.dump();
+}
+
+std::string UpdatePluginMetaRequest::returnsString() const {
+  nlohmann::json json;
+  if (returns.has_value()) {
+    for (auto& ret : returns.value()) {
+      nlohmann::json ret_json;
+      ret_json["name"] = ret.name;
+      ret_json["type"] = config_parsing::PropertyTypeToString(ret.type);
+      json.push_back(ret_json);
+    }
+  }
+  return json.dump();
+}
+
+std::string UpdatePluginMetaRequest::optionString() const {
+  nlohmann::json json;
+  if (option.has_value()) {
+    for (auto& opt : option.value()) {
+      json[opt.first] = opt.second;
+    }
+  }
+  return json.dump();
+}
+
+std::string UpdatePluginMetaRequest::ToString() const {
+  nlohmann::json json;
+  if (name.has_value()) {
+    json["name"] = name.value();
+  }
+  if (graph_id.has_value()) {
+    json["graph_id"] = graph_id.value();
+  }
+  if (description.has_value()) {
+    json["description"] = description.value();
+  }
+  if (update_time.has_value()) {
+    json["update_time"] = update_time.value();
+  }
+  // create array of json objects
+  json["params"] = nlohmann::json::array();
+  if (params.has_value()) {
+    for (auto& param : params.value()) {
+      nlohmann::json param_json;
+      param_json["name"] = param.name;
+      param_json["type"] = config_parsing::PropertyTypeToString(param.type);
+      json["params"].push_back(param_json);
+    }
+  }
+
+  json["returns"] = nlohmann::json::array();
+  if (returns.has_value()) {
+    for (auto& ret : returns.value()) {
+      nlohmann::json ret_json;
+      ret_json["name"] = ret.name;
+      ret_json["type"] = config_parsing::PropertyTypeToString(ret.type);
+      json["returns"].push_back(ret_json);
+    }
+  }
+  if (library.has_value()) {
+    json["library"] = library.value();
+  }
+  json["option"] = nlohmann::json::object();
+  if (option.has_value()) {
+    for (auto& opt : option.value()) {
+      json["option"][opt.first] = opt.second;
+    }
+  }
+  if (enable.has_value()) {
+    json["enable"] = enable.value();
+  }
+  auto dumped = json.dump();
+  VLOG(10) << "dump: " << dumped;
+  return dumped;
+}
+
+CreateJobMetaRequest CreateJobMetaRequest::NewRunning(
+    const GraphId& graph_id, int32_t process_id, const std::string& log_path,
+    const std::string& type) {
+  CreateJobMetaRequest request;
+  request.graph_id = graph_id;
+  request.process_id = process_id;
+  request.start_time = GetCurrentTimeStamp();
+  request.status = JobStatus::kRunning;
+  request.log_path = log_path;
+  request.type = type;
+  return request;
+}
+
+std::string CreateJobMetaRequest::ToString() const {
+  nlohmann::json json;
+  json["graph_id"] = graph_id;
+  json["process_id"] = process_id;
+  json["start_time"] = start_time;
+  json["status"] = std::to_string(status);
+  json["log_path"] = log_path;
+  json["type"] = type;
+  return json.dump();
+}
+
+UpdateJobMetaRequest UpdateJobMetaRequest::NewCancel() {
+  UpdateJobMetaRequest request;
+  request.status = JobStatus::kCancelled;
+  request.end_time = GetCurrentTimeStamp();
+  return request;
+}
+
+UpdateJobMetaRequest UpdateJobMetaRequest::NewFinished(int rc) {
+  UpdateJobMetaRequest request;
+  if (rc == 0) {
+    request.status = JobStatus::kSuccess;
+  } else {
+    request.status = JobStatus::kFailed;
+  }
+  request.end_time = GetCurrentTimeStamp();
+  return request;
+}
+
+JobStatus parseFromString(const std::string& status_string) {
+  if (status_string == "RUNNING") {
+    return JobStatus::kRunning;
+  } else if (status_string == "SUCCESS") {
+    return JobStatus::kSuccess;
+  } else if (status_string == "FAILED") {
+    return JobStatus::kFailed;
+  } else if (status_string == "CANCELLED") {
+    return JobStatus::kCancelled;
+  } else {
+    LOG(ERROR) << "Unknown job status: " << status_string;
+    return JobStatus::kUnknown;
+  }
+}
+
+}  // namespace gs
+
+namespace std {
+std::string to_string(const gs::JobStatus& status) {
+  switch (status) {
+  case gs::JobStatus::kRunning:
+    return "RUNNING";
+  case gs::JobStatus::kSuccess:
+    return "SUCCESS";
+  case gs::JobStatus::kFailed:
+    return "FAILED";
+  case gs::JobStatus::kCancelled:
+    return "CANCELLED";
+  case gs::JobStatus::kUnknown:
+    return "UNKNOWN";
+  }
+  return "UNKNOWN";
+}
+
+std::ostream& operator<<(std::ostream& os, const gs::JobStatus& status) {
+  os << to_string(status);
+  return os;
+}
+}  // namespace std
diff --git a/flex/storages/metadata/graph_meta_store.h b/flex/storages/metadata/graph_meta_store.h
new file mode 100644
index 000000000000..b476f26ecb51
--- /dev/null
+++ b/flex/storages/metadata/graph_meta_store.h
@@ -0,0 +1,296 @@
+/** 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 FLEX_STORAGES_METADATA_GRAPH_META_STORE_H_
+#define FLEX_STORAGES_METADATA_GRAPH_META_STORE_H_
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "flex/storages/rt_mutable_graph/schema.h"
+#include "flex/utils/property/types.h"
+#include "flex/utils/result.h"
+#include "flex/utils/service_utils.h"
+
+#include <yaml-cpp/yaml.h>
+#include "nlohmann/json.hpp"
+
+namespace gs {
+
+using GraphId = std::string;
+using PluginId = std::string;
+using JobId = std::string;
+
+// Describe the input and output of the plugin.
+struct Parameter {
+  std::string name;
+  PropertyType type;
+
+  std::string ToJson() const;
+};
+
+enum class JobStatus {
+  kRunning,
+  kSuccess,
+  kFailed,
+  kCancelled,
+  kUnknown,
+};
+
+JobStatus parseFromString(const std::string& status_string);
+
+////////////////// MetaData ///////////////////////
+struct GraphMeta {
+  GraphId id;
+  std::string name;
+  bool is_builtin;
+  std::string description;
+  uint64_t creation_time;
+  uint64_t data_update_time;
+  std::string data_import_config;
+  std::string schema;
+
+  std::string ToJson() const;
+  static GraphMeta FromJson(const std::string& json_str);
+  static GraphMeta FromJson(const nlohmann::json& json);
+};
+
+struct PluginMeta {
+  PluginId id;
+  std::string name;
+  GraphId graph_id;
+  std::string description;
+  std::vector<Parameter> params;
+  std::vector<Parameter> returns;
+  std::string library;
+  std::unordered_map<std::string, std::string>
+      option;  // other optional configuration
+
+  bool enable;    // whether the plugin is enabled.
+  bool runnable;  // whether the plugin is runnable.
+  uint64_t creation_time;
+  uint64_t update_time;
+
+  void setParamsFromJsonString(const std::string& json_str);
+
+  void setReturnsFromJsonString(const std::string& json_str);
+
+  void setOptionFromJsonString(const std::string& json_str);
+
+  std::string ToJson() const;
+
+  static PluginMeta FromJson(const std::string& json_str);
+  static PluginMeta FromJson(const nlohmann::json& json);
+};
+
+struct JobMeta {
+  JobId id;
+  GraphId graph_id;
+  int32_t process_id;
+  uint64_t start_time;
+  uint64_t end_time;
+  JobStatus status;
+  std::string log_path;  // The path to log file.
+  std::string type;
+
+  /*
+   * Convert the JobMeta to a json string.
+   * @param print_log: whether to print the real log or just the path.
+   * @return: the json string.
+   */
+  std::string ToJson(bool print_log = true) const;
+  static JobMeta FromJson(const std::string& json_str);
+  static JobMeta FromJson(const nlohmann::json& json_str);
+};
+
+////////////////// CreateMetaRequest ///////////////////////
+struct CreateGraphMetaRequest {
+  std::string name;
+  std::optional<bool> is_builtin;
+  std::string description;
+  std::string schema;  // all in one string.
+  std::optional<uint64_t> data_update_time;
+  int64_t creation_time;
+
+  bool GetIsBuiltin() const;
+
+  static CreateGraphMetaRequest FromJson(const std::string& json_str);
+
+  std::string ToString() const;
+};
+
+struct CreatePluginMetaRequest {
+  std::optional<PluginId> id;
+  std::string name;
+  GraphId graph_id;
+  int64_t creation_time;
+  std::string description;
+  std::vector<Parameter> params;
+  std::vector<Parameter> returns;
+  std::string library;
+  std::unordered_map<std::string, std::string> option;
+  bool enable;  // default true
+
+  CreatePluginMetaRequest();
+
+  std::string ToString() const;
+
+  std::string paramsString() const;
+
+  std::string returnsString() const;
+
+  std::string optionString() const;
+
+  static CreatePluginMetaRequest FromJson(const std::string& json_str);
+
+  static CreatePluginMetaRequest FromJson(const nlohmann::json& json_obj);
+};
+
+////////////////// UpdateMetaRequest ///////////////////////
+struct UpdateGraphMetaRequest {
+  std::optional<std::string> graph_name;
+  std::optional<std::string> description;
+  std::optional<uint64_t> data_update_time;
+  std::optional<std::string> data_import_config;
+
+  UpdateGraphMetaRequest(int64_t data_update_time,
+                         const std::string& data_import_config);
+};
+
+// Used internally, can update params, returns, library and option.
+struct UpdatePluginMetaRequest {
+  std::optional<std::string> name;
+  std::optional<GraphId> graph_id;
+  std::optional<std::string> description;
+  std::optional<int64_t> update_time;
+  std::optional<std::vector<Parameter>> params;
+  std::optional<std::vector<Parameter>> returns;
+  std::optional<std::string> library;
+  std::optional<std::unordered_map<std::string, std::string>> option;
+  std::optional<bool> enable;
+
+  UpdatePluginMetaRequest();
+
+  std::string paramsString() const;
+
+  std::string returnsString() const;
+
+  std::string optionString() const;
+
+  std::string ToString() const;
+
+  static UpdatePluginMetaRequest FromJson(const std::string& json_str);
+};
+
+struct CreateJobMetaRequest {
+  GraphId graph_id;
+  int32_t process_id;
+  uint64_t start_time;
+  JobStatus status;
+  std::string log_path;
+  std::string type;
+  std::string ToString() const;
+
+  static CreateJobMetaRequest NewRunning(const GraphId& graph_id,
+                                         int32_t process_id,
+                                         const std::string& log_path,
+                                         const std::string& type);
+};
+
+struct UpdateJobMetaRequest {
+  std::optional<JobStatus> status;
+  std::optional<uint64_t> end_time;
+
+  static UpdateJobMetaRequest NewCancel();
+  static UpdateJobMetaRequest NewFinished(int rc);
+};
+
+/*
+ * The metadata store is responsible for storing metadata of the graph, plugins
+ * and other information.
+ *
+ * MetadataStore should be thread safe inside.
+ */
+class IGraphMetaStore {
+ public:
+  virtual ~IGraphMetaStore() = default;
+
+  virtual Result<bool> Open() = 0;
+  virtual Result<bool> Close() = 0;
+
+  /* Graph Meta related.
+   */
+  virtual Result<GraphId> CreateGraphMeta(
+      const CreateGraphMetaRequest& request) = 0;
+  virtual Result<GraphMeta> GetGraphMeta(const GraphId& graph_id) = 0;
+  virtual Result<std::vector<GraphMeta>> GetAllGraphMeta() = 0;
+  // Will also delete the plugin meta related to the graph.
+  virtual Result<bool> DeleteGraphMeta(const GraphId& graph_id) = 0;
+  virtual Result<bool> UpdateGraphMeta(
+      const GraphId& graph_id,
+      const UpdateGraphMetaRequest& update_request) = 0;
+
+  /* Plugin Meta related.
+   */
+  virtual Result<PluginId> CreatePluginMeta(
+      const CreatePluginMetaRequest& request) = 0;
+  virtual Result<PluginMeta> GetPluginMeta(const GraphId& graph_id,
+                                           const PluginId& plugin_id) = 0;
+  virtual Result<std::vector<PluginMeta>> GetAllPluginMeta(
+      const GraphId& graph_id) = 0;
+  virtual Result<bool> DeletePluginMeta(const GraphId& graph_id,
+                                        const PluginId& plugin_id) = 0;
+  virtual Result<bool> DeletePluginMetaByGraphId(const GraphId& graph_id) = 0;
+  virtual Result<bool> UpdatePluginMeta(
+      const GraphId& graph_id, const PluginId& plugin_id,
+      const UpdatePluginMetaRequest& update_request) = 0;
+
+  /*
+  Job related MetaData.
+  */
+  virtual Result<JobId> CreateJobMeta(const CreateJobMetaRequest& request) = 0;
+  virtual Result<JobMeta> GetJobMeta(const JobId& job_id) = 0;
+  virtual Result<std::vector<JobMeta>> GetAllJobMeta() = 0;
+  virtual Result<bool> DeleteJobMeta(const JobId& job_id) = 0;
+  virtual Result<bool> UpdateJobMeta(
+      const JobId& job_id, const UpdateJobMetaRequest& update_request) = 0;
+
+  /*
+  Use a field to represent the status of the graph.
+  */
+  virtual Result<bool> LockGraphIndices(const GraphId& graph_id) = 0;
+  virtual Result<bool> UnlockGraphIndices(const GraphId& graph_id) = 0;
+  virtual Result<bool> GetGraphIndicesLocked(const GraphId& graph_id) = 0;
+  // Lock the plugin directory to avoid concurrent access.
+  virtual Result<bool> LockGraphPlugins(const GraphId& graph_id) = 0;
+  virtual Result<bool> UnlockGraphPlugins(const GraphId& graph_id) = 0;
+  virtual Result<bool> GetGraphPluginsLocked(const GraphId& graph_id) = 0;
+
+  virtual Result<bool> SetRunningGraph(const GraphId& graph_id) = 0;
+  virtual Result<GraphId> GetRunningGraph() = 0;
+  virtual Result<bool> ClearRunningGraph() = 0;
+};
+
+};  // namespace gs
+
+namespace std {
+std::string to_string(const gs::JobStatus& status);
+
+std::ostream& operator<<(std::ostream& os, const gs::JobStatus& status);
+}  // namespace std
+
+#endif  // FLEX_STORAGES_METADATA_GRAPH_META_STORE_H_
\ No newline at end of file
diff --git a/flex/storages/metadata/i_meta_store.h b/flex/storages/metadata/i_meta_store.h
new file mode 100644
index 000000000000..fe1cf2dbf8d6
--- /dev/null
+++ b/flex/storages/metadata/i_meta_store.h
@@ -0,0 +1,74 @@
+/** 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 FLEX_STORAGES_METADATA_I_META_STORE_H_
+#define FLEX_STORAGES_METADATA_I_META_STORE_H_
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "flex/utils/result.h"
+
+namespace gs {
+
+/**
+ * A general kv-base metadata store interface.
+ * The multi-thread safety should be guaranteed.
+ */
+class IMetaStore {
+ public:
+  using meta_kind_t = std::string;
+  using meta_key_t = std::string;
+  using meta_value_t = std::string;
+  using update_func_t =
+      std::function<Result<meta_value_t>(const meta_value_t&)>;
+  virtual ~IMetaStore() = default;
+
+  virtual Result<bool> Open() = 0;
+  virtual Result<bool> Close() = 0;
+
+  virtual Result<meta_key_t> CreateMeta(const meta_kind_t& meta_kind,
+                                        const meta_value_t& value) = 0;
+
+  virtual Result<meta_value_t> CreateMeta(const meta_kind_t& meta_kind,
+                                          const meta_key_t& key,
+                                          const meta_value_t& value) = 0;
+
+  virtual Result<meta_value_t> GetMeta(const meta_kind_t& meta_kind,
+                                       const meta_key_t& key) = 0;
+
+  virtual Result<std::vector<std::pair<meta_key_t, meta_value_t>>> GetAllMeta(
+      const meta_kind_t& meta_kind) = 0;
+
+  virtual Result<bool> DeleteMeta(const meta_kind_t& meta_kind,
+                                  const meta_key_t& key) = 0;
+
+  virtual Result<bool> DeleteAllMeta(const meta_kind_t& meta_kind) = 0;
+
+  virtual Result<bool> UpdateMeta(const meta_kind_t& meta_kind,
+                                  const meta_key_t& key,
+                                  const meta_value_t& value) = 0;
+
+  virtual Result<bool> UpdateMeta(const meta_kind_t& meta_kind,
+                                  const meta_key_t& key,
+                                  update_func_t update_func) = 0;
+};
+
+}  // namespace gs
+
+#endif  // FLEX_STORAGES_METADATA_I_META_STORE_H_
\ No newline at end of file
diff --git a/flex/storages/metadata/local_file_metadata_store.cc b/flex/storages/metadata/local_file_metadata_store.cc
new file mode 100644
index 000000000000..74f50a7815f3
--- /dev/null
+++ b/flex/storages/metadata/local_file_metadata_store.cc
@@ -0,0 +1,275 @@
+/** 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/metadata/local_file_metadata_store.h"
+
+namespace gs {
+
+LocalFileMetadataStore::LocalFileMetadataStore(const std::string& path)
+    : root_dir_(path) {
+  VLOG(10) << "Root dir: " << root_dir_;
+}
+
+LocalFileMetadataStore::~LocalFileMetadataStore() { Close(); }
+
+Result<bool> LocalFileMetadataStore::Open() {
+  // open directories.
+  RETURN_IF_NOT_OK(create_directory(root_dir_));
+  LOG(INFO) << "Successfully open metadata store";
+  return true;
+}
+
+Result<bool> LocalFileMetadataStore::Close() { return true; }
+
+Result<LocalFileMetadataStore::meta_key_t> LocalFileMetadataStore::CreateMeta(
+    const meta_kind_t& meta_kind, const meta_value_t& value) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  meta_key_t meta_key;
+  ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(meta_key, get_next_meta_key(meta_kind));
+  VLOG(10) << "got next meta key: " << meta_key;
+  if (is_key_exist(meta_kind, meta_key)) {
+    return Status(StatusCode::InternalError,
+                  "When creating meta, got an existing key");
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  auto res = dump_file(meta_file, value);
+  if (!res.ok()) {
+    return res.status();
+  }
+  return meta_key;
+}
+
+Result<LocalFileMetadataStore::meta_key_t> LocalFileMetadataStore::CreateMeta(
+    const meta_kind_t& meta_kind, const meta_key_t& meta_key,
+    const meta_value_t& value) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  if (is_key_exist(meta_kind, meta_key)) {
+    LOG(ERROR) << "Can not insert meta, key already exists: " << meta_kind
+               << ", meta_key: " << meta_key;
+    return Status(StatusCode::InternalError,
+                  "key " + meta_key + " already exits for meta: " + meta_kind);
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  auto res = dump_file(meta_file, value);
+  if (!res.ok()) {
+    return res.status();
+  }
+  return meta_key;
+}
+
+Result<LocalFileMetadataStore::meta_value_t> LocalFileMetadataStore::GetMeta(
+    const meta_key_t& meta_kind, const meta_key_t& meta_key) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  if (!is_key_exist(meta_kind, meta_key)) {
+    return Status(StatusCode::NotFound,
+                  "key " + meta_key + " not found for :" + meta_kind);
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  meta_value_t meta_value;
+  ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(meta_value, read_file(meta_file));
+  return meta_value;
+}
+
+Result<std::vector<std::pair<LocalFileMetadataStore::meta_key_t,
+                             LocalFileMetadataStore::meta_value_t>>>
+LocalFileMetadataStore::GetAllMeta(const meta_kind_t& meta_kind) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  VLOG(10) << "Getting all meta for: " << meta_kind;
+  std::vector<std::pair<meta_key_t, meta_value_t>> meta_values;
+  auto meta_dir = get_meta_kind_dir(meta_kind);
+  for (auto& p : std::filesystem::directory_iterator(meta_dir)) {
+    if (!std::filesystem::is_regular_file(p)) {
+      continue;
+    }
+    auto file_name = p.path().filename().string();
+    if (file_name.find(META_FILE_PREFIX) != std::string::npos) {
+      if (file_name.find(META_FILE_PREFIX) == std::string::npos) {
+        VLOG(10) << "Skip invalid file: " << file_name;
+        continue;
+      }
+      auto id_str = file_name.substr(strlen(META_FILE_PREFIX));
+      VLOG(10) << "Reading meta file: " << file_name;
+      auto meta_file = get_meta_file(meta_kind, id_str);
+      auto meta_value_res = read_file(meta_file);
+      if (meta_value_res.ok()) {
+        meta_values.push_back(
+            std::make_pair(id_str, meta_value_res.move_value()));
+      } else {
+        LOG(ERROR) << "Error when reading meta file: " << meta_file;
+      }
+    } else {
+      LOG(WARNING) << "Invalid file: " << file_name;
+    }
+  }
+  return meta_values;
+}
+
+Result<bool> LocalFileMetadataStore::DeleteMeta(const meta_kind_t& meta_kind,
+                                                const meta_key_t& meta_key) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  if (!is_key_exist(meta_kind, meta_key)) {
+    return Status(StatusCode::NotFound,
+                  "key " + meta_key + " not found for :" + meta_kind);
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  if (!std::filesystem::remove(meta_file)) {
+    return Status(StatusCode::IOError, "Failed to delete meta");
+  }
+  return true;
+}
+
+Result<bool> LocalFileMetadataStore::DeleteAllMeta(
+    const meta_kind_t& meta_kind) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  auto meta_dir = get_meta_kind_dir(meta_kind);
+  if (!std::filesystem::remove_all(meta_dir)) {
+    return Status(StatusCode::IOError, "Failed to delete meta");
+  }
+  VLOG(10) << "Remove all meta for " << meta_kind;
+  return true;
+}
+
+Result<bool> LocalFileMetadataStore::UpdateMeta(
+    const meta_kind_t& meta_kind, const meta_key_t& meta_key,
+    const meta_value_t& meta_value) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  if (!is_key_exist(meta_kind, meta_key)) {
+    return Status(StatusCode::NotFound,
+                  "key " + meta_key + " not found for :" + meta_kind);
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  auto res = dump_file(meta_file, meta_value);
+  if (!res.ok()) {
+    return res.status();
+  }
+  return true;
+}
+
+Result<bool> LocalFileMetadataStore::UpdateMeta(const meta_kind_t& meta_kind,
+                                                const meta_key_t& meta_key,
+                                                update_func_t update_func) {
+  std::unique_lock<std::mutex> lock(meta_mutex_);
+  if (!is_key_exist(meta_kind, meta_key)) {
+    return Status(StatusCode::NotFound,
+                  "key " + meta_key + " not found for :" + meta_kind);
+  }
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  auto meta_value_res = read_file(meta_file);
+  if (!meta_value_res.ok()) {
+    return meta_value_res.status();
+  }
+  auto new_meta_value = update_func(meta_value_res.value());
+  if (!new_meta_value.ok()) {
+    return new_meta_value.status();
+  }
+  auto res = dump_file(meta_file, new_meta_value.value());
+  if (!res.ok()) {
+    return res.status();
+  }
+  return true;
+}
+
+Result<LocalFileMetadataStore::meta_key_t>
+LocalFileMetadataStore::get_next_meta_key(
+    const LocalFileMetadataStore::meta_kind_t& meta_kind) const {
+  return std::to_string(get_max_id(meta_kind) + 1);
+}
+
+std::string LocalFileMetadataStore::get_root_meta_dir() const {
+  auto ret = root_dir_ + "/" + METADATA_DIR;
+  if (!std::filesystem::exists(ret)) {
+    std::filesystem::create_directory(ret);
+  }
+  return ret;
+}
+
+std::string LocalFileMetadataStore::get_meta_kind_dir(
+    const meta_kind_t& meta_kind) const {
+  auto ret = get_root_meta_dir() + "/" + meta_kind;
+  if (!std::filesystem::exists(ret)) {
+    std::filesystem::create_directory(ret);
+  }
+  return ret;
+}
+
+std::string LocalFileMetadataStore::get_meta_file(const meta_kind_t& meta_kind,
+                                                  const meta_key_t& key) const {
+  auto ret = get_meta_kind_dir(meta_kind) + "/" + META_FILE_PREFIX + key;
+  return ret;
+}
+
+int32_t LocalFileMetadataStore::get_max_id(const meta_kind_t& meta_kind) const {
+  // iterate all files in the directory, get the max id.
+  int max_id_ = 0;
+  auto dir = get_meta_kind_dir(meta_kind);
+  for (auto& p : std::filesystem::directory_iterator(dir)) {
+    if (std::filesystem::is_directory(p)) {
+      continue;
+    }
+    auto file_name = p.path().filename().string();
+    if (file_name.find(META_FILE_PREFIX) != std::string::npos) {
+      auto id_str = file_name.substr(strlen(META_FILE_PREFIX));
+      auto id = std::stoi(id_str);
+      if (id > max_id_) {
+        max_id_ = id;
+      }
+    }
+  }
+  return max_id_;
+}
+
+bool LocalFileMetadataStore::is_key_exist(
+    const LocalFileMetadataStore::meta_kind_t& meta_kind,
+    const LocalFileMetadataStore::meta_key_t& meta_key) const {
+  auto meta_file = get_meta_file(meta_kind, meta_key);
+  return std::filesystem::exists(meta_file);
+}
+
+Result<bool> LocalFileMetadataStore::dump_file(
+    const std::string& file_path, const std::string& content) const {
+  std::ofstream out_file(file_path);
+  if (!out_file.is_open()) {
+    return Result<bool>(gs::StatusCode::IOError, false);
+  }
+  out_file << content;
+  out_file.close();
+  return Result<bool>(true);
+}
+
+Result<LocalFileMetadataStore::meta_value_t> LocalFileMetadataStore::read_file(
+    const std::string& file_path) const {
+  std::ifstream in_file(file_path);
+  if (!in_file.is_open()) {
+    return Result<LocalFileMetadataStore::meta_value_t>(gs::StatusCode::IOError,
+                                                        "Failed to open file");
+  }
+  std::string content((std::istreambuf_iterator<char>(in_file)),
+                      std::istreambuf_iterator<char>());
+  in_file.close();
+  return Result<LocalFileMetadataStore::meta_value_t>(content);
+}
+
+Result<bool> LocalFileMetadataStore::create_directory(
+    const std::string& dir) const {
+  if (!std::filesystem::exists(dir)) {
+    if (!std::filesystem::create_directory(dir)) {
+      return Result<bool>(gs::StatusCode::IOError,
+                          "Failed to create directory");
+    }
+  }
+  return Result<bool>(true);
+}
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/storages/metadata/local_file_metadata_store.h b/flex/storages/metadata/local_file_metadata_store.h
new file mode 100644
index 000000000000..186aa68e7efd
--- /dev/null
+++ b/flex/storages/metadata/local_file_metadata_store.h
@@ -0,0 +1,134 @@
+/** 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 FLEX_STORAGES_METADATA_LOCAL_FILE_METADATA_STORE_H_
+#define FLEX_STORAGES_METADATA_LOCAL_FILE_METADATA_STORE_H_
+
+#include <fstream>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "flex/storages/metadata/i_meta_store.h"
+#include "flex/utils/service_utils.h"
+
+#include <boost/format.hpp>
+
+namespace gs {
+
+/**
+ * @brief LocalFileMetadataStore is a concrete implementation of MetadataStore,
+ * which stores metadata via local files.
+ *
+ * We store the graph meta and procedure meta in to files under workspace.
+ * ├── META_CLASS1
+ * │   ├── KEY1
+ * │   └── KEY2
+ * └── META_CLASS2
+ *     ├── KEY1
+ *     └── KEY2
+ */
+class LocalFileMetadataStore : public IMetaStore {
+ public:
+  using meta_key_t = IMetaStore::meta_key_t;
+  using meta_value_t = IMetaStore::meta_value_t;
+  using meta_kind_t = IMetaStore::meta_kind_t;
+
+  static constexpr const char* METADATA_DIR = "METADATA";
+  static constexpr const char* META_FILE_PREFIX = "META_";
+
+  LocalFileMetadataStore(const std::string& path);
+
+  ~LocalFileMetadataStore();
+
+  Result<bool> Open() override;
+
+  Result<bool> Close() override;
+
+  /*
+   * @brief Create a meta with a new key.
+   * @param meta_kind The kind of meta.
+   * @param value The value of the meta.
+   * @return The key of the meta.
+   */
+  Result<meta_key_t> CreateMeta(const meta_kind_t& meta_kind,
+                                const meta_value_t& value) override;
+
+  /*
+   * @brief Create a meta with a specific key.
+   * @param meta_kind The kind of meta.
+   * @param key The key of the meta.
+   * @param value The value of the meta.
+   * @return If the meta is created successfully.
+   */
+  Result<meta_key_t> CreateMeta(const meta_kind_t& meta_kind,
+                                const meta_key_t& key,
+                                const meta_value_t& value) override;
+
+  Result<meta_value_t> GetMeta(const meta_kind_t& meta_kind,
+                               const meta_key_t& key) override;
+
+  Result<std::vector<std::pair<meta_key_t, meta_value_t>>> GetAllMeta(
+      const meta_kind_t& meta_kind) override;
+
+  Result<bool> DeleteMeta(const meta_kind_t& meta_kind,
+                          const meta_key_t& key) override;
+
+  Result<bool> DeleteAllMeta(const meta_kind_t& meta_kind) override;
+
+  /*
+   * @brief Update the meta with a specific key, regardless of the original
+   * value.
+   * @param meta_kind The kind of meta.
+   * @param key The key of the meta.
+   * @param value The new value of the meta.
+   * @return If the meta is updated successfully.
+   */
+  Result<bool> UpdateMeta(const meta_kind_t& meta_kind, const meta_key_t& key,
+                          const meta_value_t& value) override;
+
+  /**
+   * @brief Update the meta with a specific key, based on the original value.
+   * @param meta_kind The kind of meta.
+   * @param key The key of the meta.
+   * @param update_func The function to update the meta.
+   * @return If the meta is updated successfully.
+   */
+  Result<bool> UpdateMeta(const meta_kind_t& meta_kind, const meta_key_t& key,
+                          update_func_t update_func) override;
+
+ private:
+  Result<meta_key_t> get_next_meta_key(const meta_kind_t& meta_kind) const;
+  std::string get_root_meta_dir() const;
+  std::string get_meta_kind_dir(const meta_kind_t& meta_kind) const;
+  std::string get_meta_file(const meta_kind_t& meta_kind,
+                            const meta_key_t& meta_key) const;
+  int32_t get_max_id(const meta_kind_t& meta_kind) const;
+  bool is_key_exist(const meta_kind_t& meta_kind,
+                    const meta_key_t& meta_key) const;
+
+  Result<bool> dump_file(const std::string& file_path,
+                         const std::string& content) const;
+  Result<meta_value_t> read_file(const std::string& file_path) const;
+
+  Result<bool> create_directory(const std::string& dir) const;
+
+  std::mutex meta_mutex_;
+
+  std::string root_dir_;
+};
+}  // namespace gs
+
+#endif  // FLEX_STORAGES_METADATA_LOCAL_FILE_METADATA_STORE_H_
\ No newline at end of file
diff --git a/flex/storages/metadata/metadata_store_factory.cc b/flex/storages/metadata/metadata_store_factory.cc
new file mode 100644
index 000000000000..86dc69f0433a
--- /dev/null
+++ b/flex/storages/metadata/metadata_store_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/metadata/metadata_store_factory.h"
+
+namespace gs {
+std::shared_ptr<IGraphMetaStore> MetadataStoreFactory::Create(
+    MetadataStoreType type, const std::string& path) {
+  switch (type) {
+  case MetadataStoreType::kLocalFile:
+    return std::make_shared<DefaultGraphMetaStore>(
+        std::make_unique<LocalFileMetadataStore>(path));
+  default:
+    LOG(FATAL) << "Unsupported metadata store type: " << static_cast<int>(type);
+  }
+  return nullptr;
+}
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/storages/metadata/metadata_store_factory.h b/flex/storages/metadata/metadata_store_factory.h
new file mode 100644
index 000000000000..181452c1c5f2
--- /dev/null
+++ b/flex/storages/metadata/metadata_store_factory.h
@@ -0,0 +1,40 @@
+/** 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 FLEX_STORAGES_METADATA_METADATA_STORE_FACTORY_H_
+#define FLEX_STORAGES_METADATA_METADATA_STORE_FACTORY_H_
+
+#include <string>
+#include "flex/storages/metadata/local_file_metadata_store.h"
+#include "flex/storages/metadata/default_graph_meta_store.h"
+#include "flex/storages/metadata/graph_meta_store.h"
+
+namespace gs {
+
+enum class MetadataStoreType {
+  kLocalFile,
+};
+/**
+ * @brief LoaderFactory is a factory class to create IFragmentLoader.
+ * Support Using dynamically built library as plugin.
+ */
+class MetadataStoreFactory {
+ public:
+  static std::shared_ptr<IGraphMetaStore> Create(MetadataStoreType type,
+                                                 const std::string& path);
+};
+}  // namespace gs
+
+#endif  // FLEX_STORAGES_METADATA_METADATA_STORE_FACTORY_H_
\ No newline at end of file
diff --git a/flex/storages/rt_mutable_graph/schema.cc b/flex/storages/rt_mutable_graph/schema.cc
index be9366b2f11f..13a80300b944 100644
--- a/flex/storages/rt_mutable_graph/schema.cc
+++ b/flex/storages/rt_mutable_graph/schema.cc
@@ -433,31 +433,61 @@ bool Schema::Equals(const Schema& other) const {
 
 namespace config_parsing {
 
-static PropertyType StringToPropertyType(const std::string& str) {
-  if (str == "int32" || str == DT_SIGNED_INT32) {
+std::string PropertyTypeToString(PropertyType type) {
+  if (type == PropertyType::kInt32) {
+    return DT_SIGNED_INT32;
+  } else if (type == PropertyType::kUInt32) {
+    return DT_UNSIGNED_INT32;
+  } else if (type == PropertyType::kBool) {
+    return DT_BOOL;
+  } else if (type == PropertyType::kDate) {
+    return DT_DATE;
+  } else if (type == PropertyType::kDay) {
+    return DT_DAY;
+  } else if (type == PropertyType::kString) {
+    return DT_STRING;
+  } else if (type == PropertyType::kStringMap) {
+    return DT_STRINGMAP;
+  } else if (type == PropertyType::kEmpty) {
+    return "Empty";
+  } else if (type == PropertyType::kInt64) {
+    return DT_SIGNED_INT64;
+  } else if (type == PropertyType::kUInt64) {
+    return DT_UNSIGNED_INT64;
+  } else if (type == PropertyType::kFloat) {
+    return DT_FLOAT;
+  } else if (type == PropertyType::kDouble) {
+    return DT_DOUBLE;
+  } else {
+    return "Empty";
+  }
+}
+
+PropertyType StringToPropertyType(const std::string& str) {
+  if (str == "int32" || str == "INT" || str == DT_SIGNED_INT32) {
     return PropertyType::kInt32;
   } else if (str == "uint32" || str == DT_UNSIGNED_INT32) {
     return PropertyType::kUInt32;
-  } else if (str == "bool" || str == DT_BOOL) {
+  } else if (str == "bool" || str == "BOOL" || str == DT_BOOL) {
     return PropertyType::kBool;
   } else if (str == "Date" || str == DT_DATE) {
     return PropertyType::kDate;
   } else if (str == "Day" || str == DT_DAY) {
     return PropertyType::kDay;
-  } else if (str == "String" || str == DT_STRING) {
+  } else if (str == "String" || str == "STRING" || str == DT_STRING) {
     // DT_STRING is a alias for VARCHAR(STRING_DEFAULT_MAX_LENGTH);
     return PropertyType::Varchar(PropertyType::STRING_DEFAULT_MAX_LENGTH);
   } else if (str == DT_STRINGMAP) {
     return PropertyType::kStringMap;
   } else if (str == "Empty") {
     return PropertyType::kEmpty;
-  } else if (str == "int64" || str == DT_SIGNED_INT64) {
+  } else if (str == "int64" || str == "LONG" || str == DT_SIGNED_INT64) {
     return PropertyType::kInt64;
   } else if (str == "uint64" || str == DT_UNSIGNED_INT64) {
     return PropertyType::kUInt64;
-  } else if (str == "float" || str == DT_FLOAT) {
+  } else if (str == "float" || str == "FLOAT" || str == DT_FLOAT) {
     return PropertyType::kFloat;
-  } else if (str == "double" || str == DT_DOUBLE) {
+  } else if (str == "double" || str == "DOUBLE" || str == DT_DOUBLE) {
     return PropertyType::kDouble;
   } else {
     return PropertyType::kEmpty;
diff --git a/flex/storages/rt_mutable_graph/schema.h b/flex/storages/rt_mutable_graph/schema.h
index edfb024e0ddd..b80e483aab9a 100644
--- a/flex/storages/rt_mutable_graph/schema.h
+++ b/flex/storages/rt_mutable_graph/schema.h
@@ -25,6 +25,11 @@
 
 namespace gs {
 
+namespace config_parsing {
+std::string PropertyTypeToString(PropertyType type);
+PropertyType StringToPropertyType(const std::string& str);
+}  // namespace config_parsing
+
 class Schema {
  public:
   // How many built-in plugins are there.
diff --git a/flex/utils/result.h b/flex/utils/result.h
index fb01f903369e..6d553b4c3b23 100644
--- a/flex/utils/result.h
+++ b/flex/utils/result.h
@@ -40,6 +40,12 @@ enum class StatusCode {
   IOError = 12,
   NotFound = 13,
   QueryFailed = 14,
+  ReopenError = 15,
+  ErrorOpenMeta = 16,
+  SQlExecutionError = 17,
+  SqlBindingError = 18,
+  Unimplemented = 19,
+  AlreadyLocked = 20,
 };
 
 class Status {
@@ -110,6 +116,12 @@ struct is_gs_result_type : std::false_type {};
 template <typename T>
 struct is_gs_result_type<Result<T>> : std::true_type {};
 
+template <typename T>
+struct is_gs_status_type : std::false_type {};
+
+template <>
+struct is_gs_status_type<Status> : std::true_type {};
+
 // define a macro, which checks the return status of a function, if ok, continue
 // to execute, otherwise, return the status.
 // the macro accept the calling code of a function, and the function name.
@@ -127,7 +139,7 @@ struct is_gs_result_type<Result<T>> : std::true_type {};
 // function, the function name, and the variable name.
 // reference:
 // https://github.com/boostorg/leaf/blob/develop/include/boost/leaf/error.hpp
-#define ASSIGN_AND_RETURN_IF_NOT_OK(var, expr)                                 \
+#define ASSIGN_AND_RETURN_IF_RESULT_NOT_OK(var, expr)                          \
   auto&& FLEX_TMP_VAR = expr;                                                  \
   static_assert(::gs::is_gs_result_type<                                       \
                     typename std::decay<decltype(FLEX_TMP_VAR)>::type>::value, \
@@ -137,6 +149,16 @@ struct is_gs_result_type<Result<T>> : std::true_type {};
   }                                                                            \
   var = std::forward<decltype(FLEX_TMP_VAR)>(FLEX_TMP_VAR).move_value()
 
+#define ASSIGN_AND_RETURN_IF_STATUS_NOT_OK(var, expr)                          \
+  auto&& FLEX_TMP_VAR = expr;                                                  \
+  static_assert(::gs::is_gs_status_type<                                       \
+                    typename std::decay<decltype(FLEX_TMP_VAR)>::type>::value, \
+                "The expression must return a Status type");                   \
+  if (!FLEX_TMP_VAR.ok()) {                                                    \
+    return FLEX_TMP_VAR;                                                       \
+  }                                                                            \
+  var = std::forward<decltype(FLEX_TMP_VAR)>(FLEX_TMP_VAR)
+
 // A Marco automatically use a auto variable to store the return value of a
 // function, which returns result, and check the status of the result, if ok,
 // continue to execute, otherwise, return the status. the macro accept the
diff --git a/flex/utils/service_utils.h b/flex/utils/service_utils.h
index bf817e0242bb..02170e7bd12d 100644
--- a/flex/utils/service_utils.h
+++ b/flex/utils/service_utils.h
@@ -38,6 +38,13 @@ namespace gs {
 
 static constexpr const char* CODEGEN_BIN = "load_plan_and_gen.sh";
 
+/// Util functions.
+inline int64_t GetCurrentTimeStamp() {
+  return std::chrono::duration_cast<std::chrono::milliseconds>(
+             std::chrono::system_clock::now().time_since_epoch())
+      .count();
+}
+
 class FlexException : public std::exception {
  public:
   explicit FlexException(std::string&& error_msg);
diff --git a/flex/utils/yaml_utils.cc b/flex/utils/yaml_utils.cc
index 9511192c9649..32ec1be8f38f 100644
--- a/flex/utils/yaml_utils.cc
+++ b/flex/utils/yaml_utils.cc
@@ -16,7 +16,7 @@
 
 #include "flex/utils/yaml_utils.h"
 #include <fstream>
-#include "nlohmann/json.hpp"
+
 namespace gs {
 std::vector<std::string> get_yaml_files(const std::string& plugin_dir) {
   std::filesystem::path dir_path = plugin_dir;
@@ -36,7 +36,7 @@ nlohmann::json convert_yaml_node_to_json(const YAML::Node& node) {
   try {
     switch (node.Type()) {
     case YAML::NodeType::Null: {
-      json = nullptr;
+      json = {};
       break;
     }
     case YAML::NodeType::Scalar: {
@@ -104,6 +104,37 @@ Result<std::string> get_json_string_from_yaml(const YAML::Node& node) {
   }
 }
 
+Result<std::string> get_yaml_string_from_yaml_node(const YAML::Node& node) {
+  try {
+    YAML::Emitter emitter;
+    write_yaml_node_to_yaml_string(node, emitter);
+    return std::string(emitter.c_str());
+  } catch (const YAML::BadConversion& e) {
+    return Result<std::string>(Status{StatusCode::IOError, e.what()});
+  }
+}
+
+Result<nlohmann::json> read_json_from_file(const std::string& file_path) {
+  if (!std::filesystem::exists(file_path)) {
+    return Result<nlohmann::json>(
+        Status{StatusCode::IOError, "File not exists: " + file_path});
+  }
+  // read string from file_path, and parse into json
+  std::ifstream file(file_path, std::ios::in);
+  if (!file.is_open()) {
+    return Result<nlohmann::json>(
+        Status{StatusCode::IOError, "Failed to open file: " + file_path});
+  }
+  std::string json_str((std::istreambuf_iterator<char>(file)),
+                       std::istreambuf_iterator<char>());
+  file.close();
+  try {
+    return nlohmann::json::parse(json_str);
+  } catch (const nlohmann::json::parse_error& e) {
+    return Result<nlohmann::json>(Status{StatusCode::InternalError, e.what()});
+  }
+}
+
 Status write_yaml_node_to_yaml_string(const YAML::Node& node,
                                       YAML::Emitter& emitter) {
   if (node.IsNull()) {
diff --git a/flex/utils/yaml_utils.h b/flex/utils/yaml_utils.h
index 500b983ad43c..f34b19f1df22 100644
--- a/flex/utils/yaml_utils.h
+++ b/flex/utils/yaml_utils.h
@@ -23,6 +23,7 @@
 #include "flex/utils/result.h"
 
 #include "glog/logging.h"
+#include "nlohmann/json.hpp"
 
 namespace gs {
 
@@ -32,9 +33,13 @@ Result<std::string> get_json_string_from_yaml(const std::string& file_path);
 
 Result<std::string> get_json_string_from_yaml(const YAML::Node& yaml_node);
 
+Result<nlohmann::json> read_json_from_file(const std::string& file_path);
+
 Status write_yaml_node_to_yaml_string(const YAML::Node& node,
                                       YAML::Emitter& emitter);
 
+Result<std::string> get_yaml_string_from_yaml_node(const YAML::Node& node);
+
 namespace config_parsing {
 template <typename T>
 bool get_scalar(YAML::Node node, const std::string& key, T& value) {


Test patch:
diff --git a/flex/tests/hqps/admin_http_test.cc b/flex/tests/hqps/admin_http_test.cc
index 0aa9e6326f25..db7dfb8931e4 100644
--- a/flex/tests/hqps/admin_http_test.cc
+++ b/flex/tests/hqps/admin_http_test.cc
@@ -16,6 +16,7 @@
 #include <boost/format.hpp>
 #include <string>
 #include "flex/proto_generated_gie/stored_procedure.pb.h"
+#include "flex/storages/metadata/graph_meta_store.h"
 #include "flex/third_party/httplib.h"
 #include "flex/utils/yaml_utils.h"
 #include "nlohmann/json.hpp"
@@ -31,9 +32,9 @@ static constexpr const char* CREATE_PROCEDURE_PAYLOAD_TEMPLATE =
     "\"query\": \"%4%\","
     "\"type\": \"cypher\"}";
 
-std::string generate_start_service_payload(const std::string& graph_name) {
-  boost::format formater("{\"graph_name\": \"%1%\"}");
-  return (formater % graph_name).str();
+std::string generate_start_service_payload(const std::string& graph_id) {
+  boost::format formater("{\"graph_id\": \"%1%\"}");
+  return (formater % graph_id).str();
 }
 
 std::string get_file_name_from_path(const std::string& file_path) {
@@ -43,20 +44,20 @@ std::string get_file_name_from_path(const std::string& file_path) {
   return file_name;
 }
 
-std::string generate_call_procedure_payload(const std::string& graph_name,
-                                            const std::string& procedure_name) {
+std::string generate_call_procedure_payload(const std::string& graph_id,
+                                            const std::string& procedure_id) {
   query::Query query;
-  query.mutable_query_name()->set_name(procedure_name);
+  query.mutable_query_name()->set_name(procedure_id);
   return query.SerializeAsString();
 }
 
-std::string generate_update_procedure_payload(const std::string& procedure_name,
+std::string generate_update_procedure_payload(const std::string& description,
                                               bool enabled) {
-  boost::format formater("{\"enable\": %1%, \"name\": \"%2%\"}");
-  return (formater % (enabled ? "true" : "false") % procedure_name).str();
+  boost::format formater("{\"enable\": %1%, \"description\": \"%2%\"}");
+  return (formater % (enabled ? "true" : "false") % description).str();
 }
 
-std::string generate_create_procedure_payload(const std::string& graph_name,
+std::string generate_create_procedure_payload(const std::string& graph_id,
                                               const std::string& procedure_path,
                                               bool enabled) {
   boost::format formater(CREATE_PROCEDURE_PAYLOAD_TEMPLATE);
@@ -68,7 +69,7 @@ std::string generate_create_procedure_payload(const std::string& graph_name,
   std::replace(query.begin(), query.end(), '\n', ' ');
   // get file name
   auto file_name = get_file_name_from_path(procedure_path);
-  return (formater % graph_name % (enabled ? "true" : "false") % file_name %
+  return (formater % graph_id % (enabled ? "true" : "false") % file_name %
           query)
       .str();
 }
@@ -95,37 +96,42 @@ std::string insert_raw_csv_dir(const std::string& raw_csv_dir,
 
 void run_builtin_graph_test(
     httplib::Client& admin_client, httplib::Client& query_client,
-    const std::string& graph_name,
+    const std::string& graph_id,
     const std::vector<std::pair<std::string, std::string>>& queries) {
   //-------0. get graph schema--------------------------------
-  auto res = admin_client.Get("/v1/graph/" + graph_name + "/schema");
+  auto res = admin_client.Get("/v1/graph/" + graph_id + "/schema");
   if (res->status != 200) {
-    LOG(FATAL) << "get graph schema failed for builtin graph" << graph_name
+    LOG(FATAL) << "get graph schema failed for builtin graph" << graph_id
                << ": " << res->body;
   }
+  std::vector<gs::PluginId> plugin_ids;
   //-------1. create procedures--------------------------------
   {
     for (auto& pair : queries) {
       auto query_name = pair.first;
       auto query_str = pair.second;
       boost::format formater(CREATE_PROCEDURE_PAYLOAD_TEMPLATE);
-      formater % graph_name % "true" % query_name % query_str;
+      formater % graph_id % "true" % query_name % query_str;
       std::string create_proc_payload0 = formater.str();
-      auto res = admin_client.Post("/v1/graph/" + graph_name + "/procedure",
+      auto res = admin_client.Post("/v1/graph/" + graph_id + "/procedure",
                                    create_proc_payload0, "text/plain");
       CHECK(res->status == 200) << "create procedure failed: " << res->body
                                 << ", for query: " << create_proc_payload0;
       LOG(INFO) << "Create procedure: " << create_proc_payload0
                 << ",response:" << res->body;
+      auto json = nlohmann::json::parse(res->body);
+      if (!json.contains("procedure_id")) {
+        LOG(FATAL) << "create procedure response does not contain plugin_id: "
+                   << res->body;
+      }
+      plugin_ids.emplace_back(json["procedure_id"].get<std::string>());
     }
   }
   //-------2. now call procedure should fail
   {
-    for (auto& pair : queries) {
-      auto query_name = pair.first;
-      auto query_str = pair.second;
+    for (auto& proc_id : plugin_ids) {
       query::Query query;
-      query.mutable_query_name()->set_name(query_name);
+      query.mutable_query_name()->set_name(proc_id);
       auto res = query_client.Post("/v1/query", query.SerializeAsString(),
                                    "text/plain");
       CHECK(res->status != 200);
@@ -137,7 +143,7 @@ void run_builtin_graph_test(
   }
   //-------3.restart service
   {
-    // graph_name is not specified, should restart on current graph.
+    // graph_id is not specified, should restart on current graph.
     std::string empty_payload;
     auto res =
         admin_client.Post("/v1/service/restart", empty_payload, "text/plain");
@@ -145,17 +151,15 @@ void run_builtin_graph_test(
   }
   {
     //-----3.1 get all procedures.
-    auto res = admin_client.Get("/v1/graph/" + graph_name + "/procedure");
+    auto res = admin_client.Get("/v1/graph/" + graph_id + "/procedure");
     CHECK(res->status == 200) << "get all procedures failed: " << res->body;
     LOG(INFO) << "get all procedures response: " << res->body;
   }
   //------4. now do the query
   {
-    for (auto& pair : queries) {
-      auto query_name = pair.first;
-      auto query_str = pair.second;
+    for (auto& plugin_id : plugin_ids) {
       query::Query query;
-      query.mutable_query_name()->set_name(query_name);
+      query.mutable_query_name()->set_name(plugin_id);
       auto res = query_client.Post("/v1/query", query.SerializeAsString(),
                                    "text/plain");
       CHECK(res->status == 200)
@@ -166,10 +170,10 @@ void run_builtin_graph_test(
   LOG(INFO) << "Pass builtin graph test";
 }
 
-void run_graph_tests(httplib::Client& cli, const std::string& graph_name,
-                     const std::string& schema_path,
-                     const std::string& import_path,
-                     const std::string& raw_data_dir) {
+gs::GraphId run_graph_tests(httplib::Client& cli,
+                            const std::string& schema_path,
+                            const std::string& import_path,
+                            const std::string& raw_data_dir) {
   //-------0. create graph--------------------------------
   // load schema_path to yaml and output yaml as json
   YAML::Node node;
@@ -193,9 +197,15 @@ void run_graph_tests(httplib::Client& cli, const std::string& graph_name,
     LOG(FATAL) << "Empty response: ";
   }
   LOG(INFO) << "create graph response: " << body;
+  // parse graph_id from response
+  nlohmann::json j = nlohmann::json::parse(body);
+  if (!j.contains("graph_id")) {
+    LOG(FATAL) << "create graph response does not contain graph_id: " << body;
+  }
+  gs::GraphId graph_id = j["graph_id"].get<std::string>();
 
   ///----1. get graph schema----------------------------
-  res = cli.Get("/v1/graph/" + graph_name + "/schema");
+  res = cli.Get("/v1/graph/" + graph_id + "/schema");
   if (res->status != 200) {
     LOG(FATAL) << "get graph schema failed: " << res->body;
   }
@@ -217,7 +227,7 @@ void run_graph_tests(httplib::Client& cli, const std::string& graph_name,
   LOG(INFO) << "list graph response: " << body;
 
   //----3. load graph-----------------------------------
-  res = cli.Post("/v1/graph/" + graph_name + "/dataloading",
+  res = cli.Post("/v1/graph/" + graph_id + "/dataloading",
                  insert_raw_csv_dir(raw_data_dir, import_path),
                  "application/json");
   if (res->status != 200) {
@@ -228,48 +238,56 @@ void run_graph_tests(httplib::Client& cli, const std::string& graph_name,
     LOG(FATAL) << "Empty response: ";
   }
   LOG(INFO) << "load graph response: " << body;
+  return graph_id;
 }
 
 // Create the procedure and call the procedure.
 void run_procedure_test(httplib::Client& client, httplib::Client& query_client,
-                        const std::string& graph_name,
+                        const std::string& graph_id,
                         const std::vector<std::pair<std::string, std::string>>&
                             builtin_graph_queries,
                         const std::vector<std::string>& procedures) {
   // First create the procedure with disabled state, then update with enabled
   // state.
   //-----0. get all procedures, should be empty----------------------
-  auto res = client.Get("/v1/graph/" + graph_name + "/procedure");
+  auto res = client.Get("/v1/graph/" + graph_id + "/procedure");
   CHECK(res->status == 200) << "get all procedures failed: " << res->body;
 
+  std::vector<gs::PluginId> plugin_ids;
   //-----1. create procedures----------------------------------------
   for (auto& procedure : procedures) {
     auto create_proc_payload =
-        generate_create_procedure_payload(graph_name, procedure, false);
+        generate_create_procedure_payload(graph_id, procedure, false);
     LOG(INFO) << "Creating procedure:" << create_proc_payload;
-    res = client.Post("/v1/graph/" + graph_name + "/procedure",
+    res = client.Post("/v1/graph/" + graph_id + "/procedure",
                       create_proc_payload, "text/plain");
     CHECK(res->status == 200) << "create procedure failed: " << res->body
                               << ", for query: " << create_proc_payload;
     LOG(INFO) << "response:" << res->body;
+    auto json = nlohmann::json::parse(res->body);
+    if (!json.contains("procedure_id")) {
+      LOG(FATAL) << "create procedure response does not contain plugin_id: "
+                 << res->body;
+    }
+    plugin_ids.emplace_back(json["procedure_id"].get<std::string>());
   }
   //-----2. get all procedures--------------------------------------
-  res = client.Get("/v1/graph/" + graph_name + "/procedure");
+  res = client.Get("/v1/graph/" + graph_id + "/procedure");
   CHECK(res->status == 200) << "get all procedures failed: " << res->body;
   LOG(INFO) << "get all procedures response: " << res->body;
   // Step4: update procedures
-  for (auto& procedure : procedures) {
-    auto proc_name = get_file_name_from_path(procedure);
+  for (size_t i = 0; i < plugin_ids.size(); ++i) {
+    auto& proc_id = plugin_ids[i];
     auto update_proc_payload =
-        generate_update_procedure_payload(proc_name, true);
-    res = client.Put("/v1/graph/" + graph_name + "/procedure/" + proc_name,
+        generate_update_procedure_payload("a example procedure", true);
+    res = client.Put("/v1/graph/" + graph_id + "/procedure/" + proc_id,
                      update_proc_payload, "text/plain");
     CHECK(res->status == 200) << "update procedure failed: " << res->body
                               << ", for query: " << update_proc_payload;
   }
 
   //-----3. start service on new graph-----------------------------------
-  auto start_service_payload = generate_start_service_payload(graph_name);
+  auto start_service_payload = generate_start_service_payload(graph_id);
   res = client.Post("/v1/service/start", start_service_payload, "text/plain");
   CHECK(res->status == 200) << "start service failed: " << res->body
                             << ", for query: " << start_service_payload;
@@ -289,27 +307,24 @@ void run_procedure_test(httplib::Client& client, httplib::Client& query_client,
   }
 
   //----4. call procedures-----------------------------------------------
-  for (auto& procedure : procedures) {
-    auto proc_name = get_file_name_from_path(procedure);
-    auto call_proc_payload =
-        generate_call_procedure_payload(graph_name, proc_name);
+  for (auto& proc_id : plugin_ids) {
+    auto call_proc_payload = generate_call_procedure_payload(graph_id, proc_id);
     res = query_client.Post("/v1/query", call_proc_payload, "text/plain");
     CHECK(res->status == 200) << "call procedure failed: " << res->body
                               << ", for query: " << call_proc_payload;
   }
   //----5. delete procedure by name-----------------------------------------
   if (procedures.size() > 0) {
-    auto proc_name = get_file_name_from_path(procedures[0]);
-    res = client.Delete("/v1/graph/" + graph_name + "/procedure/" + proc_name);
+    auto proc_id = plugin_ids[0];
+    res = client.Delete("/v1/graph/" + graph_id + "/procedure/" + proc_id);
     CHECK(res->status == 200) << "delete procedure failed: " << res->body;
   }
   //-----6. call procedure on deleted procedure------------------------------
   // Should return success, since the procedure will be deleted when restart
   // the service.
   if (procedures.size() > 0) {
-    auto proc_name = get_file_name_from_path(procedures[0]);
     auto call_proc_payload =
-        generate_call_procedure_payload(graph_name, proc_name);
+        generate_call_procedure_payload(graph_id, plugin_ids[0]);
     res = query_client.Post("/v1/query", call_proc_payload, "text/plain");
     CHECK(res->status == 200) << "call procedure failed: " << res->body
                               << ", for query: " << call_proc_payload;
@@ -317,8 +332,7 @@ void run_procedure_test(httplib::Client& client, httplib::Client& query_client,
   //-----7. get procedure by name--------------------------------------------
   // get the second procedure by name
   if (procedures.size() > 1) {
-    auto proc_name = get_file_name_from_path(procedures[1]);
-    res = client.Get("/v1/graph/" + graph_name + "/procedure/" + proc_name);
+    res = client.Get("/v1/graph/" + graph_id + "/procedure/" + plugin_ids[1]);
     CHECK(res->status == 200) << "get procedure failed: " << res->body;
   }
 }
@@ -345,8 +359,8 @@ void run_get_node_status(httplib::Client& cli) {
   LOG(INFO) << "get service status response: " << body;
 }
 
-void test_delete_graph(httplib::Client& cli, const std::string& graph_name) {
-  auto res = cli.Delete("/v1/graph/" + graph_name);
+void test_delete_graph(httplib::Client& cli, const std::string& graph_id) {
+  auto res = cli.Delete("/v1/graph/" + graph_id);
   if (res->status != 200) {
     LOG(FATAL) << "delete graph failed: " << res->body;
   }
@@ -357,12 +371,11 @@ void test_delete_graph(httplib::Client& cli, const std::string& graph_name) {
   LOG(INFO) << "delete graph response: " << body;
 }
 
-void remove_graph_if_exists(httplib::Client& cli,
-                            const std::string& graph_name) {
-  auto res = cli.Get("/v1/graph/" + graph_name + "/schema");
+void remove_graph_if_exists(httplib::Client& cli, const std::string& graph_id) {
+  auto res = cli.Get("/v1/graph/" + graph_id + "/schema");
   if (res->status == 200) {
-    LOG(INFO) << "graph " << graph_name << " exists, delete it";
-    test_delete_graph(cli, graph_name);
+    LOG(INFO) << "graph " << graph_id << " exists, delete it";
+    test_delete_graph(cli, graph_id);
   }
 }
 
@@ -389,21 +402,7 @@ int main(int argc, char** argv) {
   auto raw_data_dir = argv[5];
   std::vector<std::string> procedure_paths;
 
-  std::string builtin_graph_name = "modern_graph";
-
-  std::string graph_name;
-  {
-    // load yaml from schema_path
-    YAML::Node node;
-    try {
-      node = YAML::LoadFile(schema_path);
-    } catch (YAML::BadFile& e) {
-      LOG(ERROR) << "load schema file failed: " << e.what();
-      return -1;
-    }
-    graph_name = node["name"].as<std::string>();
-  }
-  LOG(INFO) << "graph name: " << graph_name;
+  std::string builtin_graph_id = "1";
 
   for (auto i = 6; i < argc; ++i) {
     procedure_paths.emplace_back(argv[i]);
@@ -417,18 +416,18 @@ int main(int argc, char** argv) {
   cli_query.set_read_timeout(60, 0);
   cli_query.set_write_timeout(60, 0);
 
-  remove_graph_if_exists(cli, graph_name);
+  // remove_graph_if_exists(cli, graph_id);
   std::vector<std::pair<std::string, std::string>> builtin_graph_queries = {
       {"query0", "MATCH(a) return COUNT(a);"}};
-  run_builtin_graph_test(cli, cli_query, builtin_graph_name,
+  run_builtin_graph_test(cli, cli_query, builtin_graph_id,
                          builtin_graph_queries);
-  run_graph_tests(cli, graph_name, schema_path, import_path, raw_data_dir);
+  auto graph_id = run_graph_tests(cli, schema_path, import_path, raw_data_dir);
   LOG(INFO) << "run graph tests done";
-  run_procedure_test(cli, cli_query, graph_name, builtin_graph_queries,
+  run_procedure_test(cli, cli_query, graph_id, builtin_graph_queries,
                      procedure_paths);
   LOG(INFO) << "run procedure tests done";
   run_get_node_status(cli);
-  test_delete_graph(cli, graph_name);
+  test_delete_graph(cli, graph_id);
   LOG(INFO) << "test delete graph done";
   return 0;
 }
\ No newline at end of file
diff --git a/flex/tests/hqps/engine_config_test.yaml b/flex/tests/hqps/engine_config_test.yaml
index e092baec9cbf..a5d42fca55f2 100644
--- a/flex/tests/hqps/engine_config_test.yaml
+++ b/flex/tests/hqps/engine_config_test.yaml
@@ -13,6 +13,8 @@ compute_engine:
   thread_num_per_worker: 1
   store:
     type: cpp-mcsr
+  metadata_store:
+    type: file # file/sqlite/etcd
 compiler:
   planner:
     is_on: true