refactor(interactive): Refactoring Wal parser and writer for GraphDB (#4504)

- Extract two interfaces`IWalWriter` and `IWalParser` to decouple
wal-related implementation from GraphDB.
- Provide local-file based WalWriter and WalParser.
- Introducing configuration `wal_uri`, which is by default set to
`{GRAPH_DATA_DIR}/wal`

Author: zhanglei1949

docs/flex/interactive/configuration.md

@@ -115,6 +115,7 @@ In this following table, we use the `.` notation to represent the hierarchy with
 | log_level     |  INFO   | The level of database log, INFO/WARNING/ERROR/FATAL | 0.0.1 |
 | verbose_level     |  0   | The verbose level of database log, should be a int | 0.0.3 |
 | compute_engine.thread_num_per_worker | 1 | The number of threads will be used to process the queries. Increase the number can benefit the query throughput | 0.0.1 |
+| compute_engine.wal_uri    | file://{GRAPH_DATA_DIR}/wal | The location where Interactive will store and access WALs. `GRAPH_DATA_DIR` is a placeholder that will be populated by Interactive. | 0.5 |
 | compiler.planner.is_on | true | Determines if query optimization is enabled for compiling Cypher queries  | 0.0.1 |
 | compiler.planner.opt | RBO | Specifies the optimizer to be used for query optimization. Currently, only the Rule-Based Optimizer (RBO) is supported | 0.0.1 |
 | compiler.planner.rules.FilterMatchRule | N/A | An optimization rule that pushes filter (`Where`) conditions into the `Match` clause | 0.0.1 |
@@ -125,6 +126,7 @@ In this following table, we use the `.` notation to represent the hierarchy with
 | http_service.max_content_length | 1GB | The maximum length of a http request that admin http service could handle | 0.5 |
 | storage.string_default_max_length | 256 | The default maximum size for a string field | 0.5 |
 
+
 #### TODOs
 
 We currently only allow service configuration during instance deployment. In the near future, we will support:

flex/bin/interactive_server.cc

@@ -240,8 +240,10 @@ int main(int argc, char** argv) {
       gs::init_codegen_proxy(vm, engine_config_file, graph_schema_path);
     }
     db.Close();
-    auto load_res =
-        db.Open(schema_res.value(), data_path, service_config.shard_num);
+    gs::GraphDBConfig config(schema_res.value(), data_path, "",
+                             service_config.shard_num);
+    config.wal_uri = service_config.wal_uri;
+    auto load_res = db.Open(config);
     if (!load_res.ok()) {
       LOG(FATAL) << "Failed to load graph from data directory: "
                  << load_res.status().error_message();

flex/bin/rt_server.cc

@@ -40,7 +40,9 @@ int main(int argc, char** argv) {
       "warmup graph data")("memory-level,m",
                            bpo::value<int>()->default_value(1))(
       "compiler-path,c", bpo::value<std::string>()->default_value(""))(
-      "sharding-mode", bpo::value<std::string>()->default_value("cooperative"));
+      "sharding-mode", bpo::value<std::string>()->default_value("cooperative"))(
+      "wal-uri",
+      bpo::value<std::string>()->default_value("file://{GRAPH_DATA_DIR}/wal"));
   google::InitGoogleLogging(argv[0]);
   FLAGS_logtostderr = true;
 
@@ -84,6 +86,7 @@ int main(int argc, char** argv) {
   }
   gs::GraphDBConfig config(schema.value(), data_path, compiler_path, shard_num);
   config.memory_level = memory_level;
+  config.wal_uri = vm["wal-uri"].as<std::string>();
   if (config.memory_level >= 2) {
     config.enable_auto_compaction = true;
   }

flex/engines/graph_db/CMakeLists.txt

@@ -1,6 +1,7 @@
 add_subdirectory(runtime)
 file(GLOB_RECURSE GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/app/*.cc"
   	                             "${CMAKE_CURRENT_SOURCE_DIR}/database/*.cc"
+                                     "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/*.cc"
                                      "${CMAKE_CURRENT_SOURCE_DIR}/app/builtin/*.cc")
 
 add_library(flex_graph_db SHARED ${GRAPH_DB_SRC_FILES})
@@ -19,7 +20,6 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/database/graph_db.h
               ${CMAKE_CURRENT_SOURCE_DIR}/database/update_transaction.h
               ${CMAKE_CURRENT_SOURCE_DIR}/database/compact_transaction.h
               ${CMAKE_CURRENT_SOURCE_DIR}/database/version_manager.h
-              ${CMAKE_CURRENT_SOURCE_DIR}/database/wal.h
               ${CMAKE_CURRENT_SOURCE_DIR}/database/transaction_utils.h
               ${CMAKE_CURRENT_SOURCE_DIR}/database/graph_db_operations.h
         DESTINATION include/flex/engines/graph_db/database)
@@ -29,4 +29,10 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/app/app_base.h
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/app/hqps_app.h
         DESTINATION include/flex/engines/graph_db/app)
 
+install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/database/wal/wal.h
+        ${CMAKE_CURRENT_SOURCE_DIR}/database/wal/local_wal_parser.h
+        ${CMAKE_CURRENT_SOURCE_DIR}/database/wal/local_wal_writer.h
+        DESTINATION include/flex/engines/graph_db/database/wal)
+
+
 

flex/engines/graph_db/database/compact_transaction.cc

@@ -15,13 +15,13 @@
 
 #include "flex/engines/graph_db/database/compact_transaction.h"
 #include "flex/engines/graph_db/database/version_manager.h"
-#include "flex/engines/graph_db/database/wal.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
 #include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
 
 namespace gs {
 
 CompactTransaction::CompactTransaction(MutablePropertyFragment& graph,
-                                       WalWriter& logger, VersionManager& vm,
+                                       IWalWriter& logger, VersionManager& vm,
                                        timestamp_t timestamp)
     : graph_(graph), logger_(logger), vm_(vm), timestamp_(timestamp) {
   arc_.Resize(sizeof(WalHeader));

flex/engines/graph_db/database/compact_transaction.h

@@ -22,12 +22,12 @@
 namespace gs {
 
 class MutablePropertyFragment;
-class WalWriter;
+class IWalWriter;
 class VersionManager;
 
 class CompactTransaction {
  public:
-  CompactTransaction(MutablePropertyFragment& graph, WalWriter& logger,
+  CompactTransaction(MutablePropertyFragment& graph, IWalWriter& logger,
                      VersionManager& vm, timestamp_t timestamp);
   ~CompactTransaction();
 
@@ -39,7 +39,7 @@ class CompactTransaction {
 
  private:
   MutablePropertyFragment& graph_;
-  WalWriter& logger_;
+  IWalWriter& logger_;
   VersionManager& vm_;
   timestamp_t timestamp_;
 

flex/engines/graph_db/database/graph_db.cc

@@ -23,7 +23,7 @@
 #include "flex/engines/graph_db/app/hqps_app.h"
 #include "flex/engines/graph_db/app/server_app.h"
 #include "flex/engines/graph_db/database/graph_db_session.h"
-#include "flex/engines/graph_db/database/wal.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
 #include "flex/engines/graph_db/runtime/execute/plan_parser.h"
 #include "flex/engines/graph_db/runtime/utils/cypher_runner_impl.h"
 #include "flex/utils/yaml_utils.h"
@@ -34,18 +34,20 @@ namespace gs {
 
 struct SessionLocalContext {
   SessionLocalContext(GraphDB& db, const std::string& work_dir, int thread_id,
-                      MemoryStrategy allocator_strategy)
+                      MemoryStrategy allocator_strategy,
+                      std::unique_ptr<IWalWriter> in_logger)
       : allocator(allocator_strategy,
                   (allocator_strategy != MemoryStrategy::kSyncToFile
                        ? ""
                        : thread_local_allocator_prefix(work_dir, thread_id))),
-        session(db, allocator, logger, work_dir, thread_id) {}
-  ~SessionLocalContext() { logger.close(); }
+        logger(std::move(in_logger)),
+        session(db, allocator, *logger, work_dir, thread_id) {}
+  ~SessionLocalContext() { logger->close(); }
   Allocator allocator;
   char _padding0[128 - sizeof(Allocator) % 128];
-  WalWriter logger;
-  char _padding1[4096 - sizeof(WalWriter) - sizeof(Allocator) -
-                 sizeof(_padding0)];
+  std::unique_ptr<IWalWriter> logger;
+  char _padding1[4096 - sizeof(std::unique_ptr<IWalWriter>) -
+                 sizeof(Allocator) - sizeof(_padding0)];
   GraphDBSession session;
   char _padding2[4096 - sizeof(GraphDBSession) % 4096];
 };
@@ -64,6 +66,8 @@ GraphDB::~GraphDB() {
 
     free(contexts_);
   }
+  WalWriterFactory::Finalize();
+  WalParserFactory::Finalize();
 }
 
 GraphDB& GraphDB::get() {
@@ -86,6 +90,7 @@ Result<bool> GraphDB::Open(const Schema& schema, const std::string& data_dir,
 }
 
 Result<bool> GraphDB::Open(const GraphDBConfig& config) {
+  config_ = config;
   const std::string& data_dir = config.data_dir;
   const Schema& schema = config.schema;
   if (!std::filesystem::exists(data_dir)) {
@@ -142,7 +147,7 @@ Result<bool> GraphDB::Open(const GraphDBConfig& config) {
     allocator_strategy = MemoryStrategy::kHugepagePrefered;
   }
 
-  openWalAndCreateContexts(data_dir, allocator_strategy);
+  openWalAndCreateContexts(config, data_dir, allocator_strategy);
 
   if ((!create_empty_graph) && config.warmup) {
     graph_.Warmup(thread_num_);
@@ -228,8 +233,9 @@ Result<bool> GraphDB::Open(const GraphDBConfig& config) {
           VLOG(10) << "Trigger auto compaction";
           last_compaction_at = query_num_after;
           timestamp_t ts = this->version_manager_.acquire_update_timestamp();
-          auto txn = CompactTransaction(this->graph_, this->contexts_[0].logger,
-                                        this->version_manager_, ts);
+          auto txn =
+              CompactTransaction(this->graph_, *this->contexts_[0].logger,
+                                 this->version_manager_, ts);
           OutputCypherProfiles("./" + std::to_string(ts) + "_");
           txn.Commit();
           VLOG(10) << "Finish compaction";
@@ -344,7 +350,7 @@ void GraphDB::GetAppInfo(Encoder& output) {
 
 static void IngestWalRange(SessionLocalContext* contexts,
                            MutablePropertyFragment& graph,
-                           const WalsParser& parser, uint32_t from, uint32_t to,
+                           const IWalParser& parser, uint32_t from, uint32_t to,
                            int thread_num) {
   std::atomic<uint32_t> cur_ts(from);
   std::vector<std::thread> threads(thread_num);
@@ -372,11 +378,10 @@ static void IngestWalRange(SessionLocalContext* contexts,
   }
 }
 
-void GraphDB::ingestWals(const std::vector<std::string>& wals,
-                         const std::string& work_dir, int thread_num) {
-  WalsParser parser(wals);
+void GraphDB::ingestWals(IWalParser& parser, const std::string& work_dir,
+                         int thread_num) {
   uint32_t from_ts = 1;
-  for (auto& update_wal : parser.update_wals()) {
+  for (auto& update_wal : parser.get_update_wals()) {
     uint32_t to_ts = update_wal.timestamp;
     if (from_ts < to_ts) {
       IngestWalRange(contexts_, graph_, parser, from_ts, to_ts, thread_num);
@@ -452,30 +457,36 @@ void GraphDB::initApps(
             << ", from " << plugins.size();
 }
 
-void GraphDB::openWalAndCreateContexts(const std::string& data_dir,
+void GraphDB::openWalAndCreateContexts(const GraphDBConfig& config,
+                                       const std::string& data_dir,
                                        MemoryStrategy allocator_strategy) {
-  std::string wal_dir_path = wal_dir(data_dir);
-  if (!std::filesystem::exists(wal_dir_path)) {
-    std::filesystem::create_directory(wal_dir_path);
-  }
-
-  std::vector<std::string> wal_files;
-  for (const auto& entry : std::filesystem::directory_iterator(wal_dir_path)) {
-    wal_files.push_back(entry.path().string());
-  }
-
+  WalWriterFactory::Init();
+  WalParserFactory::Init();
   contexts_ = static_cast<SessionLocalContext*>(
       aligned_alloc(4096, sizeof(SessionLocalContext) * thread_num_));
   std::filesystem::create_directories(allocator_dir(data_dir));
+
+  // Open the wal writer.
+  std::string wal_uri = config.wal_uri;
+  if (wal_uri.empty()) {
+    LOG(ERROR) << "wal_uri is not set, use default wal_uri";
+    wal_uri = wal_dir(data_dir);
+  } else if (wal_uri.find("{GRAPH_DATA_DIR}") != std::string::npos) {
+    LOG(INFO) << "Template {GRAPH_DATA_DIR} found in wal_uri, replace it with "
+                 "data_dir";
+    wal_uri = std::regex_replace(wal_uri, std::regex("\\{GRAPH_DATA_DIR\\}"),
+                                 data_dir);
+  }
+  VLOG(1) << "Using wal uri: " << wal_uri;
   for (int i = 0; i < thread_num_; ++i) {
     new (&contexts_[i])
-        SessionLocalContext(*this, data_dir, i, allocator_strategy);
+        SessionLocalContext(*this, data_dir, i, allocator_strategy,
+                            WalWriterFactory::CreateWalWriter(wal_uri));
+    contexts_[i].logger->open(wal_uri, i);
   }
-  ingestWals(wal_files, data_dir, thread_num_);
 
-  for (int i = 0; i < thread_num_; ++i) {
-    contexts_[i].logger.open(wal_dir_path, i);
-  }
+  auto wal_parser = WalParserFactory::CreateWalParser(wal_uri);
+  ingestWals(*wal_parser, data_dir, thread_num_);
 
   initApps(graph_.schema().GetPlugins());
   VLOG(1) << "Successfully restore load plugins";

flex/engines/graph_db/database/graph_db.h

@@ -41,6 +41,7 @@ class GraphDBSession;
 struct SessionLocalContext;
 
 struct GraphDBConfig {
+  GraphDBConfig() = default;
   GraphDBConfig(const Schema& schema_, const std::string& data_dir_,
                 const std::string& compiler_path_ = "", int thread_num_ = 1)
       : schema(schema_),
@@ -50,7 +51,8 @@ struct GraphDBConfig {
         warmup(false),
         enable_monitoring(false),
         enable_auto_compaction(false),
-        memory_level(1) {}
+        memory_level(1),
+        wal_uri("") {}
 
   Schema schema;
   std::string data_dir;
@@ -67,6 +69,9 @@ struct GraphDBConfig {
     3 - force hugepages;
   */
   int memory_level;
+  std::string wal_uri;  // Indicate the where shall we store the wal files.
+                        // could be file://{GRAPH_DATA_DIR}/wal or other scheme
+                        // that interactive supports
 };
 
 class GraphDB {
@@ -162,17 +167,20 @@ class GraphDB {
 
   void OutputCypherProfiles(const std::string& prefix);
 
+  inline const GraphDBConfig& config() const { return config_; }
+
  private:
   bool registerApp(const std::string& path, uint8_t index = 0);
 
-  void ingestWals(const std::vector<std::string>& wals,
-                  const std::string& work_dir, int thread_num);
+  void ingestWals(IWalParser& parser, const std::string& work_dir,
+                  int thread_num);
 
   void initApps(
       const std::unordered_map<std::string, std::pair<std::string, uint8_t>>&
           plugins);
 
-  void openWalAndCreateContexts(const std::string& data_dir_path,
+  void openWalAndCreateContexts(const GraphDBConfig& config,
+                                const std::string& data_dir,
                                 MemoryStrategy allocator_strategy);
 
   void showAppMetrics() const;
@@ -181,6 +189,7 @@ class GraphDB {
 
   friend class GraphDBSession;
 
+  GraphDBConfig config_;
   std::string work_dir_;
   SessionLocalContext* contexts_;
 

flex/engines/graph_db/database/graph_db_session.h

@@ -31,7 +31,7 @@
 namespace gs {
 
 class GraphDB;
-class WalWriter;
+class IWalWriter;
 
 class GraphDBSession {
  public:
@@ -49,7 +49,7 @@ class GraphDBSession {
   static constexpr const char* kCypherJsonStr = "\x01";
   static constexpr const char* kCypherProtoAdhocStr = "\x02";
   static constexpr const char* kCypherProtoProcedureStr = "\x03";
-  GraphDBSession(GraphDB& db, Allocator& alloc, WalWriter& logger,
+  GraphDBSession(GraphDB& db, Allocator& alloc, IWalWriter& logger,
                  const std::string& work_dir, int thread_id)
       : db_(db),
         alloc_(alloc),
@@ -183,7 +183,7 @@ class GraphDBSession {
   }
   GraphDB& db_;
   Allocator& alloc_;
-  WalWriter& logger_;
+  IWalWriter& logger_;
   std::string work_dir_;
   int thread_id_;
 

flex/engines/graph_db/database/insert_transaction.cc

@@ -17,15 +17,15 @@
 #include "flex/engines/graph_db/database/graph_db_session.h"
 #include "flex/engines/graph_db/database/transaction_utils.h"
 #include "flex/engines/graph_db/database/version_manager.h"
-#include "flex/engines/graph_db/database/wal.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
 #include "flex/engines/graph_db/runtime/utils/cypher_runner_impl.h"
 #include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
 #include "flex/utils/allocators.h"
 namespace gs {
 
 InsertTransaction::InsertTransaction(const GraphDBSession& session,
                                      MutablePropertyFragment& graph,
-                                     Allocator& alloc, WalWriter& logger,
+                                     Allocator& alloc, IWalWriter& logger,
                                      VersionManager& vm, timestamp_t timestamp)
 
     : session_(session),