fix(interactive): disable repair wheel in coordinator dockerfile (#4517)

Fixes #4454

---------

Signed-off-by: siyuan0322 <[email protected]>

fix aocc

Committed-by: [email protected] from Dev container

fix installation

Committed-by: [email protected] from Dev container

fix linking,todo: verify correctness

Committed-by: xiaolei.zl from Dev container

impl kafka writer and parser

Committed-by: xiaolei.zl from Dev container

Committed-by: xiaolei.zl from Dev container

Committed-by: xiaolei.zl from Dev container

minor fix

Committed-by: xiaolei.zl from Dev container

Committed-by: xiaolei.zl from Dev container

Committed-by: xiaolei.zl from Dev container

fix

Committed-by: xiaolei.zl from Dev container

Committed-by: xiaolei.zl from Dev container

Author: siyuan0322

.github/workflows/interactive.yml

@@ -377,6 +377,23 @@ jobs:
         bash hqps_adhoc_test.sh ${INTERACTIVE_WORKSPACE} graph_algo \
           ${GITHUB_WORKSPACE}/flex/tests/hqps/interactive_config_test.yaml
 
+    - name: Run kafka wal test
+      env:
+        GS_TEST_DIR: ${{ github.workspace }}/gstest
+      run: |
+        wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
+        tar -zxf kafka_2.13-3.9.0.tgz
+        cd kafka_2.13-3.9.0
+        KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
+        bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties
+        bin/kafka-server-start.sh config/kraft/reconfig-server.properties &
+
+        bin/kafka-topics.sh --create --topic kafka-test --bootstrap-server localhost:9092
+
+        ./tests/hqps/kafka_test localhost:902 kafka-test
+
+        pkill -f kafka-server-start.sh
+
     - name: Run Gremlin test on modern graph
       env:
         GS_TEST_DIR: ${{ github.workspace }}/gstest

.gitmodules

@@ -13,3 +13,6 @@
 [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

docs/flex/interactive/development/dev_and_test.md

@@ -262,3 +262,40 @@ In Interactive's execution engine, transactions such as `ReadTransaction`, `Upda
 2. If a transaction returns `false` during the `commit()` process, the error occurred prior to applying the WAL to the graph data. This type of failure could arise during the construction of the WAL or during its writing phase.
 
 3. It is important to note that errors can still occur when replaying the WAL to the graph database. Replaying might fail due to limitations in resources or due to unforeseen bugs. **However,** any errors encountered during this stage will be handled via exceptions or may result in process failure. Currently, there is no established mechanism to handle such failures. Future improvements should focus on implementing failover strategies, potentially allowing the GraphDB to continue replaying the WAL until it succeeds.
+
+## Persisting WAL to kafka
+
+Kafka-based WAL storages is also provided. Follows [kafka-quick-start](https://kafka.apache.org/quickstart).
+
+### Install kafka
+
+```bash
+wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
+tar -zxf kafka_2.13-3.9.0.tgz
+cd kafka_2.13-3.9.0
+```
+
+### kafka with kraft
+
+```bash
+KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
+bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties
+bin/kafka-server-start.sh config/kraft/reconfig-server.properties
+```
+
+### Create topic
+
+```bash
+bin/kafka-topics.sh --create --topic kafka-test --bootstrap-server localhost:9092
+# describe the topic
+bin/kafka-topics.sh --describe --topic kafka-test --bootstrap-server localhost:9092
+```
+
+### Test KafkaWalWriter and KafkaWalParser
+
+```bash
+cd flex && mkdir build
+cd build && cmake .. -DBUILD_TEST=ON && make -j
+./tests/hqps/kafka_test localhost:902 kafka-test
+# run the kafka tst
+```

flex/CMakeLists.txt

@@ -17,6 +17,7 @@ option(USE_PTHASH "Whether to use pthash" OFF)
 option(OPTIMIZE_FOR_HOST "Whether to optimize on host" ON) # Whether to build optimized code on host
 option(USE_STATIC_ARROW "Whether to use static arrow" OFF) # Whether to link arrow statically, default is OFF
 option(BUILD_WITH_OTEL "Whether to build with opentelemetry-cpp" OFF) # Whether to build with opentelemetry-cpp, default is OFF
+option(BUILD_KAFKA_WAL_WRITER_PARSER "Whether to build kafka wal writer and wal parser" ON) # Whether to build kafka wal writer and wal parser, default is ON
 
 #print options
 message(STATUS "Build test: ${BUILD_TEST}")
@@ -165,6 +166,38 @@ 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,11 +4,20 @@ 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")   
+endif()
+
 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 ${LIBGRAPELITE_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
 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/database/wal/kafka_wal_parser.cc

@@ -0,0 +1,154 @@
+/** 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/database/wal/kafka_wal_parser.h"
+#include <fcntl.h>
+#include <sys/mman.h>
+#include <unistd.h>
+#include <filesystem>
+#include "flex/engines/graph_db/database/wal/wal.h"
+
+namespace gs {
+
+std::vector<cppkafka::TopicPartition> get_all_topic_partitions(
+    const cppkafka::Configuration& config, const std::string& topic_name) {
+  std::vector<cppkafka::TopicPartition> partitions;
+  cppkafka::Consumer consumer(config);  // tmp consumer
+  auto meta_vector = consumer.get_metadata().get_topics({topic_name});
+  if (meta_vector.empty()) {
+    LOG(WARNING) << "Failed to get metadata for topic " << topic_name
+                 << ", maybe the topic does not exist";
+    return {};
+  }
+  auto metadata = meta_vector.front().get_partitions();
+  for (const auto& partition : metadata) {
+    partitions.push_back(cppkafka::TopicPartition(
+        topic_name, partition.get_id(), 0));  // from the beginning
+  }
+  return partitions;
+}
+
+std::unique_ptr<IWalParser> KafkaWalParser::Make() {
+  const char* broker_list = std::getenv("KAFKA_BROKER_LIST");
+  if (broker_list == nullptr) {
+    LOG(FATAL) << "KAFKA_BROKER_LIST is not set";
+  }
+  const char* group_id = std::getenv("KAFKA_GROUP_ID");
+  std::string group_id_str = group_id ? group_id : "interactive_consumer";
+  cppkafka::Configuration config = {{"metadata.broker.list", broker_list},
+                                    {"group.id", group_id_str}};
+  return std::unique_ptr<IWalParser>(new KafkaWalParser(config));
+}
+
+KafkaWalParser::KafkaWalParser(const cppkafka::Configuration& config)
+    : consumer_(nullptr),
+      insert_wal_list_(NULL),
+      insert_wal_list_size_(0),
+      last_ts_(0),
+      config_(config) {
+  consumer_ = std::make_unique<cppkafka::Consumer>(config);
+}
+
+void KafkaWalParser::open(const std::string& topic_name) {
+  auto topic_partitions = get_all_topic_partitions(config_, topic_name);
+  open(topic_partitions);
+}
+
+void KafkaWalParser::open(
+    const std::vector<cppkafka::TopicPartition>& topic_partitions) {
+  consumer_->assign(topic_partitions);
+  insert_wal_list_ = static_cast<WalContentUnit*>(
+      mmap(NULL, IWalWriter::MAX_WALS_NUM * sizeof(WalContentUnit),
+           PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_NORESERVE,
+           -1, 0));
+  insert_wal_list_size_ = IWalWriter::MAX_WALS_NUM;
+
+  while (true) {
+    auto msgs = consumer_->poll_batch(MAX_BATCH_SIZE);
+    if (msgs.empty() || msgs.empty()) {
+      LOG(INFO) << "No message are polled, the topic has been all consumed.";
+      break;
+    }
+    for (auto& msg : msgs) {
+      if (msg) {
+        if (msg.get_error()) {
+          if (!msg.is_eof()) {
+            LOG(INFO) << "[+] Received error notification: " << msg.get_error();
+          }
+        } else {
+          message_vector_.emplace_back(msg.get_payload());
+        }
+      }
+    }
+  }
+
+  for (auto& wal : message_vector_) {
+    LOG(INFO) << "Got wal:" << wal.size();
+    const char* payload = wal.data();
+    const WalHeader* header = reinterpret_cast<const WalHeader*>(payload);
+    uint32_t cur_ts = header->timestamp;
+    if (cur_ts == 0) {
+      LOG(WARNING) << "Invalid timestamp 0, skip";
+      continue;
+    }
+    int length = header->length;
+    if (header->type) {
+      UpdateWalUnit unit;
+      unit.timestamp = cur_ts;
+      unit.ptr = const_cast<char*>(payload + sizeof(WalHeader));
+      unit.size = length;
+      update_wal_list_.push_back(unit);
+    } else {
+      if (insert_wal_list_[cur_ts].ptr) {
+        LOG(WARNING) << "Duplicated timestamp " << cur_ts << ", skip";
+      }
+      insert_wal_list_[cur_ts].ptr =
+          const_cast<char*>(payload + sizeof(WalHeader));
+      insert_wal_list_[cur_ts].size = length;
+    }
+    last_ts_ = std::max(cur_ts, last_ts_);
+  }
+
+  LOG(INFO) << "last_ts: " << last_ts_;
+  if (!update_wal_list_.empty()) {
+    std::sort(update_wal_list_.begin(), update_wal_list_.end(),
+              [](const UpdateWalUnit& lhs, const UpdateWalUnit& rhs) {
+                return lhs.timestamp < rhs.timestamp;
+              });
+  }
+}
+
+void KafkaWalParser::close() {
+  if (consumer_) {
+    consumer_.reset();
+  }
+  if (insert_wal_list_ != NULL) {
+    munmap(insert_wal_list_, insert_wal_list_size_ * sizeof(WalContentUnit));
+  }
+}
+
+uint32_t KafkaWalParser::last_ts() const { return last_ts_; }
+
+const WalContentUnit& KafkaWalParser::get_insert_wal(uint32_t ts) const {
+  if (insert_wal_list_[ts].ptr == NULL) {
+    LOG(WARNING) << "No WAL for timestamp " << ts;
+  }
+  return insert_wal_list_[ts];
+}
+const std::vector<UpdateWalUnit>& KafkaWalParser::get_update_wals() const {
+  return update_wal_list_;
+}
+
+}  // namespace gs

flex/engines/graph_db/database/wal/kafka_wal_parser.h

@@ -0,0 +1,67 @@
+/** 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_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
+#define ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
+
+#include <vector>
+#include "flex/engines/graph_db/database/wal/wal.h"
+
+#include "cppkafka/cppkafka.h"
+
+namespace gs {
+
+/*
+ * Get all partitions of the given topic.
+ */
+std::vector<cppkafka::TopicPartition> get_all_topic_partitions(
+    const cppkafka::Configuration& config, const std::string& topic_name);
+
+class KafkaWalParser : public IWalParser {
+ public:
+  static constexpr const std::chrono::milliseconds POLL_TIMEOUT =
+      std::chrono::milliseconds(100);
+  static constexpr const size_t MAX_BATCH_SIZE = 1000;
+
+  static std::unique_ptr<IWalParser> Make();
+
+  // always track all partitions and from begining
+  KafkaWalParser(const cppkafka::Configuration& config);
+  ~KafkaWalParser() { close(); }
+
+  void open(const std::string& topic_name) override;
+  void close() override;
+
+  uint32_t last_ts() const override;
+  const WalContentUnit& get_insert_wal(uint32_t ts) const override;
+  const std::vector<UpdateWalUnit>& get_update_wals() const override;
+
+  //////Kafka specific methods
+  void open(const std::vector<cppkafka::TopicPartition>& partitions);
+
+ private:
+  std::unique_ptr<cppkafka::Consumer> consumer_;
+  WalContentUnit* insert_wal_list_;
+  size_t insert_wal_list_size_;
+  uint32_t last_ts_;
+
+  std::vector<UpdateWalUnit> update_wal_list_;
+  std::vector<std::string> message_vector_;  // used to hold the polled messages
+  cppkafka::Configuration config_;
+};
+
+}  // namespace gs
+
+#endif  // ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_