feat: [NCS] Add Near-Compute Storage (NCS) core infrastructure

CMakeLists.txt

@@ -14,6 +14,7 @@ project(milvus_common CXX C)
 
 option(ENABLE_UNIT_TESTS "Enable unit tests" OFF)
 option(ENABLE_SYNCPOINT "Enable sync point for testing" OFF)
+option(USE_REDIS "Enable Redis NCS support" ON)
 
 set( CMAKE_CXX_STANDARD 17 )
 set( CMAKE_CXX_STANDARD_REQUIRED on )
@@ -25,6 +26,7 @@ set( CMAKE_EXPORT_COMPILE_COMMANDS ON )
 set(MILVUS_COMMON_WORKSPACE ${CMAKE_CURRENT_SOURCE_DIR})
 include_directories(${MILVUS_COMMON_WORKSPACE}/include)
 
+
 set(CMAKE_CXX_FLAGS "-Wall -fPIC ${CMAKE_CXX_FLAGS}")
 
 if (WITH_ASAN)
@@ -41,19 +43,28 @@ find_package(glog REQUIRED)
 find_package(fmt REQUIRED)
 find_package(prometheus-cpp REQUIRED)
 
+if(USE_REDIS)
+    find_package(hiredis REQUIRED)
+endif()
+
 list(APPEND COMMON_LINKER_LIBS glog::glog)
 list(APPEND COMMON_LINKER_LIBS prometheus-cpp::core prometheus-cpp::push)
 list(APPEND COMMON_LINKER_LIBS fmt::fmt-header-only)
 list(APPEND COMMON_LINKER_LIBS Folly::folly)
 list(APPEND COMMON_LINKER_LIBS gflags::gflags)
 
+if(USE_REDIS)
+    list(APPEND COMMON_LINKER_LIBS hiredis::hiredis)
+endif()
+
 list(APPEND COMMON_LINKER_LIBS opentelemetry-cpp::opentelemetry_trace)
 list(APPEND COMMON_LINKER_LIBS opentelemetry-cpp::opentelemetry_exporter_ostream_span)
 list(APPEND COMMON_LINKER_LIBS opentelemetry-cpp::opentelemetry_exporter_otlp_grpc)
 list(APPEND COMMON_LINKER_LIBS opentelemetry-cpp::opentelemetry_exporter_otlp_http)
 list(APPEND COMMON_LINKER_LIBS opentelemetry-cpp::opentelemetry_exporter_jaeger_trace)
 
 file(GLOB_RECURSE SRC_FILES src/*.cpp src/*.cc)
+
 if(__X86_64)
    set_source_files_properties(src/knowhere/thread_pool.cc PROPERTIES
      COMPILE_OPTIONS "-msse4.2"
@@ -70,12 +81,27 @@ else()
     list(APPEND COMMON_LINKER_LIBS ${LIBAIO_LIBRARY})
 endif()
 
+# Conditionally exclude Redis source files if USE_REDIS is OFF
+if(NOT USE_REDIS)
+    list(REMOVE_ITEM SRC_FILES 
+        ${CMAKE_CURRENT_SOURCE_DIR}/src/ncs/RedisNcs.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/src/ncs/RedisNcsConnector.cpp
+    )
+    message(STATUS "Redis support disabled, excluding Redis NCS files from build")
+else()
+    message(STATUS "Redis support enabled, including Redis NCS files in build")
+endif()
+
 add_library(milvus-common SHARED ${SRC_FILES})
 
 if (ENABLE_SYNCPOINT)
     add_definitions(-DENABLE_SYNCPOINT)
 endif()
 
+if (USE_REDIS)
+    target_compile_definitions(milvus-common PUBLIC USE_REDIS)
+endif()
+
 target_link_libraries(milvus-common PUBLIC
     ${COMMON_LINKER_LIBS}
 )

Dockerfile.builder

@@ -0,0 +1,50 @@
+# Builder image for milvus-common unit tests (based on ubuntu-22.04)
+# Usage examples:
+#  docker build -f Dockerfile.builder -t milvus-common-builder:latest .
+#  docker run --rm -it -v$(pwd):/workspace -w /workspace milvus-common-builder:latest  # get an interactive shell
+
+FROM ubuntu:22.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV LANG=C.UTF-8
+ENV LC_ALL=C.UTF-8
+
+# Install toolchain + dependencies
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends \
+       software-properties-common \
+       ca-certificates \
+       curl \
+       gnupg2 \
+       cmake \
+       libopenblas-dev \
+       libaio-dev \
+       python3 \
+       python3-pip \
+       build-essential \
+       libpci3 \
+       redis-server \
+ && add-apt-repository ppa:ubuntu-toolchain-r/test -y \
+ && apt-get update \
+ && apt-get install -y --no-install-recommends gcc-12 g++-12 \
+ && rm -rf /var/lib/apt/lists/*
+
+# Make gcc-12 / g++-12 available via CC/CXX env vars
+ENV CC=gcc-12
+ENV CXX=g++-12
+
+# Install a specific conan
+RUN pip3 install --no-cache-dir conan==1.61.0
+
+# Add default conan remote if reachable (best-effort)
+RUN conan remote add default-conan-local https://milvus01.jfrog.io/artifactory/api/conan/default-conan-local || true
+
+WORKDIR /workspace
+
+# Create entrypoint script that starts Redis and then runs bash
+RUN echo '#!/bin/bash\nredis-server --daemonize yes\nexec "$@"' > /entrypoint.sh \
+ && chmod +x /entrypoint.sh
+
+# Default entrypoint starts Redis, then runs bash for interactive use
+ENTRYPOINT ["/entrypoint.sh"]
+CMD ["/bin/bash"]

README.md

@@ -25,3 +25,27 @@ conan build ..
 # run ut
 ./test/test_cachinglayer/cachinglayer_test
 ```
+
+## Build using the provided Docker builder image (alternative)
+
+If you don't want to install the toolchain and conan locally, you can use the included
+`Dockerfile.builder` image which mirrors the CI environment (Ubuntu 22.04, gcc-12,
+conan 1.61). 
+
+```bash
+# Build the builder image (run from repository root)
+docker build -f Dockerfile.builder -t milvus-common-builder:latest .
+
+# Start an interactive shell with the repository mounted at /workspace
+docker run --rm -it   -v "$(pwd)":/workspace   -v "${HOME}/.conan":/root/.conan --add-host=host.docker.internal:host-gateway   -w /workspace   milvus-common-builder:latest
+
+# Inside the container run the same build commands as CI:
+mkdir -p build && cd build
+conan install .. --build=missing -o with_ut=True -o with_asan=True -s compiler.libcxx=libstdc++11 -s compiler.version=12 -s build_type=Release
+conan build ..
+
+# Run tests inside the container (example)
+./test/test_cachinglayer/cachinglayer_test
+```
+
+

conanfile.py

@@ -26,7 +26,8 @@ class MilvusCommonConan(ConanFile):
         "libevent/2.1.12#4fd19d10d3bed63b3a8952c923454bc0",
         "openssl/3.1.2#02594c4c0a6e2b4feb3cd15119993597",
         "folly/2023.10.30.10@milvus/dev",
-        "boost/1.82.0"
+        "boost/1.82.0",
+        "hiredis/1.2.0"
     )
 
     options = {

include/common/EasyAssert.h

@@ -71,7 +71,8 @@ enum ErrorCode {
     TextIndexNotFound = 2041,
     InvalidParameter = 2042,
     InsufficientResource = 2043,
-
+    NcsUploadError = 2044,
+
     KnowhereError = 2099
 };
 

include/ncs/InMemNcsConnector.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "ncs/ncs.h"
+#include "ncs/InMemoryKV.h"
+#include <memory>
+
+namespace milvus {
+
+class InMemNcsConnector : public NcsConnector {
+public:
+    friend class InMemoryNcsConnectorCreator;
+
+    // Interface implementations
+    std::vector<NcsStatus> multiGet(const std::vector<uint32_t>& keys, const std::vector<boost::span<uint8_t>>& buffs) override;
+    std::vector<NcsStatus> multiPut(const std::vector<uint32_t>& keys, const std::vector<boost::span<uint8_t>>& buffs) override;
+    std::vector<NcsStatus> multiDelete(const std::vector<uint32_t>& keys) override;
+
+private:
+    explicit InMemNcsConnector(uint64_t bucketId);  // Private constructor
+};
+
+class InMemoryNcsConnectorCreator : public NcsConnectorCreator {
+public:
+    NcsConnector* factoryMethod(const NcsDescriptor* descriptor) override;
+    const std::string& getKind() const override;
+
+private:
+    static const std::string KIND;
+};
+
+} // namespace milvus

include/ncs/InMemoryKV.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <unordered_map>
+#include <vector>
+#include <cstdint>
+#include <unistd.h>
+
+#include <boost/core/span.hpp>
+
+namespace milvus {
+
+class InMemoryKV {
+public:
+    static InMemoryKV* Instance();
+
+    // Put a value into an existing bucket. Returns true on success, false if the
+    // bucket does not exist.
+    bool put(uint64_t bucketId, uint32_t key, const boost::span<uint8_t>& value);
+    bool createBucket(uint64_t bucketId);
+    bool deleteBucket(uint64_t bucketId);
+    bool hasBucket(uint64_t bucketId) const;
+    bool get(uint64_t bucketId, uint32_t key, const boost::span<uint8_t>& buff) const;
+    bool deleteKey(uint64_t bucketId, uint32_t key);
+private:
+    InMemoryKV() = default;
+    using BucketMap = std::unordered_map<uint32_t, std::vector<uint8_t>>;
+    using DataMap = std::unordered_map<uint64_t, BucketMap>;
+
+    DataMap data_;
+};
+
+} // namespace milvus

include/ncs/InMemoryNcs.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include "ncs/ncs.h"
+#include "ncs/InMemoryKV.h"
+#include <memory>
+#include <unistd.h>
+
+
+namespace milvus {
+
+class InMemoryNcs : public Ncs {
+public:
+    NcsStatus createBucket(uint64_t bucketId) override;
+    NcsStatus deleteBucket(uint64_t bucketId) override;
+    NcsBucketStatus getBucketNcsStatus(uint64_t bucketId) override;
+    bool isBucketExist(uint64_t bucketId) override;
+    ~InMemoryNcs() override = default;
+private:
+    InMemoryNcs() = default;
+    friend class InMemoryNcsFactory;
+};
+
+class InMemoryNcsFactory : public NcsFactory {
+public:
+    static const std::string KIND;
+    std::unique_ptr<Ncs> createNcs(const nlohmann::json& params = nlohmann::json{}) override;
+    const std::string& getKind() const override;
+};
+
+} // namespace milvus

include/ncs/RedisNcs.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#ifdef USE_REDIS
+
+#include "ncs/ncs.h"
+#include "ncs/RedisTypes.h"
+#include "log/Log.h"
+#include <memory>
+#include <string>
+#include <mutex>
+#include <hiredis/hiredis.h>
+
+namespace milvus {
+
+class RedisNcs : public Ncs {
+public:
+    NcsStatus createBucket(uint64_t bucketId) override;
+    NcsStatus deleteBucket(uint64_t bucketId) override;
+    NcsBucketStatus getBucketNcsStatus(uint64_t bucketId) override;
+    bool isBucketExist(uint64_t bucketId) override;
+    ~RedisNcs() override;
+
+private:
+    explicit RedisNcs(const std::string& host, int port);
+    ncs::RedisContextPtr context_;
+    std::string host_;
+    int port_ = 0;
+    std::mutex mutex_;
+
+    friend class RedisNcsFactory;
+};
+
+class RedisNcsFactory : public NcsFactory {
+public:
+    static const std::string KIND;
+    std::unique_ptr<Ncs> createNcs(const nlohmann::json& params = nlohmann::json{}) override;
+    const std::string& getKind() const override;
+};
+
+} // namespace milvus
+
+#endif // USE_REDIS

include/ncs/RedisNcsConnector.h

@@ -0,0 +1,61 @@
+#pragma once
+
+#ifdef USE_REDIS
+
+#include "ncs/ncs.h"
+#include "ncs/RedisTypes.h"
+#include "log/Log.h"
+#include <memory>
+#include <string>
+#include <vector>
+#include <hiredis/hiredis.h>
+
+namespace milvus {
+
+/**
+ * @brief Redis-based NCS connector with single connection.
+ * 
+ * This connector uses a single Redis connection and is NOT thread-safe.
+ * Each thread should have its own connector instance.
+ * 
+ * Thread-safety is achieved at a higher level (e.g., NCSReader) by using
+ * thread_local connectors, one per thread.
+ * 
+ * Uses Redis pipelining for efficient batch operations.
+ */
+class RedisNcsConnector : public NcsConnector {
+public:
+    ~RedisNcsConnector() override;
+
+    std::vector<NcsStatus> multiGet(const std::vector<uint32_t>& keys, const std::vector<boost::span<uint8_t>>& buffs) override;
+    std::vector<NcsStatus> multiPut(const std::vector<uint32_t>& keys, const std::vector<boost::span<uint8_t>>& buffs) override;
+    std::vector<NcsStatus> multiDelete(const std::vector<uint32_t>& keys) override;
+
+private:
+    explicit RedisNcsConnector(uint64_t bucketId, const std::string& host, int port);
+
+    /**
+     * @brief Ensure connection is valid, reconnect if needed.
+     * @return true if connection is valid, false otherwise.
+     */
+    bool ensureConnected();
+
+    ncs::RedisReplyPtr getSafeReply();
+
+    ncs::RedisContextPtr ctx_{nullptr, redisFree};
+    std::string host_;
+    int port_ = 0;
+
+    friend class RedisNcsConnectorCreator;
+};
+
+class RedisNcsConnectorCreator : public NcsConnectorCreator {
+public:
+    static const std::string KIND;
+    NcsConnector* factoryMethod(const NcsDescriptor* descriptor) override;
+    const std::string& getKind() const override;
+};
+
+} // namespace milvus
+
+#endif // USE_REDIS

include/ncs/RedisTypes.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#ifdef USE_REDIS
+
+#include <memory>
+#include <hiredis/hiredis.h>
+
+namespace milvus {
+namespace ncs {
+
+using RedisReplyPtr = std::unique_ptr<redisReply, decltype(&freeReplyObject)>;
+using RedisContextPtr = std::unique_ptr<redisContext, decltype(&redisFree)>;
+
+} // namespace ncs
+} // namespace milvus
+
+#endif // USE_REDIS