background processes simulator application for re-building vector indexes

CMakeLists.txt

@@ -190,7 +190,45 @@ if(BUILD_EXAMPLES)
     BUILD_RPATH ${RUST_TARGET_DIR}
   )
 
-  add_custom_target(examples DEPENDS simple full s3)
+  # s3vector_concurrent_service - concurrent vector service with background indexing
+  find_package(nlohmann_json QUIET)
+  if(NOT nlohmann_json_FOUND)
+    message(STATUS "nlohmann_json not found, fetching from GitHub...")
+    include(FetchContent)
+    FetchContent_Declare(
+      nlohmann_json
+      GIT_REPOSITORY https://github.com/nlohmann/json.git
+      GIT_TAG v3.11.3
+      GIT_SHALLOW TRUE
+    )
+    FetchContent_MakeAvailable(nlohmann_json)
+  endif()
+  find_package(CURL REQUIRED)
+  find_package(OpenSSL REQUIRED)
+  add_executable(s3vector_concurrent_service
+    examples/s3vector_concurrent_service.cpp
+    examples/s3_lock.cpp
+    examples/s3_http.cpp
+  )
+  target_link_libraries(s3vector_concurrent_service
+    lancedb
+    Threads::Threads
+    ${ARROW_LIBRARIES}
+    nlohmann_json::nlohmann_json
+    CURL::libcurl
+    OpenSSL::Crypto
+  )
+  target_include_directories(s3vector_concurrent_service
+    PRIVATE ${ARROW_INCLUDE_DIRS}
+    PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
+    PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/examples
+  )
+  target_compile_options(s3vector_concurrent_service PRIVATE ${ARROW_CFLAGS_OTHER})
+  set_target_properties(s3vector_concurrent_service PROPERTIES
+    BUILD_RPATH ${RUST_TARGET_DIR}
+  )
+
+  add_custom_target(examples DEPENDS simple full s3 s3vector_concurrent_service)
 endif()
 
 if(BUILD_DOCS)
@@ -351,6 +389,30 @@ if(BUILD_TESTS)
   )
   list(APPEND TEST_TARGETS lancedb_index_tests)
 
+  # Add test executable for scalar index detailed tests (runs with valgrind)
+  add_executable(lancedb_scalar_index_tests
+    tests/test_main.cpp
+    tests/test_common.cpp
+    tests/test_scalar_index.cpp
+  )
+  target_link_libraries(lancedb_scalar_index_tests
+    PRIVATE
+    lancedb
+    Catch2::Catch2
+    Threads::Threads
+    ${ARROW_LIBRARIES}
+  )
+  target_include_directories(lancedb_scalar_index_tests
+    PRIVATE ${ARROW_INCLUDE_DIRS}
+    PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
+    PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/tests
+  )
+  target_compile_options(lancedb_scalar_index_tests PRIVATE ${ARROW_CFLAGS_OTHER})
+  set_target_properties(lancedb_scalar_index_tests PROPERTIES
+    BUILD_RPATH ${RUST_TARGET_DIR}
+  )
+  list(APPEND TEST_TARGETS lancedb_scalar_index_tests)
+
   # Add test executable for vector tests (runs without valgrind)
   add_executable(lancedb_vector_index_tests
     tests/test_main.cpp
@@ -490,6 +552,19 @@ if(BUILD_TESTS)
         --suppressions=${CMAKE_CURRENT_SOURCE_DIR}/valgrind.supp
         $<TARGET_FILE:lancedb_index_tests>
     )
+    # Run scalar index detailed tests with valgrind
+    add_test(NAME lancedb_scalar_index_tests
+      COMMAND ${TEST_ENV_PREFIX} ${VALGRIND_EXECUTABLE}
+        --tool=memcheck
+        --leak-check=full
+        --show-leak-kinds=definite
+        --errors-for-leak-kinds=definite
+        --track-origins=yes
+        --error-exitcode=1
+        --log-file=${CMAKE_BINARY_DIR}/valgrind_scalar_index.txt
+        --suppressions=${CMAKE_CURRENT_SOURCE_DIR}/valgrind.supp
+        $<TARGET_FILE:lancedb_scalar_index_tests>
+    )
     # Run query tests with valgrind
     add_test(NAME lancedb_query_tests
       COMMAND ${TEST_ENV_PREFIX} ${VALGRIND_EXECUTABLE}
@@ -510,6 +585,7 @@ if(BUILD_TESTS)
     add_test(NAME lancedb_table_meta_tests COMMAND ${TEST_ENV_PREFIX} $<TARGET_FILE:lancedb_table_meta_tests>)
     add_test(NAME lancedb_index_tests COMMAND ${TEST_ENV_PREFIX} $<TARGET_FILE:lancedb_index_tests>)
     add_test(NAME lancedb_query_tests COMMAND ${TEST_ENV_PREFIX} $<TARGET_FILE:lancedb_query_tests>)
+    add_test(NAME lancedb_scalar_index_tests COMMAND ${TEST_ENV_PREFIX} $<TARGET_FILE:lancedb_scalar_index_tests>)
   endif()
 
   # Run vector index tests WITHOUT valgrind (too slow under valgrind)

examples/base_lock.h

@@ -0,0 +1,22 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Base_lock - Abstract interface for distributed/local locking mechanisms.
+ *
+ * Implementations:
+ *   - FileLock:  flock-based, for local/testing use (s3vector_concurrent_service.cpp)
+ *   - NoOpLock:  no-op, always succeeds (s3vector_concurrent_service.cpp)
+ *   - S3Lock:    S3 conditional-write based distributed lock (s3_lock.h / s3_lock.cpp)
+ */
+
+#pragma once
+
+class Base_lock {
+public:
+    virtual ~Base_lock() = default;
+    virtual bool lock_exclusive() = 0;
+    virtual bool try_lock_exclusive() = 0;
+    virtual void unlock() = 0;
+    virtual bool is_locked() const = 0;
+    virtual bool is_valid() const = 0;
+};

examples/s3_http.cpp

@@ -0,0 +1,236 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * S3 HTTP client implementation using libcurl + AWS SigV4.
+ *
+ * No dependency on the main application code.
+ * Dependencies: s3_http.h, s3_sigv4.h, libcurl, OpenSSL.
+ */
+
+#include "s3_http.h"
+#include "s3_sigv4.h"
+
+#include <string>
+
+
+// ============================================================================
+// Internal helpers
+// ============================================================================
+
+namespace {
+
+// Build URL and canonical URI for an object operation (path-style only for now)
+struct S3Url {
+    std::string url;
+    std::string canonical_uri;
+    std::string host;
+};
+
+S3Url build_object_url(const S3HttpConfig& config, const std::string& bucket,
+                       const std::string& key) {
+    S3Url result;
+    result.host = sigv4::extract_host(config.endpoint);
+
+    if (config.use_path_style) {
+        result.url = config.endpoint + "/" + bucket + "/" + key;
+        result.canonical_uri = "/" + bucket + "/" + key;
+    } else {
+        auto scheme_end = config.endpoint.find("://");
+        if (scheme_end != std::string::npos) {
+            result.url = config.endpoint.substr(0, scheme_end + 3) + bucket + "." +
+                         config.endpoint.substr(scheme_end + 3) + "/" + key;
+        }
+        result.canonical_uri = "/" + key;
+        result.host = bucket + "." + result.host;
+    }
+    return result;
+}
+
+S3Url build_bucket_url(const S3HttpConfig& config, const std::string& bucket) {
+    S3Url result;
+    result.host = sigv4::extract_host(config.endpoint);
+    result.url = config.endpoint + "/" + bucket + "/";
+    result.canonical_uri = "/" + bucket + "/";
+    return result;
+}
+
+// Execute a signed S3 request
+S3HttpResult execute_request(const std::string& method,
+                             const S3Url& s3url,
+                             const S3HttpConfig& config,
+                             const std::string& body = "",
+                             const std::string& content_type = "",
+                             bool if_none_match = false,
+                             bool head_only = false) {
+    S3HttpResult result;
+
+    CURL* curl = curl_easy_init();
+    if (!curl) return result;
+
+    std::string payload_hash = sigv4::sha256_hex(body);
+
+    // For conditional PUT (If-None-Match: *), we need to sign the extra header
+    std::string extra_headers_canonical;
+    std::string extra_signed_headers;
+    if (if_none_match) {
+        extra_headers_canonical = "if-none-match:*\n";
+        extra_signed_headers = "if-none-match";
+
+        // if-none-match sorts between host and x-amz-*, so we must build
+        // the canonical headers manually with correct sort order
+        auto ts = sigv4::get_timestamp();
+        std::string service = "s3";
+        std::string scope = ts.date_stamp + "/" + config.region + "/" + service + "/aws4_request";
+
+        std::string canonical_headers =
+            "host:" + s3url.host + "\n"
+            "if-none-match:*\n"
+            "x-amz-content-sha256:" + payload_hash + "\n"
+            "x-amz-date:" + ts.iso8601 + "\n";
+        std::string signed_headers = "host;if-none-match;x-amz-content-sha256;x-amz-date";
+
+        std::string canonical_request =
+            method + "\n" +
+            sigv4::uri_encode_path(s3url.canonical_uri) + "\n"
+            "\n" +
+            canonical_headers + "\n" +
+            signed_headers + "\n" +
+            payload_hash;
+
+        std::string string_to_sign =
+            "AWS4-HMAC-SHA256\n" +
+            ts.iso8601 + "\n" +
+            scope + "\n" +
+            sigv4::sha256_hex(canonical_request);
+
+        std::string date_key = sigv4::hmac_sha256_raw("AWS4" + config.secret_access_key, ts.date_stamp);
+        std::string date_region_key = sigv4::hmac_sha256_raw(date_key, config.region);
+        std::string date_region_service_key = sigv4::hmac_sha256_raw(date_region_key, service);
+        std::string signing_key = sigv4::hmac_sha256_raw(date_region_service_key, "aws4_request");
+        std::string signature = sigv4::hmac_sha256_hex(signing_key, string_to_sign);
+
+        std::string authorization = "AWS4-HMAC-SHA256 Credential=" + config.access_key_id + "/" + scope +
+                                   ", SignedHeaders=" + signed_headers +
+                                   ", Signature=" + signature;
+
+        struct curl_slist* headers = nullptr;
+        headers = curl_slist_append(headers, ("Host: " + s3url.host).c_str());
+        headers = curl_slist_append(headers, ("Authorization: " + authorization).c_str());
+        headers = curl_slist_append(headers, ("x-amz-date: " + ts.iso8601).c_str());
+        headers = curl_slist_append(headers, ("x-amz-content-sha256: " + payload_hash).c_str());
+        headers = curl_slist_append(headers, "If-None-Match: *");
+        if (!content_type.empty()) {
+            headers = curl_slist_append(headers, ("Content-Type: " + content_type).c_str());
+        }
+
+        curl_easy_setopt(curl, CURLOPT_URL, s3url.url.c_str());
+        curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method.c_str());
+        if (!body.empty()) {
+            curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str());
+            curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
+        }
+        curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
+        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, sigv4::curl_write_callback);
+        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &result.body);
+        curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L);
+
+        CURLcode res = curl_easy_perform(curl);
+        curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &result.http_code);
+        curl_slist_free_all(headers);
+        curl_easy_cleanup(curl);
+
+        result.ok = (res == CURLE_OK && result.http_code >= 200 && result.http_code < 300);
+        return result;
+    }
+
+    // Standard request (no If-None-Match)
+    auto signed_req = sigv4::sign_request(
+        method, s3url.canonical_uri, s3url.host, payload_hash,
+        config.region, config.access_key_id, config.secret_access_key);
+
+    struct curl_slist* headers = nullptr;
+    headers = curl_slist_append(headers, ("Host: " + s3url.host).c_str());
+    headers = curl_slist_append(headers, ("Authorization: " + signed_req.authorization).c_str());
+    headers = curl_slist_append(headers, ("x-amz-date: " + signed_req.x_amz_date).c_str());
+    headers = curl_slist_append(headers, ("x-amz-content-sha256: " + payload_hash).c_str());
+    if (!content_type.empty()) {
+        headers = curl_slist_append(headers, ("Content-Type: " + content_type).c_str());
+    }
+    if (body.empty() && method == "PUT") {
+        headers = curl_slist_append(headers, "Content-Length: 0");
+    }
+
+    curl_easy_setopt(curl, CURLOPT_URL, s3url.url.c_str());
+    curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method.c_str());
+    if (head_only) {
+        curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
+    }
+    if (!body.empty()) {
+        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str());
+        curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
+    }
+    curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
+    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, sigv4::curl_write_callback);
+    curl_easy_setopt(curl, CURLOPT_WRITEDATA, &result.body);
+    curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L);
+
+    CURLcode res = curl_easy_perform(curl);
+    curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &result.http_code);
+
+    curl_slist_free_all(headers);
+    curl_easy_cleanup(curl);
+
+    result.ok = (res == CURLE_OK && result.http_code >= 200 && result.http_code < 300);
+    return result;
+}
+
+} // anonymous namespace
+
+// ============================================================================
+// Object operations
+// ============================================================================
+
+S3HttpResult s3_put_object(const S3HttpConfig& config,
+                           const std::string& bucket, const std::string& key,
+                           const std::string& body,
+                           const std::string& content_type,
+                           bool if_none_match) {
+    auto s3url = build_object_url(config, bucket, key);
+    return execute_request("PUT", s3url, config, body, content_type, if_none_match);
+}
+
+S3HttpResult s3_get_object(const S3HttpConfig& config,
+                           const std::string& bucket, const std::string& key) {
+    auto s3url = build_object_url(config, bucket, key);
+    return execute_request("GET", s3url, config);
+}
+
+S3HttpResult s3_delete_object(const S3HttpConfig& config,
+                              const std::string& bucket, const std::string& key) {
+    auto s3url = build_object_url(config, bucket, key);
+    auto result = execute_request("DELETE", s3url, config);
+    // 204 and 404 are both acceptable for delete
+    if (result.http_code == 204 || result.http_code == 404) {
+        result.ok = true;
+    }
+    return result;
+}
+
+// ============================================================================
+// Bucket operations
+// ============================================================================
+
+S3HttpResult s3_create_bucket(const S3HttpConfig& config, const std::string& bucket) {
+    auto s3url = build_bucket_url(config, bucket);
+    auto result = execute_request("PUT", s3url, config);
+    // 409 = BucketAlreadyOwnedByYou — treat as success
+    if (result.http_code == 409) {
+        result.ok = true;
+    }
+    return result;
+}
+
+S3HttpResult s3_head_bucket(const S3HttpConfig& config, const std::string& bucket) {
+    auto s3url = build_bucket_url(config, bucket);
+    return execute_request("HEAD", s3url, config, "", "", false, true);
+}