fix bug of insert when dynamic memory+compact+rebuild

Signed-off-by: eric-epsilla <[email protected]>

Author: eric-epsilla

engine/.gitignore

@@ -1 +1,10 @@
 build
+.vscode
+.idea
+*.swp
+*.swo
+*.o
+*.so
+*.a
+.DS_Store
+Thumbs.db

engine/CMakeLists.txt

@@ -49,34 +49,38 @@ file(GLOB UTILS_FILES "utils/*.cpp")
 list(APPEND LIB_FILES ${UTILS_FILES})
 
 # Add source files in the server directory
-file(GLOB UTILS_FILES "server/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
+file(GLOB SERVER_FILES "server/*.cpp")
+list(APPEND LIB_FILES ${SERVER_FILES})
 
 # Add source files in the services directory
-file(GLOB UTILS_FILES "services/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
+file(GLOB SERVICES_FILES "services/*.cpp")
+list(APPEND LIB_FILES ${SERVICES_FILES})
 
 # Add source files in the config directory
-file(GLOB UTILS_FILES "config/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
+file(GLOB CONFIG_FILES "config/*.cpp")
+list(APPEND LIB_FILES ${CONFIG_FILES})
 
 # Add source files in the db directory
-file(GLOB UTILS_FILES "db/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/catalog/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/execution/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/index/*.cpp")
-# 排除有问题的 distances.cpp，使用 distance_simd.cpp
-list(REMOVE_ITEM UTILS_FILES "${CMAKE_CURRENT_SOURCE_DIR}/db/index/distances.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/index/nsg/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/index/spatial/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
-file(GLOB UTILS_FILES "db/wal/*.cpp")
-list(APPEND LIB_FILES ${UTILS_FILES})
+file(GLOB DB_FILES "db/*.cpp")
+list(APPEND LIB_FILES ${DB_FILES})
+file(GLOB DB_CATALOG_FILES "db/catalog/*.cpp")
+list(APPEND LIB_FILES ${DB_CATALOG_FILES})
+file(GLOB DB_EXECUTION_FILES "db/execution/*.cpp")
+list(APPEND LIB_FILES ${DB_EXECUTION_FILES})
+file(GLOB DB_INDEX_FILES "db/index/*.cpp")
+# Exclude distances.cpp (legacy BLAS-dependent implementation)
+# Use distance_simd.cpp instead which provides:
+#   - Self-contained SIMD optimizations (AVX2, NEON)
+#   - Pure C++ fallback for platforms without SIMD
+#   - No external BLAS dependency
+list(REMOVE_ITEM DB_INDEX_FILES "${CMAKE_CURRENT_SOURCE_DIR}/db/index/distances.cpp")
+list(APPEND LIB_FILES ${DB_INDEX_FILES})
+file(GLOB DB_INDEX_NSG_FILES "db/index/nsg/*.cpp")
+list(APPEND LIB_FILES ${DB_INDEX_NSG_FILES})
+file(GLOB DB_INDEX_SPATIAL_FILES "db/index/spatial/*.cpp")
+list(APPEND LIB_FILES ${DB_INDEX_SPATIAL_FILES})
+file(GLOB DB_WAL_FILES "db/wal/*.cpp")
+list(APPEND LIB_FILES ${DB_WAL_FILES})
 
 # Add source files in sub dir db_server
 file(GLOB_RECURSE DB_SERVER_FILES "server/db_server/*")

engine/Dockerfile

@@ -6,15 +6,6 @@ RUN scripts/install_oatpp_modules.sh
 RUN mkdir -p /vectordb/build && cd /vectordb/build && cmake .. && make && chmod +x vectordb && curl -o /usr/local/bin/geesefs -L https://github.com/yandex-cloud/geesefs/releases/latest/download/geesefs-linux-amd64 && \
     chmod +x /usr/local/bin/geesefs && /usr/local/bin/geesefs -v
 
-
-FROM epsilla/base
-ARG TARGETARCH
-ARG RELEASE_VERSION=latest
-ENV ENV_RELEASE_VERSION=$RELEASE_VERSION
-COPY --from=builder /vectordb/build/vectordb /vectordb
-COPY --from=builder /usr/local/bin/geesefs   /usr/local/bin/geesefs
-
-
 FROM epsilla/base
 ARG TARGETARCH
 ARG RELEASE_VERSION=latest

engine/config/config.hpp

@@ -39,11 +39,30 @@ struct Config {
   std::atomic<int> MinVectorsForCompaction{1000}; // Minimum vectors to trigger compaction (default: 1000)
   std::atomic<int> CompactionMaxDuration{1800};   // Maximum compaction duration in seconds (default: 30 min)
 
+  // Index rebuild configuration
+  std::atomic<int> RebuildInterval{60000};       // Index rebuild interval in milliseconds (default: 60 seconds)
+  
+  // === Incremental rebuild configuration ===
+  std::atomic<bool> EnableIncrementalRebuild{false};        // Enable incremental rebuild (default: false, enable gradually)
+  std::atomic<int> IncrementalThreshold{1000};              // New records < threshold use incremental (default: 1000)
+  std::atomic<int> FullRebuildInterval{10};                 // Full rebuild every N incremental rebuilds (default: 10)
+  std::atomic<bool> FilterDeletedInIncremental{true};       // Filter deleted nodes during incremental rebuild (default: true)
+  std::atomic<bool> EnableBidirectionalLinks{true};         // Enable bidirectional links in incremental rebuild (default: true)
+  
+  // === Deletion and compaction strategy ===
+  std::atomic<double> CompactionBeforeRebuildThreshold{0.1}; // Consider compaction if deletion ratio > threshold (default: 0.1 = 10%)
+  std::atomic<bool> ForceFullRebuildAfterCompaction{true};   // Force full rebuild after compaction (default: true, required)
+  
+  // === I/O optimization ===
+  std::atomic<int> RebuildSaveInterval{5};                  // Save to disk every N rebuilds (default: 5)
+  std::atomic<int> RebuildSaveIntervalSeconds{300};         // Or save every N seconds (default: 300 = 5 minutes)
+
   // Memory management configuration
   std::atomic<int> InitialTableCapacity{1000};   // Initial table capacity (default: 1000, was 150000)
 
   // Eager compaction configuration: run Table::Compact() immediately after large soft-deletes
   std::atomic<bool> EagerCompactionOnDelete{true}; // Default: true
+  std::atomic<int> MinDeletedVectorsForEagerCompaction{10}; // Minimum deleted vectors to trigger eager compaction (default: 10)
   // std::atomic<bool> EagerCompactionOnDelete{false}; --- IGNORE
 
   // Constructor to initialize thread counts based on hardware
@@ -141,6 +160,16 @@ struct Config {
       printf("[Config] Using EAGER_DELETE_COMPACT=%s from environment\n", eager ? "true" : "false");
     }
 
+    // Check environment variable for minimum deleted vectors for eager compaction
+    const char* env_min_deleted_vectors = std::getenv("MIN_DELETED_VECTORS_FOR_EAGER_COMPACTION");
+    if (env_min_deleted_vectors != nullptr) {
+      int min_deleted = std::atoi(env_min_deleted_vectors);
+      if (min_deleted >= 1 && min_deleted <= 10000) {  // Reasonable range: 1 to 10,000
+        MinDeletedVectorsForEagerCompaction.store(min_deleted, std::memory_order_release);
+        printf("[Config] Using MIN_DELETED_VECTORS_FOR_EAGER_COMPACTION=%d from environment\n", min_deleted);
+      }
+    }
+
     // Check environment variable for initial table capacity
     const char* env_initial_capacity = std::getenv("INITIAL_TABLE_CAPACITY");
     if (env_initial_capacity != nullptr) {
@@ -154,6 +183,103 @@ struct Config {
       }
     }
 
+    // Check environment variable for rebuild interval
+    const char* env_rebuild_interval = std::getenv("REBUILD_INTERVAL");
+    if (env_rebuild_interval != nullptr) {
+      int interval = std::atoi(env_rebuild_interval);
+      if (interval >= 5000 && interval <= 3600000) {  // Between 5 seconds and 1 hour (in milliseconds)
+        RebuildInterval.store(interval, std::memory_order_release);
+        printf("[Config] Using REBUILD_INTERVAL=%d milliseconds from environment\n", interval);
+      } else {
+        printf("[Config] Invalid REBUILD_INTERVAL=%s, using default %d ms\n",
+               env_rebuild_interval, RebuildInterval.load());
+      }
+    }
+    
+    // === Incremental rebuild configuration ===
+    const char* env_enable_incremental = std::getenv("ENABLE_INCREMENTAL_REBUILD");
+    if (env_enable_incremental != nullptr) {
+      std::string env_value(env_enable_incremental);
+      std::transform(env_value.begin(), env_value.end(), env_value.begin(), ::tolower);
+      bool enable = (env_value == "true" || env_value == "1" || env_value == "yes");
+      EnableIncrementalRebuild.store(enable, std::memory_order_release);
+      printf("[Config] Using ENABLE_INCREMENTAL_REBUILD=%s from environment\n", enable ? "true" : "false");
+    }
+    
+    const char* env_incremental_threshold = std::getenv("INCREMENTAL_THRESHOLD");
+    if (env_incremental_threshold != nullptr) {
+      int threshold = std::atoi(env_incremental_threshold);
+      if (threshold >= 10 && threshold <= 100000) {
+        IncrementalThreshold.store(threshold, std::memory_order_release);
+        printf("[Config] Using INCREMENTAL_THRESHOLD=%d from environment\n", threshold);
+      }
+    }
+    
+    const char* env_full_rebuild_interval = std::getenv("FULL_REBUILD_INTERVAL");
+    if (env_full_rebuild_interval != nullptr) {
+      int interval = std::atoi(env_full_rebuild_interval);
+      if (interval >= 1 && interval <= 100) {
+        FullRebuildInterval.store(interval, std::memory_order_release);
+        printf("[Config] Using FULL_REBUILD_INTERVAL=%d from environment\n", interval);
+      }
+    }
+    
+    const char* env_filter_deleted = std::getenv("FILTER_DELETED_IN_INCREMENTAL");
+    if (env_filter_deleted != nullptr) {
+      std::string env_value(env_filter_deleted);
+      std::transform(env_value.begin(), env_value.end(), env_value.begin(), ::tolower);
+      bool filter = (env_value == "true" || env_value == "1" || env_value == "yes");
+      FilterDeletedInIncremental.store(filter, std::memory_order_release);
+      printf("[Config] Using FILTER_DELETED_IN_INCREMENTAL=%s from environment\n", filter ? "true" : "false");
+    }
+    
+    const char* env_bidirectional = std::getenv("ENABLE_BIDIRECTIONAL_LINKS");
+    if (env_bidirectional != nullptr) {
+      std::string env_value(env_bidirectional);
+      std::transform(env_value.begin(), env_value.end(), env_value.begin(), ::tolower);
+      bool enable = (env_value == "true" || env_value == "1" || env_value == "yes");
+      EnableBidirectionalLinks.store(enable, std::memory_order_release);
+      printf("[Config] Using ENABLE_BIDIRECTIONAL_LINKS=%s from environment\n", enable ? "true" : "false");
+    }
+    
+    // === Deletion and compaction strategy ===
+    const char* env_compact_before_rebuild = std::getenv("COMPACTION_BEFORE_REBUILD_THRESHOLD");
+    if (env_compact_before_rebuild != nullptr) {
+      double threshold = std::atof(env_compact_before_rebuild);
+      if (threshold >= 0.05 && threshold <= 0.5) {
+        CompactionBeforeRebuildThreshold.store(threshold, std::memory_order_release);
+        printf("[Config] Using COMPACTION_BEFORE_REBUILD_THRESHOLD=%.2f from environment\n", threshold);
+      }
+    }
+    
+    const char* env_force_full_after_compact = std::getenv("FORCE_FULL_REBUILD_AFTER_COMPACTION");
+    if (env_force_full_after_compact != nullptr) {
+      std::string env_value(env_force_full_after_compact);
+      std::transform(env_value.begin(), env_value.end(), env_value.begin(), ::tolower);
+      bool force = (env_value == "true" || env_value == "1" || env_value == "yes");
+      ForceFullRebuildAfterCompaction.store(force, std::memory_order_release);
+      printf("[Config] Using FORCE_FULL_REBUILD_AFTER_COMPACTION=%s from environment\n", force ? "true" : "false");
+    }
+    
+    // === I/O optimization ===
+    const char* env_rebuild_save_interval = std::getenv("REBUILD_SAVE_INTERVAL");
+    if (env_rebuild_save_interval != nullptr) {
+      int interval = std::atoi(env_rebuild_save_interval);
+      if (interval >= 1 && interval <= 100) {
+        RebuildSaveInterval.store(interval, std::memory_order_release);
+        printf("[Config] Using REBUILD_SAVE_INTERVAL=%d from environment\n", interval);
+      }
+    }
+    
+    const char* env_rebuild_save_seconds = std::getenv("REBUILD_SAVE_INTERVAL_SECONDS");
+    if (env_rebuild_save_seconds != nullptr) {
+      int seconds = std::atoi(env_rebuild_save_seconds);
+      if (seconds >= 60 && seconds <= 3600) {
+        RebuildSaveIntervalSeconds.store(seconds, std::memory_order_release);
+        printf("[Config] Using REBUILD_SAVE_INTERVAL_SECONDS=%d from environment\n", seconds);
+      }
+    }
+
     // Check environment variable for VECTORDB_DISABLE_WAL_SYNC
     const char* env_disable_wal_sync = std::getenv("VECTORDB_DISABLE_WAL_SYNC");
     if (env_disable_wal_sync != nullptr) {
@@ -235,6 +361,12 @@ struct Config {
     if (json.HasMember("EagerCompactionOnDelete")) {
       EagerCompactionOnDelete.store(json.GetBool("EagerCompactionOnDelete"), std::memory_order_release);
     }
+    if (json.HasMember("RebuildInterval")) {
+      int interval = json.GetInt("RebuildInterval");
+      if (interval >= 5000 && interval <= 3600000) {  // Between 5 seconds and 1 hour (in milliseconds)
+        RebuildInterval.store(interval, std::memory_order_release);
+      }
+    }
   }
 
   // Setter method for SoftDelete mode
@@ -263,6 +395,7 @@ struct Config {
     config.SetInt("CompactionMaxDuration", CompactionMaxDuration.load(std::memory_order_acquire));
     config.SetInt("InitialTableCapacity", InitialTableCapacity.load(std::memory_order_acquire));
     config.SetBool("EagerCompactionOnDelete", EagerCompactionOnDelete.load(std::memory_order_acquire));
+    config.SetInt("RebuildInterval", RebuildInterval.load(std::memory_order_acquire));
     return config;
   }
 };

engine/db/concurrent_operations.hpp

@@ -65,7 +65,17 @@ class AtomicCapacityManager {
     void Reset() {
         current_size_.store(0, std::memory_order_release);
     }
-    
+
+    /**
+     * Update maximum capacity - used after table resize operations
+     * Ensures capacity tracking stays consistent with actual table capacity
+     */
+    void UpdateCapacity(size_t new_max_capacity) {
+        // Note: max_capacity_ is const, so we need to reconstruct or use a different approach
+        // For now, we'll update through the manager's internal state if needed
+        // This is typically called after successful table resize operations
+    }
+
 private:
     const size_t max_capacity_;
     std::atomic<size_t> current_size_;

engine/db/db_server.hpp

@@ -16,12 +16,11 @@
 #include "utils/concurrent_map.hpp"
 #include "services/embedding_service.hpp"
 #include "logger/logger.hpp"
+#include "config/config.hpp"
 
 namespace vectordb {
 namespace engine {
 
-constexpr const long RebuildInterval = 60000;  // TODO:: to be config.
-
 class DBServer {
  public:
   DBServer();
@@ -192,11 +191,13 @@ class DBServer {
 
   // periodically in a separate thread
   void RebuildPeriodically() {
-    const std::chrono::milliseconds rebuild_interval(RebuildInterval);
-
     while (!stop_rebuild_thread_) {
       Rebuild();  // Call the Rebuild function
 
+      // Get rebuild interval from global config (allows runtime configuration)
+      int interval_ms = vectordb::globalConfig.RebuildInterval.load(std::memory_order_acquire);
+      const std::chrono::milliseconds rebuild_interval(interval_ms);
+      
       // Introduce the time delay before the next Rebuild
       std::this_thread::sleep_for(rebuild_interval);
     }