feat: multi-group raft and leader-aware routing

CMakeLists.txt

@@ -1,174 +1,94 @@
-cmake_minimum_required(VERSION 3.16)
-project(sqlEngine VERSION 0.2.0 LANGUAGES CXX)
+cmake_minimum_required(VERSION 3.10)
+project(cloudSQL)
 
-# C++ Standard
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
-set(CMAKE_CXX_EXTENSIONS OFF)
 
-# Position Independent Code (required for TSan and good practice for libraries)
-set(CMAKE_POSITION_INDEPENDENT_CODE ON)
-
-# Build options
-option(BUILD_TESTS "Build tests" ON)
-option(BUILD_SHARED_LIBS "Build shared libraries" OFF)
-option(BUILD_COVERAGE "Build with coverage instrumentation" OFF)
-option(STRICT_LINT "Treat all warnings as errors" ON)
-
-set(USE_SANITIZER "address" CACHE STRING "Sanitizer to use: address, thread, undefined, or 'address,undefined'")
-
-# Configure Sanitizers
-if (NOT USE_SANITIZER STREQUAL "none")
-    if (CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU")
-        set(CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU")
-        string(REPLACE "," ";" SAN_LIST ${USE_SANITIZER})
-        foreach (SAN ${SAN_LIST})
-            list(APPEND SAN_FLAGS "-fsanitize=${SAN}")
-        endforeach()
-        list(APPEND SAN_FLAGS "-fno-omit-frame-pointer")
-
-        if (SAN_FLAGS)
-            add_compile_options(${SAN_FLAGS})
-            add_link_options(${SAN_FLAGS})
-            message(STATUS "Enabled sanitizers: ${USE_SANITIZER}")
-        endif()
-    endif()
-endif()
-
-# Enable compiler warnings
-if (CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
-    # Base warnings
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -Wpedantic -Wshadow -Woverloaded-virtual -Wold-style-cast -Wnon-virtual-dtor")
-
-    # Advanced safety warnings
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wconversion -Wsign-conversion -Wnull-dereference -Wdouble-promotion -Wformat=2")
-
-    if (STRICT_LINT)
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror")
-    endif()
-endif()
-
-if (MSVC)
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /W4")
-    if (STRICT_LINT)
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /WX")
-    endif()
-endif()
-
-# Clang-Tidy integration
-if (NOT DEFINED CMAKE_CXX_CLANG_TIDY)
-    find_program(CLANG_TIDY_BIN NAMES clang-tidy
-        PATHS /opt/homebrew/opt/llvm/bin /usr/local/opt/llvm/bin /usr/bin /usr/local/bin)
-
-    if (CLANG_TIDY_BIN)
-        message(STATUS "Found clang-tidy: ${CLANG_TIDY_BIN}")
-        set(CMAKE_CXX_CLANG_TIDY "${CLANG_TIDY_BIN}")
-    else()
-        message(WARNING "clang-tidy not found")
-    endif()
-endif()
-
-# Clang-Format target
-find_program(CLANG_FORMAT_BIN NAMES clang-format
-    PATHS /opt/homebrew/opt/llvm/bin /usr/local/opt/llvm/bin /usr/bin /usr/local/bin)
-
-if (CLANG_FORMAT_BIN)
-    file(GLOB_RECURSE ALL_CXX_FILES
-        "${CMAKE_CURRENT_SOURCE_DIR}/src/*.cpp"
-        "${CMAKE_CURRENT_SOURCE_DIR}/include/*.hpp"
-        "${CMAKE_CURRENT_SOURCE_DIR}/tests/*.cpp")
-    add_custom_target(format
-        COMMAND ${CLANG_FORMAT_BIN} -i ${ALL_CXX_FILES}
-        COMMENT "Formatting all source files with clang-format")
-    add_custom_target(check-format
-        COMMAND ${CLANG_FORMAT_BIN} --dry-run --Werror ${ALL_CXX_FILES}
-        COMMENT "Checking code formatting compliance")
+# Build Options
+option(STRICT_LINT "Enable strict linting" ON)
+option(ENABLE_ASAN "Enable AddressSanitizer" OFF)
+option(ENABLE_TSAN "Enable ThreadSanitizer" OFF)
+
+# Add include directories
+include_directories(include)
+
+# Find GoogleTest
+include(FetchContent)
+FetchContent_Declare(
+  googletest
+  URL https://github.com/google/googletest/archive/refs/heads/main.zip
+)
+set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
+FetchContent_MakeAvailable(googletest)
+
+# Core Library
+set(CORE_SOURCES
+    src/common/config.cpp
+    src/catalog/catalog.cpp
+    src/storage/storage_manager.cpp
+    src/storage/buffer_pool_manager.cpp
+    src/storage/lru_replacer.cpp
+    src/storage/heap_table.cpp
+    src/storage/btree_index.cpp
+    src/parser/lexer.cpp
+    src/parser/parser.cpp
+    src/parser/statement.cpp
+    src/parser/expression.cpp
+    src/executor/operator.cpp
+    src/executor/query_executor.cpp
+    src/network/rpc_client.cpp
+    src/network/rpc_server.cpp
+    src/network/server.cpp
+    src/transaction/lock_manager.cpp
+    src/transaction/transaction_manager.cpp
+    src/recovery/log_manager.cpp
+    src/recovery/recovery_manager.cpp
+    src/distributed/raft_group.cpp
+    src/distributed/raft_manager.cpp
+    src/distributed/distributed_executor.cpp
+)
+
+add_library(sqlEngineCore ${CORE_SOURCES})
+
+# Sanitizers
+if(ENABLE_ASAN)
+    target_compile_options(sqlEngineCore PUBLIC -fsanitize=address)
+    target_link_libraries(sqlEngineCore PUBLIC -fsanitize=address)
 endif()
 
-if (BUILD_COVERAGE)
-    if (CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU")
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --coverage")
-        set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} --coverage")
-        set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} --coverage")
-    endif()
+if(ENABLE_TSAN)
+    target_compile_options(sqlEngineCore PUBLIC -fsanitize=thread)
+    target_link_libraries(sqlEngineCore PUBLIC -fsanitize=thread)
 endif()
 
-# Source directories
-set(SRC_DIR "${CMAKE_CURRENT_SOURCE_DIR}/src")
-set(INCLUDE_DIR "${CMAKE_CURRENT_SOURCE_DIR}/include")
-
-# Collect all source files
-file(GLOB_RECURSE ALL_SOURCES "${SRC_DIR}/*.cpp")
-
-# Separate main.cpp from other sources
-set(CORE_SOURCES ${ALL_SOURCES})
-list(FILTER CORE_SOURCES EXCLUDE REGEX "src/main\\.cpp$")
-
-# Create core library
-add_library(sqlEngineCore STATIC ${CORE_SOURCES})
-target_include_directories(sqlEngineCore PUBLIC ${INCLUDE_DIR})
+# Main executable
+add_executable(cloudSQL src/main.cpp)
+target_link_libraries(cloudSQL sqlEngineCore)
 
-# Create main executable
-add_executable(sqlEngine src/main.cpp)
-target_link_libraries(sqlEngine PRIVATE sqlEngineCore)
+# Test helper macro
+macro(add_cloudsql_test NAME SOURCE)
+    add_executable(${NAME} ${SOURCE})
+    target_link_libraries(${NAME} sqlEngineCore GTest::gtest_main GTest::gmock)
+    add_test(NAME ${NAME} COMMAND ${NAME})
+endmacro()
 
-# Find and link required libraries
-find_package(Threads REQUIRED)
-target_link_libraries(sqlEngineCore PUBLIC Threads::Threads)
-
-# System libraries for networking
-if (UNIX AND NOT APPLE)
-    target_link_libraries(sqlEngineCore PUBLIC rt)
-endif()
-
-# Build test executables
-if (BUILD_TESTS)
+# Tests
+if(BUILD_TESTING)
     enable_testing()
-    include(FetchContent)
-    FetchContent_Declare(
-      googletest
-      URL https://github.com/google/googletest/archive/refs/heads/main.zip
-    )
-    # For Windows: prevent overriding the parent project's compiler/linker settings
-    set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
-    FetchContent_MakeAvailable(googletest)
-
-    macro(add_cloudsql_test name source)
-        add_executable(${name} ${source})
-        target_link_libraries(${name} PRIVATE sqlEngineCore GTest::gtest_main GTest::gmock_main)
-        add_test(NAME ${name} COMMAND ${name})
-    endmacro()
-
-    add_cloudsql_test(sqlEngine_tests tests/cloudSQL_tests.cpp)
-    add_cloudsql_test(lock_manager_tests tests/lock_manager_tests.cpp)
-    add_cloudsql_test(server_tests tests/server_tests.cpp)
-    add_cloudsql_test(transaction_manager_tests tests/transaction_manager_tests.cpp)
+    add_cloudsql_test(cloudSQL_tests tests/cloudSQL_tests.cpp)
     add_cloudsql_test(statement_tests tests/statement_tests.cpp)
+    add_cloudsql_test(transaction_manager_tests tests/transaction_manager_tests.cpp)
+    add_cloudsql_test(lock_manager_tests tests/lock_manager_tests.cpp)
     add_cloudsql_test(recovery_tests tests/recovery_tests.cpp)
     add_cloudsql_test(recovery_manager_tests tests/recovery_manager_tests.cpp)
     add_cloudsql_test(buffer_pool_tests tests/buffer_pool_tests.cpp)
     add_cloudsql_test(raft_tests tests/raft_tests.cpp)
     add_cloudsql_test(distributed_tests tests/distributed_tests.cpp)
     add_cloudsql_test(raft_sim_tests tests/raft_simulation_tests.cpp)
+    add_cloudsql_test(multi_raft_tests tests/multi_raft_tests.cpp)
     add_cloudsql_test(distributed_txn_tests tests/distributed_txn_tests.cpp)
 
     add_custom_target(run-tests
         COMMAND ${CMAKE_CTEST_COMMAND}
         COMMENT "Running all tests via CTest")
 endif()
-
-# Installation
-install(TARGETS sqlEngine RUNTIME DESTINATION bin)
-install(DIRECTORY ${INCLUDE_DIR} DESTINATION include)
-
-# Print configuration
-message(STATUS "=============================================")
-message(STATUS "cloudSQL Build Configuration")
-message(STATUS "=============================================")
-message(STATUS "  Version: ${PROJECT_VERSION}")
-message(STATUS "  C++ Standard: ${CMAKE_CXX_STANDARD}")
-message(STATUS "  Compiler: ${CMAKE_CXX_COMPILER_ID}")
-message(STATUS "  Build Tests: ${BUILD_TESTS}")
-message(STATUS "  Sanitizer: ${USE_SANITIZER}")
-message(STATUS "  Strict Lint: ${STRICT_LINT}")
-message(STATUS "=============================================")

include/catalog/catalog.hpp

@@ -18,6 +18,7 @@
 #include <vector>
 
 #include "common/value.hpp"
+#include "distributed/raft_types.hpp"
 
 namespace cloudsql {
 
@@ -145,8 +146,12 @@ struct DatabaseInfo {
 /**
  * @brief System Catalog class
  */
-class Catalog {
+class Catalog : public raft::RaftStateMachine {
    public:
+    /**
+     * @brief Apply a committed log entry (from RaftStateMachine)
+     */
+    void apply(const raft::LogEntry& entry) override;
     /**
      * @brief Default constructor
      */

include/common/cluster_manager.hpp

@@ -15,6 +15,10 @@
 #include "common/config.hpp"
 #include "executor/types.hpp"
 
+namespace cloudsql::raft {
+class RaftManager;
+}
+
 namespace cloudsql::cluster {
 
 /**
@@ -34,7 +38,7 @@ struct NodeInfo {
  */
 class ClusterManager {
    public:
-    explicit ClusterManager(const config::Config* config) : config_(config) {
+    explicit ClusterManager(const config::Config* config) : config_(config), raft_manager_(nullptr) {
         // Add self to node map if in distributed mode
         if (config_ != nullptr && config_->mode != config::RunMode::Standalone) {
             self_node_.id = "local_node";  // Will be replaced by unique ID later
@@ -54,6 +58,16 @@ class ClusterManager {
         nodes_[id] = {id, address, port, role, std::chrono::system_clock::now(), true};
     }
 
+    /**
+     * @brief Set Raft manager for this node
+     */
+    void set_raft_manager(raft::RaftManager* rm) { raft_manager_ = rm; }
+
+    /**
+     * @brief Get Raft manager for this node
+     */
+    [[nodiscard]] raft::RaftManager* get_raft_manager() const { return raft_manager_; }
+
     /**
      * @brief Update heartbeat for a node
      */
@@ -65,6 +79,26 @@ class ClusterManager {
         }
     }
 
+    /**
+     * @brief Update leader ID for a specific Raft group
+     */
+    void set_leader(uint16_t group_id, const std::string& leader_id) {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        group_leaders_[group_id] = leader_id;
+    }
+
+    /**
+     * @brief Get current leader for a Raft group
+     */
+    [[nodiscard]] std::string get_leader(uint16_t group_id) const {
+        const std::scoped_lock<std::mutex> lock(mutex_);
+        auto it = group_leaders_.find(group_id);
+        if (it != group_leaders_.end()) {
+            return it->second;
+        }
+        return "";
+    }
+
     /**
      * @brief Get list of active data nodes
      */
@@ -138,8 +172,10 @@ class ClusterManager {
 
    private:
     const config::Config* config_;
+    raft::RaftManager* raft_manager_;
     NodeInfo self_node_;
     std::unordered_map<std::string, NodeInfo> nodes_;
+    std::unordered_map<uint16_t, std::string> group_leaders_;
     /* context_id -> table_name -> rows */
     std::unordered_map<std::string, std::unordered_map<std::string, std::vector<executor::Tuple>>>
         shuffle_buffers_;

include/distributed/raft_group.hpp

@@ -39,14 +39,19 @@ class RaftGroup {
     void start();
     void stop();
 
+    /**
+     * @brief Set the state machine to apply committed entries to
+     */
+    void set_state_machine(RaftStateMachine* state_machine) { state_machine_ = state_machine; }
+
     // Raft RPC Handlers (called by RaftManager)
     void handle_request_vote(const network::RpcHeader& header, const std::vector<uint8_t>& payload,
                              int client_fd);
     void handle_append_entries(const network::RpcHeader& header,
                                const std::vector<uint8_t>& payload, int client_fd);
 
     // Client interface
-    bool replicate(const std::string& command);
+    bool replicate(const std::vector<uint8_t>& data);
     [[nodiscard]] bool is_leader() const { return state_.load() == NodeState::Leader; }
     [[nodiscard]] uint16_t group_id() const { return group_id_; }
 
@@ -67,6 +72,7 @@ class RaftGroup {
     std::string node_id_;
     cluster::ClusterManager& cluster_manager_;
     network::RpcServer& rpc_server_;
+    RaftStateMachine* state_machine_ = nullptr;
 
     // State
     std::atomic<NodeState> state_{NodeState::Follower};

include/distributed/raft_types.hpp

@@ -30,7 +30,20 @@ enum class NodeState : uint8_t { Follower, Candidate, Leader, Shutdown };
 struct LogEntry {
     term_t term = 0;
     index_t index = 0;
-    std::string data;  // Serialized command (e.g., DDL SQL)
+    std::vector<uint8_t> data;  // Binary payload
+};
+
+/**
+ * @brief Interface for the state machine that Raft replicates to
+ */
+class RaftStateMachine {
+   public:
+    virtual ~RaftStateMachine() = default;
+
+    /**
+     * @brief Apply a committed log entry to the state machine
+     */
+    virtual void apply(const LogEntry& entry) = 0;
 };
 
 /**