fix cmakelist.txt

Author: liulx20

.gitmodules

@@ -13,6 +13,3 @@
 [submodule "flex/third_party/parallel-hashmap"]
 	path = flex/third_party/parallel-hashmap
 	url = https://github.com/greg7mdp/parallel-hashmap.git
-[submodule "flex/third_party/cppkafka"]
-	path = flex/third_party/cppkafka
-	url = https://github.com/mfontanini/cppkafka.git

flex/CMakeLists.txt

@@ -122,6 +122,21 @@ if (NOT yaml-cpp_FOUND)
     message(FATAL_ERROR "yaml-cpp not found, please install the yaml-cpp library")
 endif ()
 
+#find CppKafka-------------------------------------------------------------------
+if (BUILD_KAFKA_WAL_WRITER_PARSER)
+    find_package(CppKafka)
+    if (NOT CppKafka_FOUND)
+        message(FATAL_ERROR "cppkafka not found, please install cppkafka library")
+    else()
+        include_directories(SYSTEM ${CppKafka_INCLUDE_DIRS})
+        message(STATUS "cpp kafka include dir: ${CppKafka_INCLUDE_DIRS}")
+        set(CppKafka_LIBRARIES CppKafka::cppkafka)
+        add_definitions(-DBUILD_KAFKA_WAL_WRITER_PARSER)
+        message(STATUS "cppkafka found")
+    endif ()
+endif()
+
+
 #find boost----------------------------------------------------------------------
 find_package(Boost REQUIRED COMPONENTS system filesystem
              # required by folly
@@ -166,38 +181,6 @@ endif ()
 find_package(Protobuf REQUIRED)
 include_directories(${Protobuf_INCLUDE_DIRS})
 
-#find CppKafka-------------------------------------------------------------------
-if (BUILD_KAFKA_WAL_WRITER_PARSER)
-    find_package(CppKafka)
-    if (NOT CppKafka_FOUND)
-        message(STATUS "cppkafka not found, try to build with third_party/cppkafka")
-        set(CPPKAFKA_BOOST_STATIC_LIBS OFF)
-        add_subdirectory(third_party/cppkafka)
-        if (NOT EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka/CMakeLists.txt)
-            message(FATAL_ERROR "cppkafka not found, please run git submodule update --init --recursive")
-        endif ()
-        # Apply third_party/cppkafka_without_boost.patch 
-        # to remove the dependency of boost in cppkafka
-        # run git apply ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka_without_boost.patch
-        add_custom_target(apply_patch
-            COMMAND git apply ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka_without_boost.patch || true
-            WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka
-            COMMENT "Applying patch to cppkafka"
-            VERBATIM)
-
-        include_directories(SYSTEM ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka/include)
-        set (CppKafka_INSTALL_DIR ${CMAKE_CURRENT_SOURCE_DIR}/third_party/cppkafka/build)
-        list (APPEND CMAKE_PREFIX_PATH ${CppKafka_INSTALL_DIR})
-        set(CppKafka_LIBRARIES cppkafka)
-    else()
-        include_directories(SYSTEM ${CppKafka_INCLUDE_DIRS})
-        message(STATUS "cpp kafka include dir: ${CppKafka_INCLUDE_DIRS}")
-        set(CppKafka_LIBRARIES CppKafka::cppkafka)
-    endif ()
-    add_definitions(-DBUILD_KAFKA_WAL_WRITER_PARSER)
-    message(STATUS "cppkafka found")
-endif()
-
 if (BUILD_WITH_OTEL)
     find_package(opentelemetry-cpp CONFIG)
     if (OPENTELEMETRY_CPP_FOUND)

flex/engines/graph_db/CMakeLists.txt

@@ -4,9 +4,9 @@ file(GLOB_RECURSE GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/app/*.cc"
                                      "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/*.cc"
                                      "${CMAKE_CURRENT_SOURCE_DIR}/app/builtin/*.cc")
 
-if (BUILD_KAFKA_WAL_WRITER_PARSER)
-        list(APPEND GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_writer.cc"
-                                       "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_parser.cc")   
+if (NOT BUILD_KAFKA_WAL_WRITER_PARSER)
+        list(REMOVE_ITEM GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_writer.cc"
+                                            "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_parser.cc")   
 endif()
 
 add_library(flex_graph_db SHARED ${GRAPH_DB_SRC_FILES})
@@ -17,7 +17,6 @@ target_link_libraries(flex_graph_db runtime_execute)
 if (BUILD_KAFKA_WAL_WRITER_PARSER)
         target_link_libraries(flex_graph_db ${CppKafka_LIBRARIES})
 endif()
-add_dependencies(flex_graph_db apply_patch)
 install_flex_target(flex_graph_db)
 
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/database/graph_db.h

flex/engines/graph_db/app/kafka_wal_ingester_app.cc

@@ -0,0 +1,121 @@
+/** 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/engines/graph_db/app/kafka_wal_ingester_app.h"
+#include "cppkafka/cppkafka.h"
+#include "flex/engines/graph_db/database/graph_db.h"
+#include "flex/engines/graph_db/database/wal/kafka_wal_parser.h"
+
+namespace gs {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+
+class KafkaWalConsumer {
+ public:
+  struct CustomComparator {
+    inline bool operator()(const std::string& lhs, const std::string& rhs) {
+      const WalHeader* header1 = reinterpret_cast<const WalHeader*>(lhs.data());
+      const WalHeader* header2 = reinterpret_cast<const WalHeader*>(rhs.data());
+      return header1->timestamp > header2->timestamp;
+    }
+  };
+  static constexpr const std::chrono::milliseconds POLL_TIMEOUT =
+      std::chrono::milliseconds(100);
+
+  // always track all partitions and from begining
+  KafkaWalConsumer(cppkafka::Configuration config,
+                   const std::string& topic_name, int32_t thread_num) {
+    auto topic_partitions = get_all_topic_partitions(config, topic_name);
+    consumers_.reserve(topic_partitions.size());
+    for (size_t i = 0; i < topic_partitions.size(); ++i) {
+      consumers_.emplace_back(std::make_unique<cppkafka::Consumer>(config));
+      consumers_.back()->assign({topic_partitions[i]});
+    }
+  }
+
+  std::string poll() {
+    for (auto& consumer : consumers_) {
+      auto msg = consumer->poll();
+      if (msg) {
+        if (msg.get_error()) {
+          if (!msg.is_eof()) {
+            LOG(INFO) << "[+] Received error notification: " << msg.get_error();
+          }
+        } else {
+          std::string payload = msg.get_payload();
+          message_queue_.push(payload);
+          consumer->commit(msg);
+        }
+      }
+    }
+    if (message_queue_.empty()) {
+      return "";
+    }
+    std::string payload = message_queue_.top();
+    message_queue_.pop();
+    return payload;
+  }
+
+ private:
+  std::vector<std::unique_ptr<cppkafka::Consumer>> consumers_;
+  std::priority_queue<std::string, std::vector<std::string>, CustomComparator>
+      message_queue_;
+};
+
+bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
+                                Encoder& output) {
+  // TODO: get value from config
+  std::string kafka_brokers = std::string(input.get_string());
+  std::string brokers = std::string(input.get_string());
+  std::string topic_name = std::string(input.get_string());
+  std::string group_id = std::string(input.get_string());
+  std::string engine_endpoint = std::string(input.get_string());
+  LOG(INFO) << "Kafka brokers: " << kafka_brokers;
+  LOG(INFO) << "engine endpoint: " << engine_endpoint;
+
+  cppkafka::Configuration config = {{"metadata.broker.list", kafka_brokers},
+                                    {"group.id", group_id},
+                                    // Disable auto commit
+                                    {"enable.auto.commit", false}};
+  gs::KafkaWalConsumer consumer(config, topic_name, 1);
+  // TODO: how to make it stop
+  while (!graph.db().kafka_wal_ingester_state()) {
+    auto res = consumer.poll();
+    if (res.empty()) {
+      std::this_thread::sleep_for(gs::KafkaWalConsumer::POLL_TIMEOUT);
+      continue;
+    }
+
+    auto header = reinterpret_cast<const WalHeader*>(res.data());
+    if (header->type == 0) {
+      auto txn = graph.GetInsertTransaction();
+      txn.IngestWal(graph.graph(), txn.timestamp(),
+                    const_cast<char*>(res.data()) + sizeof(WalHeader),
+                    header->length, txn.allocator());
+    } else if (header->type == 1) {
+      auto txn = graph.GetUpdateTransaction();
+      txn.IngestWal(graph.graph(), graph.db().work_dir(), txn.timestamp(),
+                    const_cast<char*>(res.data()) + sizeof(WalHeader),
+                    header->length, txn.allocator());
+    } else {
+      LOG(ERROR) << "Unknown wal type: " << header->type;
+    }
+  }
+  return true;
+}
+AppWrapper KafkaWalIngesterAppFactory::CreateApp(const GraphDB& db) {
+  return AppWrapper(new KafkaWalIngesterApp(), NULL);
+}
+#endif
+}  // namespace gs

flex/engines/graph_db/app/kafka_wal_ingester_app.h

@@ -0,0 +1,45 @@
+/** 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 ENGINES_KAFKA_WAL_INGESTER_APP_H_
+#define ENGINES_KAFKA_WAL_INGESTER_APP_H_
+
+#include "flex/engines/graph_db/app/app_base.h"
+#include "flex/engines/graph_db/database/graph_db_session.h"
+
+namespace gs {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+// Ingest wal from kafka
+class KafkaWalIngesterApp : public WriteAppBase {
+ public:
+  KafkaWalIngesterApp() {}
+
+  AppType type() const override { return AppType::kBuiltIn; }
+
+  bool Query(GraphDBSession& graph, Decoder& input, Encoder& output) override;
+};
+
+class KafkaWalIngesterAppFactory : public AppFactoryBase {
+ public:
+  KafkaWalIngesterAppFactory() = default;
+  ~KafkaWalIngesterAppFactory() = default;
+
+  AppWrapper CreateApp(const GraphDB& db) override;
+};
+#endif
+
+}  // namespace gs
+
+#endif  // ENGINES_KAFKA_WAL_INGESTER_APP_H_

flex/engines/graph_db/database/graph_db.cc

@@ -28,6 +28,10 @@
 #include "flex/engines/graph_db/runtime/utils/cypher_runner_impl.h"
 #include "flex/utils/yaml_utils.h"
 
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+#include "flex/engines/graph_db/app/kafka_wal_ingester_app.h"
+#endif
+
 #include "flex/third_party/httplib.h"
 
 namespace gs {
@@ -274,6 +278,38 @@ void GraphDB::Close() {
   std::fill(app_factories_.begin(), app_factories_.end(), nullptr);
 }
 
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+bool GraphDB::kafka_wal_ingester_state() const {
+  return kafka_wal_ingester_thread_running_.load();
+}
+
+void GraphDB::start_kafka_wal_ingester(const std::string& kafka_brokers,
+                                       const std::string& brokers,
+                                       const std::string& topic_name,
+                                       const std::string& group_id,
+                                       const std::string& engine_endpoint) {
+  if (kafka_wal_ingester_thread_running_) {
+    kafka_wal_ingester_thread_running_ = false;
+    kafka_wal_ingester_thread_.join();
+  }
+  std::vector<char> buffer;
+  gs::Encoder encoder(buffer);
+  encoder.put_string(kafka_brokers);
+  encoder.put_string(brokers);
+  encoder.put_string(topic_name);
+  encoder.put_string(group_id);
+  encoder.put_string(engine_endpoint);
+  gs::Decoder decoder(buffer.data(), buffer.size());
+  KafkaWalIngesterApp().Query(GetSession(0), decoder, encoder);
+}
+
+void GraphDB::stop_kafka_wal_ingester() {
+  kafka_wal_ingester_thread_running_ = false;
+  kafka_wal_ingester_thread_.join();
+}
+
+#endif
+
 ReadTransaction GraphDB::GetReadTransaction(int thread_id) {
   return contexts_[thread_id].session.GetReadTransaction();
 }

flex/engines/graph_db/database/graph_db.h

@@ -169,6 +169,18 @@ class GraphDB {
 
   inline const GraphDBConfig& config() const { return config_; }
 
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+  bool kafka_wal_ingester_state() const;
+
+  void start_kafka_wal_ingester(const std::string& kafka_brokers,
+                                const std::string& brokers,
+                                const std::string& topic_name,
+                                const std::string& group_id,
+                                const std::string& engine_endpoint);
+
+  void stop_kafka_wal_ingester();
+#endif
+
  private:
   bool registerApp(const std::string& path, uint8_t index = 0);
 
@@ -204,6 +216,9 @@ class GraphDB {
   std::thread monitor_thread_;
   bool monitor_thread_running_;
 
+  std::thread kafka_wal_ingester_thread_;
+  std::atomic<bool> kafka_wal_ingester_thread_running_;
+
   timestamp_t last_compaction_ts_;
   bool compact_thread_running_ = false;
   std::thread compact_thread_;

flex/engines/graph_db/database/insert_transaction.cc

@@ -226,6 +226,8 @@ const Schema& InsertTransaction::schema() const { return graph_.schema(); }
 
 const GraphDBSession& InsertTransaction::GetSession() const { return session_; }
 
+Allocator& InsertTransaction::allocator() const { return alloc_; }
+
 #define likely(x) __builtin_expect(!!(x), 1)
 
 bool InsertTransaction::get_vertex_with_retries(MutablePropertyFragment& graph,

flex/engines/graph_db/database/insert_transaction.h

@@ -61,6 +61,8 @@ class InsertTransaction {
 
   const GraphDBSession& GetSession() const;
 
+  Allocator& allocator() const;
+
  private:
   void clear();
 

flex/engines/graph_db/database/update_transaction.cc

@@ -102,6 +102,8 @@ UpdateTransaction::~UpdateTransaction() { release(); }
 
 timestamp_t UpdateTransaction::timestamp() const { return timestamp_; }
 
+Allocator& UpdateTransaction::allocator() { return alloc_; }
+
 bool UpdateTransaction::Commit() {
   if (timestamp_ == std::numeric_limits<timestamp_t>::max()) {
     return true;

flex/engines/graph_db/database/update_transaction.h

@@ -153,6 +153,8 @@ class UpdateTransaction {
 
   const GraphDBSession& GetSession() const;
 
+  Allocator& allocator();
+
  private:
   friend class GraphDBSession;
   bool batch_commit(UpdateBatch& batch);