diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 36cd0b16c..baa0c50f3 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -154,7 +154,7 @@ if (SVS_RUNTIME_ENABLE_LVQ_LEANVEC) else() # Links to LTO-enabled static library, requires GCC/G++ 11.2 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "11.2" AND CMAKE_CXX_COMPILER_VERSION VERSION_LESS "11.3") - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-lto-nightly-2026-10-08-1629.tar.gz" + set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-lto-nightly-2026-10-15-1661.tar.gz" CACHE STRING "URL to download SVS shared library") else() # The fallback is correct but slower, so nothing downstream fails and CI diff --git a/bindings/c/src/dispatcher_dynamic_vamana.cpp b/bindings/c/src/dispatcher_dynamic_vamana.cpp index 9686ebc29..ff1653b21 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.cpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.cpp @@ -125,8 +125,13 @@ svs::DynamicVamana load_stream_dynamic_vamana_index( // svs_c.h lets the caller drop the stream once loading returns. That holds only while // assemble copies data out; a view allocator here would leave the index dangling. - return svs::DynamicVamana::assemble( - *stream, distance, std::move(pool), allocator, graph_allocator + return svs::DynamicVamana::assemble( + *stream, + svs::make_stream_loader< + svs::graphs::SimpleGraph>(graph_allocator), + svs::make_stream_loader(allocator), + distance, + std::move(pool) ); } diff --git a/bindings/c/src/dispatcher_vamana.cpp b/bindings/c/src/dispatcher_vamana.cpp index 09c6b3b8f..9cb724259 100644 --- a/bindings/c/src/dispatcher_vamana.cpp +++ b/bindings/c/src/dispatcher_vamana.cpp @@ -98,14 +98,17 @@ svs::Vamana load_stream_vamana_index( ) { using value_type = typename DataLoader::allocator_type::value_type; using data_type = typename DataLoader::data_type; + auto graph_allocator = allocator_builder.build_for_graph(); + auto data_allocator = allocator_builder.build(); // svs_c.h lets the caller drop the stream once loading returns. That holds only while // assemble copies data out; a view allocator here would leave the index dangling. - return svs::Vamana::assemble( + return svs::Vamana::assemble( *stream, + svs::make_stream_loader< + svs::graphs::SimpleGraph>(graph_allocator), + svs::make_stream_loader(data_allocator), distance, - std::move(pool), - allocator_builder.build(), - allocator_builder.build_for_graph() + std::move(pool) ); } diff --git a/bindings/cpp/CMakeLists.txt b/bindings/cpp/CMakeLists.txt index fffc8f037..18580ae54 100644 --- a/bindings/cpp/CMakeLists.txt +++ b/bindings/cpp/CMakeLists.txt @@ -143,7 +143,7 @@ if (SVS_RUNTIME_ENABLE_LVQ_LEANVEC) else() # Links to LTO-enabled static library, requires GCC/G++ 11.2 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "11.2" AND CMAKE_CXX_COMPILER_VERSION VERSION_LESS "11.3") - set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-lto-nightly-2026-10-01-1583.tar.gz" + set(SVS_URL "https://github.com/intel/ScalableVectorSearch/releases/download/nightly/svs-shared-library-lto-nightly-2026-10-15-1661.tar.gz" CACHE STRING "URL to download SVS shared library") else() if(SVS_REQUIRE_LTO_ARCHIVE) diff --git a/bindings/cpp/src/dynamic_vamana_index_impl.h b/bindings/cpp/src/dynamic_vamana_index_impl.h index f89e1bb59..3e2e53d54 100644 --- a/bindings/cpp/src/dynamic_vamana_index_impl.h +++ b/bindings/cpp/src/dynamic_vamana_index_impl.h @@ -545,18 +545,20 @@ class DynamicVamanaIndexImpl { ); } else { using storage_type = storage::StorageType_t; + using graph_type = + svs::graphs::SimpleGraph>; auto threadpool = default_threadpool(); svs::DistanceDispatcher distance_dispatcher(to_svs_distance(metric)); return distance_dispatcher([&](auto&& distance) { - return new svs::DynamicVamana( - svs::DynamicVamana::assemble( - stream, - std::forward(distance), - std::move(threadpool) - ) - ); + return new svs::DynamicVamana(svs::DynamicVamana::assemble( + stream, + svs::StreamLoader{}, + svs::StreamLoader{}, + std::forward(distance), + std::move(threadpool) + )); }); } } diff --git a/bindings/cpp/src/vamana_index_impl.h b/bindings/cpp/src/vamana_index_impl.h index 000ec6391..07998fb4f 100644 --- a/bindings/cpp/src/vamana_index_impl.h +++ b/bindings/cpp/src/vamana_index_impl.h @@ -502,11 +502,12 @@ class VamanaIndexImpl { using storage_type = storage::StorageType_t; auto threadpool = default_threadpool(); - return new svs::Vamana(svs::Vamana::assemble( + return new svs::Vamana(svs::Vamana::assemble( stream, + svs::StreamLoader>{}, + svs::make_stream_loader(std::forward(args)...), to_svs_distance(metric), - std::move(threadpool), - std::forward(args)... + std::move(threadpool) )); } } diff --git a/include/svs/core/loading.h b/include/svs/core/loading.h index b3adad0dd..4ab117b4f 100644 --- a/include/svs/core/loading.h +++ b/include/svs/core/loading.h @@ -17,8 +17,69 @@ #pragma once #include "svs/lib/misc.h" +#include "svs/lib/saveload/load.h" #include "svs/lib/threads.h" +#include +#include +#include +#include +#include + +namespace svs { + +/// +/// @brief Selects the element type and loader arguments for one component of an index. +/// +/// The orchestrator supplies the stream for native streams, or the unpacked component +/// directory for directory archives. +/// +/// @tparam T The type loaded from the component. +/// @tparam Args The types of the arguments forwarded to the load function (e.g. +/// allocators). +/// +template struct StreamLoader { + using return_type = T; + + explicit StreamLoader(Args... args) + : args_{std::move(args)...} {} + + T load(std::istream& stream) const { + return std::apply( + [&](const Args&... args) { return lib::load_from_stream(stream, args...); }, + args_ + ); + } + + T load(const std::filesystem::path& path) const { + return std::apply( + [&](const Args&... args) { return lib::load_from_disk(path, args...); }, + args_ + ); + } + + std::tuple args_; +}; + +/// +/// @brief Build a ``StreamLoader`` for ``T``, deducing the loader argument types. +/// +/// @tparam T The type loaded from the component. +/// +/// @param args Arguments forwarded to the load function, stored by value. +/// +template +StreamLoader...> make_stream_loader(Args&&... args) { + return StreamLoader...>{std::forward(args)...}; +} + +template inline constexpr bool is_stream_loader_v = false; + +template +inline constexpr bool is_stream_loader_v> = true; + +} // namespace svs + namespace svs::detail { /// diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index 7167f3023..3ec385b91 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -131,7 +131,9 @@ class ConcurrentDynamicVamana : public DynamicVamana { Distance distance, ThreadPoolProto threadpool_proto, bool debug_load_from_static = false - ) { + ) + requires(!std::is_base_of_v> && !is_stream_loader_v> && !is_stream_loader_v>) + { auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); auto make = [&](auto distance_function) { return ConcurrentDynamicVamana( @@ -159,20 +161,31 @@ class ConcurrentDynamicVamana : public DynamicVamana { /// /// Accepts both the native stream format and the directory-archive format. /// - /// @tparam Data The ``SegmentedBlocked`` dataset type to load. + /// @param stream Stream positioned at the start of a saved index. + /// @param graph_loader ``StreamLoader`` producing a concurrent ``SimpleBlockedGraph``. + /// @param data_loader ``StreamLoader`` producing ``SegmentedBlocked`` storage. + /// @param distance Distance functor or ``svs::DistanceType`` enum. + /// @param threadpool_proto Thread pool or number of threads to use. + /// @param debug_load_from_static Load a static index config with identity IDs. /// template < manager::QueryTypeDefinition QueryTypes, - typename Data, + typename GraphLoader, + typename DataLoader, typename Distance, - typename ThreadPoolProto, - typename... DataLoaderArgs> + typename ThreadPoolProto> static ConcurrentDynamicVamana assemble( std::istream& stream, + const GraphLoader& graph_loader, + const DataLoader& data_loader, Distance distance, ThreadPoolProto threadpool_proto, - DataLoaderArgs&&... data_args - ) { + bool debug_load_from_static = false + ) + requires(is_stream_loader_v && is_stream_loader_v) + { + using Graph = typename GraphLoader::return_type; + using Data = typename DataLoader::return_type; static_assert( index::vamana::concurrent::is_segmented_blocked_v< typename Data::allocator_type>, @@ -187,16 +200,11 @@ class ConcurrentDynamicVamana : public DynamicVamana { manager::as_typelist(), index::vamana::concurrent::auto_dynamic_assemble( stream, - [&]() -> default_graph_type { - return default_graph_type::load(stream); - }, - [&]() -> Data { - return lib::load_from_stream( - stream, SVS_FWD(data_args)... - ); - }, + [&]() -> Graph { return graph_loader.load(stream); }, + [&]() -> Data { return data_loader.load(stream); }, std::move(distance_function), - std::move(threadpool) + std::move(threadpool), + debug_load_from_static ) ); }; @@ -225,10 +233,11 @@ class ConcurrentDynamicVamana : public DynamicVamana { return assemble( config_path, - SVS_LAZY(default_graph_type::load(graph_path)), - lib::load_from_disk(data_path, SVS_FWD(data_args)...), + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), std::move(distance), - std::move(threadpool) + std::move(threadpool), + debug_load_from_static ); } }; diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h index 7ebfaaaa5..067c7c486 100644 --- a/include/svs/orchestrators/dynamic_vamana.h +++ b/include/svs/orchestrators/dynamic_vamana.h @@ -365,6 +365,7 @@ class DynamicVamana : public manager::IndexManager { typename DataLoader, typename Distance, typename ThreadPoolProto> + requires(!std::is_base_of_v> && !is_stream_loader_v> && !is_stream_loader_v>) static DynamicVamana assemble( ConfigProto&& config_proto, GraphLoader&& graph_loader, @@ -393,39 +394,47 @@ class DynamicVamana : public manager::IndexManager { /// /// @brief Assemble a DynamicVamana index from a serialized stream. /// + /// Accepts either the native stream format or a directory archive packed into the + /// stream. + /// /// @tparam QueryTypes The set of query element types supported by the resulting /// index. - /// @tparam Data The dataset type to load. + /// @tparam GraphLoader A ``StreamLoader`` for the graph. + /// @tparam DataLoader A ``StreamLoader`` for the dataset. /// @tparam Distance Distance functor or ``svs::DistanceType`` enum. /// @tparam ThreadPoolProto Thread pool type or size_t. - /// @tparam DataAllocator The type of allocator used for the dataset. - /// @tparam GraphAllocator The type of allocator used for the graph. Defaults to an - /// exact-size ``HugepageAllocator``; a blocked default commits a full block. /// /// @param stream Stream containing the serialized index. - /// @param distance Distance functor or enum. + /// @param graph_loader The ``StreamLoader`` for the graph to use. + /// @param data_loader The ``StreamLoader`` for the data to use. + /// @param distance Distance functor or ``svs::DistanceType`` enum. /// @param threadpool_proto Thread pool or number of threads to use. - /// @param data_allocator Allocator instance to use for the dataset. - /// @param graph_allocator Allocator instance to use for the graph. + /// @param debug_load_from_static Internal/unstable: load files produced by the static + /// index using an identity ID translation. + /// @param logger The logger to use for this index. Defaults to the global SVS logger + /// (``svs::logging::get()``). /// template < manager::QueryTypeDefinition QueryTypes, - typename Data, + typename GraphLoader, + typename DataLoader, typename Distance, - typename ThreadPoolProto, - typename DataAllocator = typename Data::allocator_type, - typename GraphAllocator = HugepageAllocator> + typename ThreadPoolProto> + requires(is_stream_loader_v && is_stream_loader_v) static DynamicVamana assemble( std::istream& stream, + const GraphLoader& graph_loader, + const DataLoader& data_loader, const Distance& distance, ThreadPoolProto threadpool_proto, - const DataAllocator& data_allocator = {}, - const GraphAllocator& graph_allocator = {} + bool debug_load_from_static = false, + svs::logging::logger_ptr logger = svs::logging::get() ) { + using Graph = typename GraphLoader::return_type; + using Data = typename DataLoader::return_type; auto deserializer = svs::lib::detail::Deserializer::build(stream); if (deserializer.is_native()) { auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); - using GraphType = graphs::SimpleGraph; if constexpr (std::is_same_v, DistanceType>) { auto dispatcher = DistanceDispatcher(distance); return dispatcher([&](auto distance_function) { @@ -433,15 +442,13 @@ class DynamicVamana : public manager::IndexManager { index::vamana::auto_dynamic_assemble( stream, // lazy graph loader - [&]() -> GraphType { - return GraphType::load(stream, graph_allocator); - }, + [&]() -> Graph { return graph_loader.load(stream); }, // lazy data loader - [&]() -> Data { - return lib::load_from_stream(stream, data_allocator); - }, + [&]() -> Data { return data_loader.load(stream); }, distance_function, - std::move(threadpool) + std::move(threadpool), + debug_load_from_static, + logger ) ); }); @@ -450,15 +457,13 @@ class DynamicVamana : public manager::IndexManager { index::vamana::auto_dynamic_assemble( stream, // lazy graph loader - [&]() -> GraphType { - return GraphType::load(stream, graph_allocator); - }, + [&]() -> Graph { return graph_loader.load(stream); }, // lazy data loader - [&]() -> Data { - return lib::load_from_stream(stream, data_allocator); - }, + [&]() -> Data { return data_loader.load(stream); }, distance, - std::move(threadpool) + std::move(threadpool), + debug_load_from_static, + logger ) ); } @@ -486,15 +491,55 @@ class DynamicVamana : public manager::IndexManager { return assemble( config_path, - svs::GraphLoader{graph_path, graph_allocator}, - lib::load_from_disk(data_path, data_allocator), + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), distance, threads::as_threadpool(std::move(threadpool_proto)), - false + debug_load_from_static, + logger ); } } + // Assembly from stream wrapper + /// + /// @brief Assemble a DynamicVamana index from a serialized stream. + /// + /// @tparam QueryTypes The set of query element types supported by the resulting + /// index. + /// @tparam Data The dataset type to load. + /// @tparam Distance Distance functor or ``svs::DistanceType`` enum. + /// @tparam ThreadPoolProto Thread pool type or size_t. + /// @tparam DataLoaderArgs Types of extra arguments forwarded to the data loader + /// (e.g. an allocator). + /// + /// @param stream Stream containing the serialized index. + /// @param distance Distance functor or enum. + /// @param threadpool_proto Thread pool or number of threads to use. + /// @param data_args Extra arguments forwarded to the data loader (e.g. an allocator). + /// + template < + manager::QueryTypeDefinition QueryTypes, + typename Data, + typename Distance, + typename ThreadPoolProto, + typename... DataLoaderArgs> + static DynamicVamana assemble( + std::istream& stream, + const Distance& distance, + ThreadPoolProto threadpool_proto, + DataLoaderArgs&&... data_args + ) { + using GraphType = svs::GraphLoader<>::return_type; + return assemble( + stream, + StreamLoader{}, + make_stream_loader(SVS_FWD(data_args)...), + distance, + std::move(threadpool_proto) + ); + } + /// @copydoc svs::Vamana::batch_iterator template svs::VamanaIterator batch_iterator( diff --git a/include/svs/orchestrators/vamana.h b/include/svs/orchestrators/vamana.h index 6618dfbff..5787e8a7a 100644 --- a/include/svs/orchestrators/vamana.h +++ b/include/svs/orchestrators/vamana.h @@ -305,6 +305,18 @@ class Vamana : public manager::IndexManager { using base_type = manager::IndexManager; using VamanaSearchParameters = index::vamana::VamanaSearchParameters; + /// @brief Graph type ``assemble`` uses by default when loading ``Data`` from a stream. + /// + /// View-backed data needs a graph viewing the same buffer, so its allocator is rebound + /// to ``uint32_t``; otherwise the graph uses a ``HugepageAllocator``. + template + using default_graph_type = graphs::SimpleGraph< + uint32_t, + std::conditional_t< + is_view_type_v, + lib::rebind_allocator_t, + HugepageAllocator>>; + /// @private struct BuildTag {}; /// @private @@ -442,15 +454,19 @@ class Vamana : public manager::IndexManager { typename Distance, typename ThreadPoolProto> static Vamana assemble( - ConfigLoaderProto config_proto, + ConfigLoaderProto&& config_proto, const GraphLoaderType& graph_loader, DataLoader&& data_loader, const Distance& distance, ThreadPoolProto threadpool_proto, svs::logging::logger_ptr logger = svs::logging::get() - ) { - // If given an `enum` for the distance type, than we need to dispatch over that - // enum. + ) + // Excludes `std::istream` so a `std::stringstream` config prefers the + // `std::istream&` overload below instead of binding here via `ConfigLoaderProto&&`. + requires(!std::is_base_of_v> && !is_stream_loader_v> && !is_stream_loader_v>) + { + // If given an `enum` for the distance type, than we need to dispatch over + // that enum. // // Otherwise, we can directly forward the provided distance function. auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); @@ -480,70 +496,23 @@ class Vamana : public manager::IndexManager { } } - /// - /// @brief Assemble a Vamana index from an in-memory, view-backed stream. - /// - /// @param stream The stream to load from. See ``svs::Vamana::save``. - /// @param distance The distance functor or ``svs::DistanceType`` enum to use for - /// similarity search computations. - /// @param threadpool_proto Precursor for the thread pool to use. Can either be an - /// acceptable thread pool instance or an integer specifying the number of - /// threads to use. - /// @param data_args Forwarded to the dataset loader. An allocator passed here must be - /// bound to ``stream``. - /// - /// The stream must be an in-memory stream in native format; the returned index views - /// its buffer directly, so the stream must outlive the index. - /// - /// @copydoc threadpool_requirements - /// - /// @sa save, build - /// - template < - manager::QueryTypeDefinition QueryTypes, - typename Data, - typename Distance, - typename ThreadPoolProto, - typename... DataLoaderArgs> - requires is_view_type_v - static Vamana assemble( - std::istream& stream, - const Distance& distance, - ThreadPoolProto threadpool_proto, - DataLoaderArgs&&... data_args - ) { - auto deserializer = svs::lib::detail::Deserializer::build(stream); - if (!deserializer.is_native()) { - throw ANNEXCEPTION( - "Cannot load a view-backed Vamana index from a directory archive. " - "Directory archives are unpacked to a temporary directory and cannot " - "back a view; use the native stream format instead." - ); - } - - using Allocator = lib::rebind_allocator_t; - using GraphType = graphs::SimpleGraph; - auto load_graph = [&]() -> GraphType { return GraphType::load(stream); }; - auto load_data = [&]() -> Data { - return lib::load_from_stream(stream, SVS_FWD(data_args)...); - }; - - return assemble_native( - stream, load_graph, load_data, distance, std::move(threadpool_proto) - ); - } - /// /// @brief Assemble a Vamana index from a stream. /// - /// @param stream The stream to load from. See ``svs::Vamana::save``. + /// @param stream The stream to load from. See ``svs::Vamana::save``. Accepts either + /// the native stream format or a directory archive. + /// @param graph_loader The ``StreamLoader`` for the graph. + /// @param data_loader The ``StreamLoader`` for the dataset. If view-backed + /// (``Data``'s allocator is a view allocator), loading requires the native + /// format, and the returned index views the stream's buffer directly, so the + /// stream must outlive the index. /// @param distance The distance functor or ``svs::DistanceType`` enum to use for /// similarity search computations. /// @param threadpool_proto Precursor for the thread pool to use. Can either be an /// acceptable thread pool instance or an integer specifying the number of /// threads to use. - /// @param data_allocator Allocator to use for the loaded data. - /// @param graph_allocator Allocator to use for the loaded graph. + /// @param logger The logger to use for this index. Defaults to the global SVS logger + /// (``svs::logging::get()``). /// /// @copydoc threadpool_requirements /// @@ -551,31 +520,40 @@ class Vamana : public manager::IndexManager { /// template < manager::QueryTypeDefinition QueryTypes, - typename Data, + typename GraphLoader, + typename DataLoader, typename Distance, - typename ThreadPoolProto, - typename DataAllocator = typename Data::allocator_type, - typename GraphAllocator = HugepageAllocator> - requires(!is_view_type_v) + typename ThreadPoolProto> static Vamana assemble( std::istream& stream, + const GraphLoader& graph_loader, + const DataLoader& data_loader, const Distance& distance, ThreadPoolProto threadpool_proto, - const DataAllocator& data_allocator = {}, - const GraphAllocator& graph_allocator = {} - ) { + svs::logging::logger_ptr logger = svs::logging::get() + ) + requires(is_stream_loader_v && is_stream_loader_v) + { + using Graph = typename GraphLoader::return_type; + using Data = typename DataLoader::return_type; auto deserializer = svs::lib::detail::Deserializer::build(stream); if (deserializer.is_native()) { - using GraphType = graphs::SimpleGraph; - auto load_graph = [&]() -> GraphType { - return GraphType::load(stream, graph_allocator); - }; - auto load_data = [&]() -> Data { - return lib::load_from_stream(stream, data_allocator); - }; + auto load_graph = [&]() -> Graph { return graph_loader.load(stream); }; + auto load_data = [&]() -> Data { return data_loader.load(stream); }; return assemble_native( - stream, load_graph, load_data, distance, std::move(threadpool_proto) + stream, + load_graph, + load_data, + distance, + std::move(threadpool_proto), + std::move(logger) + ); + } else if constexpr (is_view_type_v) { + throw ANNEXCEPTION( + "Cannot load a view-backed Vamana index from a directory archive. " + "Directory archives are unpacked to a temporary directory and cannot " + "back a view; use the native stream format instead." ); } else { namespace fs = std::filesystem; @@ -601,14 +579,54 @@ class Vamana : public manager::IndexManager { return assemble( config_path, - svs::GraphLoader{graph_path, graph_allocator}, - lib::load_from_disk(data_path, data_allocator), + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), distance, - threads::as_threadpool(std::move(threadpool_proto)) + threads::as_threadpool(std::move(threadpool_proto)), + std::move(logger) ); } } + /// + /// @brief Assemble a Vamana index from a stream, using default loaders. + /// + /// @param stream The stream to load from. See ``svs::Vamana::save``. If view-backed + /// (``Data``'s allocator is a view allocator), the returned index views the + /// stream's buffer directly, so the stream must outlive the index. + /// @param distance The distance functor or ``svs::DistanceType`` enum to use for + /// similarity search computations. + /// @param threadpool_proto Precursor for the thread pool to use. Can either be an + /// acceptable thread pool instance or an integer specifying the number of + /// threads to use. + /// @param data_args Forwarded to the dataset loader. An allocator passed here must be + /// bound to ``stream`` when ``Data``'s allocator is a view allocator. + /// + /// @copydoc threadpool_requirements + /// + /// @sa save, build + /// + template < + manager::QueryTypeDefinition QueryTypes, + typename Data, + typename Distance, + typename ThreadPoolProto, + typename... DataLoaderArgs> + static Vamana assemble( + std::istream& stream, + const Distance& distance, + ThreadPoolProto threadpool_proto, + DataLoaderArgs&&... data_args + ) { + return assemble( + stream, + StreamLoader>{}, + make_stream_loader(SVS_FWD(data_args)...), + distance, + std::move(threadpool_proto) + ); + } + /// /// @brief Construct the a Vamana Index for the given dataset. /// @@ -770,7 +788,8 @@ class Vamana : public manager::IndexManager { const GraphLoaderFn& load_graph, const DataLoaderFn& load_data, const Distance& distance, - ThreadPoolProto threadpool_proto + ThreadPoolProto threadpool_proto, + svs::logging::logger_ptr logger ) { auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); if constexpr (std::is_same_v) { @@ -782,7 +801,8 @@ class Vamana : public manager::IndexManager { load_graph, load_data, distance_function, - std::move(threadpool) + std::move(threadpool), + std::move(logger) ); }); } else { @@ -792,7 +812,8 @@ class Vamana : public manager::IndexManager { load_graph, load_data, distance, - std::move(threadpool) + std::move(threadpool), + std::move(logger) ); } } diff --git a/tests/svs/core/loading.cpp b/tests/svs/core/loading.cpp index 896c9048d..9336704f7 100644 --- a/tests/svs/core/loading.cpp +++ b/tests/svs/core/loading.cpp @@ -19,11 +19,15 @@ // Additional headers. #include "svs/lib/misc.h" +#include "svs/lib/saveload.h" #include "svs/lib/threads.h" // Catch2 #include "catch2/catch_test_macros.hpp" +// stl +#include + namespace { template struct Loadable { @@ -59,6 +63,26 @@ struct MoveOnly { size_t value_; }; +// Context-free loadable type whose load takes an extra argument. +struct StreamLoaderToy { + int32_t val_; + size_t moveonly_value_ = 0; + + static constexpr std::string_view serialization_schema = "svstest_stream_loader_toy"; + static constexpr svs::lib::Version save_version{0, 0, 0}; + + svs::lib::SaveTable metadata() const { + return svs::lib::SaveTable( + serialization_schema, save_version, {SVS_LIST_SAVE_(val)} + ); + } + + static StreamLoaderToy + load(const svs::lib::ContextFreeLoadTable& table, const MoveOnly& arg) { + return StreamLoaderToy{SVS_LOAD_MEMBER_AT_(table, val), arg.value_}; + } +}; + } // namespace CATCH_TEST_CASE("Dispatch Load", "[core][dispatch_load]") { @@ -148,3 +172,27 @@ CATCH_TEST_CASE("Dispatch Load", "[core][dispatch_load]") { } } } + +CATCH_TEST_CASE("Stream Loader", "[core][stream_loader]") { + // `StreamLoader` must work when an `Args` element is move-only. + static_assert(!std::is_copy_constructible_v< + svs::StreamLoader>); + + auto x = StreamLoaderToy{42, 0}; + auto stream = std::stringstream(); + svs::lib::save_to_stream(x, stream); + + CATCH_SECTION("make_stream_loader") { + auto loader = svs::make_stream_loader(MoveOnly(10)); + auto result = loader.load(stream); + CATCH_REQUIRE(result.val_ == 42); + CATCH_REQUIRE(result.moveonly_value_ == 10); + } + + CATCH_SECTION("Direct construction") { + auto loader = svs::StreamLoader{MoveOnly(10)}; + auto result = loader.load(stream); + CATCH_REQUIRE(result.val_ == 42); + CATCH_REQUIRE(result.moveonly_value_ == 10); + } +} diff --git a/tests/svs/orchestrators/concurrent_dynamic_vamana.cpp b/tests/svs/orchestrators/concurrent_dynamic_vamana.cpp index 6e75631e0..699152e1a 100644 --- a/tests/svs/orchestrators/concurrent_dynamic_vamana.cpp +++ b/tests/svs/orchestrators/concurrent_dynamic_vamana.cpp @@ -322,10 +322,13 @@ CATCH_TEST_CASE( CATCH_SECTION("Native stream") { std::stringstream stream; index.save(stream); - svs::DynamicVamana loaded = - svs::ConcurrentDynamicVamana::assemble( - stream, svs::distance::DistanceL2(), num_threads - ); + svs::DynamicVamana loaded = svs::ConcurrentDynamicVamana::assemble( + stream, + svs::StreamLoader>{}, + svs::StreamLoader{}, + svs::distance::DistanceL2(), + num_threads + ); CATCH_REQUIRE(loaded.size() == index.size()); require_same_results(index, loaded, k); } @@ -339,10 +342,13 @@ CATCH_TEST_CASE( ); svs::lib::DirectoryArchiver::pack(tempdir, stream); } - svs::DynamicVamana loaded = - svs::ConcurrentDynamicVamana::assemble( - stream, svs::DistanceType::L2, num_threads - ); + svs::DynamicVamana loaded = svs::ConcurrentDynamicVamana::assemble( + stream, + svs::StreamLoader>{}, + svs::StreamLoader{}, + svs::DistanceType::L2, + num_threads + ); CATCH_REQUIRE(loaded.size() == index.size()); require_same_results(index, loaded, k); }