refactor(interactive): refactor runtime (#4418)

Fixes

Author: liulx20

.gitmodules

@@ -10,3 +10,6 @@
 	path = learning_engine/graphlearn-for-pytorch
 	url = https://github.com/alibaba/graphlearn-for-pytorch.git
 
+[submodule "flex/third_party/parallel-hashmap"]
+	path = flex/third_party/parallel-hashmap
+	url = https://github.com/greg7mdp/parallel-hashmap.git

flex/CMakeLists.txt

@@ -207,6 +207,7 @@ add_definitions(-DRAPIDJSON_HAS_CXX11_RANGE_FOR=1)
 if (BUILD_ODPS_FRAGMENT_LOADER)
     include_directories(SYSTEM ${CMAKE_SOURCE_DIR}/third_party/odps/include)
 endif()
+include_directories(SYSTEM ${CMAKE_SOURCE_DIR}/third_party/parallel-hashmap)
 
 macro(install_flex_target target)
     install(TARGETS ${target}

flex/bin/adhoc_runner.cc

@@ -20,10 +20,12 @@
 #include <iostream>
 #include <vector>
 #include "flex/engines/graph_db/database/graph_db.h"
-#include "flex/engines/graph_db/runtime/adhoc/runtime.h"
+#include "flex/engines/graph_db/runtime/common/operators/retrieve/sink.h"
+#include "flex/engines/graph_db/runtime/execute/plan_parser.h"
+#include "flex/engines/graph_db/runtime/utils/opr_timer.h"
+#include "flex/proto_generated_gie/physical.pb.h"
 
 namespace bpo = boost::program_options;
-namespace bl = boost::leaf;
 
 std::string read_pb(const std::string& filename) {
   std::ifstream file(filename, std::ios::binary);
@@ -78,27 +80,11 @@ void load_params(const std::string& filename,
 gs::runtime::Context eval_plan(
     const physical::PhysicalPlan& plan, gs::ReadTransaction& txn,
     const std::map<std::string, std::string>& params) {
-  gs::runtime::Context ctx;
-  {
-    ctx = bl::try_handle_all(
-        [&plan, &txn, &params]() {
-          return gs::runtime::runtime_eval(plan, txn, params);
-        },
-        [&ctx](const gs::Status& err) {
-          LOG(FATAL) << "Error in execution: " << err.error_message();
-          return ctx;
-        },
-        [&](const bl::error_info& err) {
-          LOG(FATAL) << "boost leaf error: " << err.error().value() << ", "
-                     << err.exception()->what();
-          return ctx;
-        },
-        [&]() {
-          LOG(FATAL) << "Unknown error in execution";
-          return ctx;
-        });
-  }
-  return ctx;
+  gs::runtime::GraphReadInterface gri(txn);
+  gs::runtime::OprTimer timer;
+  return gs::runtime::PlanParser::get()
+      .parse_read_pipeline(gri.schema(), gs::runtime::ContextMeta(), plan)
+      .Execute(gri, gs::runtime::Context(), params, timer);
 }
 
 int main(int argc, char** argv) {
@@ -187,7 +173,7 @@ int main(int argc, char** argv) {
     auto& m = map[i % params_num];
     auto ctx = eval_plan(pb, txn, m);
     gs::Encoder output(outputs[i]);
-    gs::runtime::eval_sink(ctx, txn, output);
+    gs::runtime::Sink::sink(ctx, txn, output);
   }
   t1 += grape::GetCurrentTime();
 
@@ -197,7 +183,7 @@ int main(int argc, char** argv) {
     auto ctx = eval_plan(pb, txn, m);
     outputs[i].clear();
     gs::Encoder output(outputs[i]);
-    gs::runtime::eval_sink(ctx, txn, output);
+    gs::runtime::Sink::sink(ctx, txn, output);
   }
   t2 += grape::GetCurrentTime();
 
@@ -207,7 +193,7 @@ int main(int argc, char** argv) {
     auto ctx = eval_plan(pb, txn, m);
     outputs[i].clear();
     gs::Encoder output(outputs[i]);
-    gs::runtime::eval_sink(ctx, txn, output);
+    gs::runtime::Sink::sink(ctx, txn, output);
   }
   t3 += grape::GetCurrentTime();
 
@@ -231,4 +217,4 @@ int main(int argc, char** argv) {
   }
 
   return 0;
-}
+}

flex/bin/bulk_loader.cc

@@ -152,6 +152,16 @@ int main(int argc, char** argv) {
     return -1;
   }
 
+  {
+    std::error_code ec;
+    std::filesystem::copy(graph_schema_path, data_dir_path / "graph.yaml",
+                          std::filesystem::copy_options::overwrite_existing,
+                          ec);
+    if (ec) {
+      LOG(FATAL) << "Failed to copy graph schema file: " << ec.message();
+    }
+  }
+
   work_dir = data_dir_path.string();
 
   // Register handlers for SIGKILL, SIGINT, SIGTERM, SIGSEGV, SIGABRT

flex/bin/rt_server.cc

@@ -34,9 +34,8 @@ int main(int argc, char** argv) {
                                       bpo::value<uint32_t>()->default_value(1),
                                       "shard number of actor system")(
       "http-port,p", bpo::value<uint16_t>()->default_value(10000),
-      "http port of query handler")("graph-config,g", bpo::value<std::string>(),
-                                    "graph schema config file")(
-      "data-path,d", bpo::value<std::string>(), "data directory path")(
+      "http port of query handler")("data-path,d", bpo::value<std::string>(),
+                                    "data directory path")(
       "warmup,w", bpo::value<bool>()->default_value(false),
       "warmup graph data")("memory-level,m",
                            bpo::value<int>()->default_value(1));
@@ -62,14 +61,8 @@ int main(int argc, char** argv) {
   uint32_t shard_num = vm["shard-num"].as<uint32_t>();
   uint16_t http_port = vm["http-port"].as<uint16_t>();
 
-  std::string graph_schema_path = "";
   std::string data_path = "";
 
-  if (!vm.count("graph-config")) {
-    LOG(ERROR) << "graph-config is required";
-    return -1;
-  }
-  graph_schema_path = vm["graph-config"].as<std::string>();
   if (!vm.count("data-path")) {
     LOG(ERROR) << "data-path is required";
     return -1;
@@ -81,7 +74,7 @@ int main(int argc, char** argv) {
 
   double t0 = -grape::GetCurrentTime();
   auto& db = gs::GraphDB::get();
-
+  std::string graph_schema_path = data_path + "/graph.yaml";
   auto schema = gs::Schema::LoadFromYaml(graph_schema_path);
   if (!schema.ok()) {
     LOG(FATAL) << "Failed to load schema: " << schema.status().error_message();

flex/engines/graph_db/CMakeLists.txt

@@ -7,7 +7,7 @@ add_library(flex_graph_db SHARED ${GRAPH_DB_SRC_FILES})
 
 target_include_directories(flex_graph_db PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>)
 target_link_libraries(flex_graph_db flex_rt_mutable_graph flex_utils ${LIBGRAPELITE_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
-target_link_libraries(flex_graph_db flex_plan_proto runtime_adhoc)
+target_link_libraries(flex_graph_db runtime_execute)
 install_flex_target(flex_graph_db)
 
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/database/graph_db.h
@@ -28,7 +28,5 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/app/app_base.h
         DESTINATION include/flex/engines/graph_db/app)
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/app/hqps_app.h
         DESTINATION include/flex/engines/graph_db/app)
-install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/app/adhoc_app.h
-        DESTINATION include/flex/engines/graph_db/app)