From f681f8cce1e5ada4726472610163d178c3906844 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Tue, 13 Oct 2026 02:27:06 -0700 Subject: [PATCH 1/9] feat(core): add StreamLoader StreamLoader carries the element type and loader arguments (e.g. an allocator) for one component of an index saved to a stream. The orchestrator hands it either the stream (native format) or the unpacked component directory (directory archive). Co-Authored-By: Claude Opus 5.5 --- include/svs/core/loading.h | 48 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/include/svs/core/loading.h b/include/svs/core/loading.h index b3adad0dd..28b56b848 100644 --- a/include/svs/core/loading.h +++ b/include/svs/core/loading.h @@ -17,8 +17,56 @@ #pragma once #include "svs/lib/misc.h" +#include "svs/lib/saveload/load.h" #include "svs/lib/threads.h" +#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_; +}; + +template inline constexpr bool is_stream_loader_v = false; + +template +inline constexpr bool is_stream_loader_v> = true; + +} // namespace svs + namespace svs::detail { /// From 4f2975f16be22460cbfdc9efaf550d59a1148e81 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Tue, 13 Oct 2026 02:27:09 -0700 Subject: [PATCH 2/9] fix(orchestrators): route stream assemble through the loader-based assemble Vamana, DynamicVamana and ConcurrentDynamicVamana now load from a stream through the existing assemble(config_proto, graph_loader, data_loader, ...) when config_proto is a std::istream and both loaders are StreamLoader. The stream body moved unchanged into a private helper. - The released assemble(stream, distance, pool, data_args...) on Vamana and DynamicVamana stays as a thin wrapper with its v0.4.0 signature and default graph allocators. - Removed, never released: Vamana's view-backed overload and the (data_allocator, graph_allocator) overloads from #402, and ConcurrentDynamicVamana's stream overload from #413. - The stream path used to drop the logger and debug_load_from_static; both are now passed through. - The C API and C++ runtime call the loader-based form. Public binding headers are unchanged. Co-Authored-By: Claude Opus 5.5 --- bindings/c/src/dispatcher_dynamic_vamana.cpp | 10 +- bindings/c/src/dispatcher_vamana.cpp | 12 +- bindings/cpp/src/dynamic_vamana_index_impl.h | 16 +- bindings/cpp/src/vamana_index_impl.h | 8 +- .../orchestrators/concurrent_dynamic_vamana.h | 96 +++--- include/svs/orchestrators/dynamic_vamana.h | 228 +++++++++------ include/svs/orchestrators/vamana.h | 273 ++++++++++-------- .../concurrent_dynamic_vamana.cpp | 22 +- 8 files changed, 390 insertions(+), 275 deletions(-) diff --git a/bindings/c/src/dispatcher_dynamic_vamana.cpp b/bindings/c/src/dispatcher_dynamic_vamana.cpp index 96a706f27..407d366d6 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.cpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.cpp @@ -118,8 +118,14 @@ 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::StreamLoader< + svs::graphs::SimpleGraph, + decltype(graph_allocator)>{graph_allocator}, + svs::StreamLoader{allocator}, + distance, + std::move(pool) ); } diff --git a/bindings/c/src/dispatcher_vamana.cpp b/bindings/c/src/dispatcher_vamana.cpp index e2414d95e..09d5e54d6 100644 --- a/bindings/c/src/dispatcher_vamana.cpp +++ b/bindings/c/src/dispatcher_vamana.cpp @@ -92,14 +92,18 @@ 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::StreamLoader< + svs::graphs::SimpleGraph, + decltype(graph_allocator)>{graph_allocator}, + svs::StreamLoader{data_allocator}, distance, - std::move(pool), - allocator_builder.build(), - allocator_builder.build_for_graph() + std::move(pool) ); } 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..c558c5049 100644 --- a/bindings/cpp/src/vamana_index_impl.h +++ b/bindings/cpp/src/vamana_index_impl.h @@ -502,11 +502,13 @@ 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::StreamLoader...>{ + std::forward(args)...}, to_svs_distance(metric), - std::move(threadpool), - std::forward(args)... + std::move(threadpool) )); } } diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index 7167f3023..bb20a93ed 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -132,47 +132,75 @@ class ConcurrentDynamicVamana : public DynamicVamana { ThreadPoolProto threadpool_proto, bool debug_load_from_static = false ) { - auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); - auto make = [&](auto distance_function) { - return ConcurrentDynamicVamana( - AssembleTag{}, - manager::as_typelist(), - index::vamana::concurrent::auto_dynamic_assemble( - std::forward(config_proto), - std::forward(graph_loader), - std::forward(data_loader), - std::move(distance_function), - std::move(threadpool), - debug_load_from_static - ) + if constexpr (std::is_base_of_v>) { + static_assert( + is_stream_loader_v> && + is_stream_loader_v>, + "Assembling from a std::istream requires svs::StreamLoader graph and " + "data loaders." + ); + return assemble_from_stream( + config_proto, + std::forward(graph_loader), + std::forward(data_loader), + std::move(distance), + std::move(threadpool_proto), + debug_load_from_static ); - }; - if constexpr (std::is_same_v, DistanceType>) { - return DistanceDispatcher(distance)(make); } else { - return make(std::move(distance)); + static_assert( + !is_stream_loader_v>, + "StreamLoader requires a std::istream config." + ); + static_assert( + !is_stream_loader_v>, + "StreamLoader requires a std::istream config." + ); + auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); + auto make = [&](auto distance_function) { + return ConcurrentDynamicVamana( + AssembleTag{}, + manager::as_typelist(), + index::vamana::concurrent::auto_dynamic_assemble( + std::forward(config_proto), + std::forward(graph_loader), + std::forward(data_loader), + std::move(distance_function), + std::move(threadpool), + debug_load_from_static + ) + ); + }; + if constexpr (std::is_same_v, DistanceType>) { + return DistanceDispatcher(distance)(make); + } else { + return make(std::move(distance)); + } } } + private: /// /// @brief Reload a ConcurrentDynamicVamana index saved with ``save(std::ostream&)``. /// /// Accepts both the native stream format and the directory-archive format. /// - /// @tparam Data The ``SegmentedBlocked`` dataset type to load. - /// template < manager::QueryTypeDefinition QueryTypes, - typename Data, + typename GraphLoader, + typename DataLoader, typename Distance, - typename ThreadPoolProto, - typename... DataLoaderArgs> - static ConcurrentDynamicVamana assemble( + typename ThreadPoolProto> + static ConcurrentDynamicVamana assemble_from_stream( std::istream& stream, + GraphLoader&& graph_loader, + DataLoader&& data_loader, Distance distance, ThreadPoolProto threadpool_proto, - DataLoaderArgs&&... data_args + bool debug_load_from_static ) { + using Graph = typename std::remove_cvref_t::return_type; + using Data = typename std::remove_cvref_t::return_type; static_assert( index::vamana::concurrent::is_segmented_blocked_v< typename Data::allocator_type>, @@ -187,16 +215,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 +248,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..7e5cf08f1 100644 --- a/include/svs/orchestrators/dynamic_vamana.h +++ b/include/svs/orchestrators/dynamic_vamana.h @@ -374,19 +374,47 @@ class DynamicVamana : public manager::IndexManager { bool debug_load_from_static = false, svs::logging::logger_ptr logger = svs::logging::get() ) { - return DynamicVamana( - AssembleTag(), - manager::as_typelist(), - index::vamana::auto_dynamic_assemble( - std::forward(config_proto), - std::forward(graph_loader), - std::forward(data_loader), + if constexpr (std::is_base_of_v>) { + static_assert( + is_stream_loader_v>, + "Assembling from a stream requires a StreamLoader for the graph." + ); + static_assert( + is_stream_loader_v>, + "Assembling from a stream requires a StreamLoader for the data." + ); + return assemble_from_stream( + config_proto, + graph_loader, + data_loader, distance, - threads::as_threadpool(std::move(threadpool_proto)), + std::move(threadpool_proto), debug_load_from_static, std::move(logger) - ) - ); + ); + } else { + static_assert( + !is_stream_loader_v>, + "A StreamLoader requires a std::istream config to assemble from." + ); + static_assert( + !is_stream_loader_v>, + "A StreamLoader requires a std::istream config to assemble from." + ); + return DynamicVamana( + AssembleTag(), + manager::as_typelist(), + index::vamana::auto_dynamic_assemble( + std::forward(config_proto), + std::forward(graph_loader), + std::forward(data_loader), + distance, + threads::as_threadpool(std::move(threadpool_proto)), + debug_load_from_static, + std::move(logger) + ) + ); + } } // Assembly from stream @@ -398,101 +426,34 @@ class DynamicVamana : public manager::IndexManager { /// @tparam Data The dataset type to load. /// @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. + /// @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_allocator Allocator instance to use for the dataset. - /// @param graph_allocator Allocator instance to use for the graph. + /// @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 DataAllocator = typename Data::allocator_type, - typename GraphAllocator = HugepageAllocator> + typename... DataLoaderArgs> static DynamicVamana assemble( std::istream& stream, const Distance& distance, ThreadPoolProto threadpool_proto, - const DataAllocator& data_allocator = {}, - const GraphAllocator& graph_allocator = {} + DataLoaderArgs&&... data_args ) { - 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) { - return make_dynamic_vamana>( - index::vamana::auto_dynamic_assemble( - stream, - // lazy graph loader - [&]() -> GraphType { - return GraphType::load(stream, graph_allocator); - }, - // lazy data loader - [&]() -> Data { - return lib::load_from_stream(stream, data_allocator); - }, - distance_function, - std::move(threadpool) - ) - ); - }); - } else { - return make_dynamic_vamana>( - index::vamana::auto_dynamic_assemble( - stream, - // lazy graph loader - [&]() -> GraphType { - return GraphType::load(stream, graph_allocator); - }, - // lazy data loader - [&]() -> Data { - return lib::load_from_stream(stream, data_allocator); - }, - distance, - std::move(threadpool) - ) - ); - } - } else { - namespace fs = std::filesystem; - lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; - lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); - - const auto config_path = tempdir.get() / "config"; - if (!fs::is_directory(config_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" - ); - } - - const auto graph_path = tempdir.get() / "graph"; - if (!fs::is_directory(graph_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" - ); - } - - const auto data_path = tempdir.get() / "data"; - if (!fs::is_directory(data_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); - } - - return assemble( - config_path, - svs::GraphLoader{graph_path, graph_allocator}, - lib::load_from_disk(data_path, data_allocator), - distance, - threads::as_threadpool(std::move(threadpool_proto)), - false - ); - } + using GraphType = svs::GraphLoader<>::return_type; + return assemble( + stream, + StreamLoader{}, + StreamLoader...>{SVS_FWD(data_args)...}, + distance, + std::move(threadpool_proto) + ); } /// @copydoc svs::Vamana::batch_iterator @@ -564,6 +525,93 @@ class DynamicVamana : public manager::IndexManager { /// @brief Check if the index has been consolidated - no `Deleted` entries remain. bool is_consolidated() const { return impl_->is_consolidated(); } + + private: + template < + manager::QueryTypeDefinition QueryTypes, + typename GraphLoader, + typename DataLoader, + typename Distance, + typename ThreadPoolProto> + static DynamicVamana assemble_from_stream( + std::istream& stream, + const GraphLoader& graph_loader, + const DataLoader& data_loader, + const Distance& distance, + ThreadPoolProto threadpool_proto, + bool debug_load_from_static, + svs::logging::logger_ptr logger + ) { + using Data = typename DataLoader::return_type; + using GraphType = typename GraphLoader::return_type; + auto deserializer = svs::lib::detail::Deserializer::build(stream); + if (deserializer.is_native()) { + auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); + if constexpr (std::is_same_v, DistanceType>) { + auto dispatcher = DistanceDispatcher(distance); + return dispatcher([&](auto distance_function) { + return make_dynamic_vamana>( + index::vamana::auto_dynamic_assemble( + stream, + // lazy graph loader + [&]() -> GraphType { return graph_loader.load(stream); }, + // lazy data loader + [&]() -> Data { return data_loader.load(stream); }, + distance_function, + std::move(threadpool), + debug_load_from_static, + logger + ) + ); + }); + } else { + return make_dynamic_vamana>( + index::vamana::auto_dynamic_assemble( + stream, + // lazy graph loader + [&]() -> GraphType { return graph_loader.load(stream); }, + // lazy data loader + [&]() -> Data { return data_loader.load(stream); }, + distance, + std::move(threadpool), + debug_load_from_static, + logger + ) + ); + } + } else { + namespace fs = std::filesystem; + lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; + lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); + + const auto config_path = tempdir.get() / "config"; + if (!fs::is_directory(config_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" + ); + } + + const auto graph_path = tempdir.get() / "graph"; + if (!fs::is_directory(graph_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" + ); + } + + const auto data_path = tempdir.get() / "data"; + if (!fs::is_directory(data_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); + } + + return assemble( + config_path, + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), + distance, + threads::as_threadpool(std::move(threadpool_proto)), + debug_load_from_static, + logger + ); + } + } }; /// diff --git a/include/svs/orchestrators/vamana.h b/include/svs/orchestrators/vamana.h index 6618dfbff..bcd02865d 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,58 +454,86 @@ 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. - // - // Otherwise, we can directly forward the provided distance function. - auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); - if constexpr (std::is_same_v) { - auto dispatcher = DistanceDispatcher(distance); - return dispatcher([&](auto distance_function) { + if constexpr (std::is_base_of_v< + std::istream, + std::remove_cvref_t>) { + static_assert( + is_stream_loader_v>, + "Assembling from a stream requires a StreamLoader for the graph." + ); + static_assert( + is_stream_loader_v>, + "Assembling from a stream requires a StreamLoader for the data." + ); + return assemble_from_stream( + config_proto, + graph_loader, + data_loader, + distance, + std::move(threadpool_proto), + std::move(logger) + ); + } else { + static_assert( + !is_stream_loader_v>, + "A StreamLoader requires a std::istream config to assemble from." + ); + static_assert( + !is_stream_loader_v>, + "A StreamLoader requires a std::istream config to assemble from." + ); + // 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)); + if constexpr (std::is_same_v) { + auto dispatcher = DistanceDispatcher(distance); + return dispatcher([&](auto distance_function) { + return make_vamana>( + AssembleTag(), + std::forward(config_proto), + graph_loader, + std::forward(data_loader), + distance_function, + std::move(threadpool), + std::move(logger) + ); + }); + } else { return make_vamana>( AssembleTag(), std::forward(config_proto), graph_loader, std::forward(data_loader), - distance_function, + distance, std::move(threadpool), std::move(logger) ); - }); - } else { - return make_vamana>( - AssembleTag(), - std::forward(config_proto), - graph_loader, - std::forward(data_loader), - distance, - std::move(threadpool), - std::move(logger) - ); + } } } /// - /// @brief Assemble a Vamana index from an in-memory, view-backed stream. + /// @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``. 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``. - /// - /// 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. + /// bound to ``stream`` when ``Data``'s allocator is a view allocator. /// /// @copydoc threadpool_requirements /// @@ -505,110 +545,21 @@ class Vamana : public manager::IndexManager { 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) + return assemble( + stream, + StreamLoader>{}, + StreamLoader...>{SVS_FWD(data_args)...}, + 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 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. - /// - /// @copydoc threadpool_requirements - /// - /// @sa save, build - /// - template < - manager::QueryTypeDefinition QueryTypes, - typename Data, - typename Distance, - typename ThreadPoolProto, - typename DataAllocator = typename Data::allocator_type, - typename GraphAllocator = HugepageAllocator> - requires(!is_view_type_v) - static Vamana assemble( - std::istream& stream, - const Distance& distance, - ThreadPoolProto threadpool_proto, - const DataAllocator& data_allocator = {}, - const GraphAllocator& graph_allocator = {} - ) { - 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); - }; - - return assemble_native( - stream, load_graph, load_data, distance, std::move(threadpool_proto) - ); - } else { - namespace fs = std::filesystem; - lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; - lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); - - const auto config_path = tempdir.get() / "config"; - if (!fs::is_directory(config_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" - ); - } - - const auto graph_path = tempdir.get() / "graph"; - if (!fs::is_directory(graph_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" - ); - } - - const auto data_path = tempdir.get() / "data"; - if (!fs::is_directory(data_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); - } - - return assemble( - config_path, - svs::GraphLoader{graph_path, graph_allocator}, - lib::load_from_disk(data_path, data_allocator), - distance, - threads::as_threadpool(std::move(threadpool_proto)) - ); - } - } - /// /// @brief Construct the a Vamana Index for the given dataset. /// @@ -759,6 +710,75 @@ class Vamana : public manager::IndexManager { } private: + template < + manager::QueryTypeDefinition QueryTypes, + typename GraphLoaderType, + typename DataLoader, + typename Distance, + typename ThreadPoolProto> + static Vamana assemble_from_stream( + std::istream& stream, + const GraphLoaderType& graph_loader, + const DataLoader& data_loader, + const Distance& distance, + ThreadPoolProto threadpool_proto, + svs::logging::logger_ptr logger + ) { + using Data = typename DataLoader::return_type; + using GraphType = typename GraphLoaderType::return_type; + + auto deserializer = svs::lib::detail::Deserializer::build(stream); + if (deserializer.is_native()) { + auto load_graph = [&]() -> GraphType { 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), + 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; + lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; + lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); + + const auto config_path = tempdir.get() / "config"; + if (!fs::is_directory(config_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" + ); + } + + const auto graph_path = tempdir.get() / "graph"; + if (!fs::is_directory(graph_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" + ); + } + + const auto data_path = tempdir.get() / "data"; + if (!fs::is_directory(data_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); + } + + return assemble( + config_path, + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), + distance, + threads::as_threadpool(std::move(threadpool_proto)), + std::move(logger) + ); + } + } + template < manager::QueryTypeDefinition QueryTypes, typename GraphLoaderFn, @@ -770,7 +790,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 +803,8 @@ class Vamana : public manager::IndexManager { load_graph, load_data, distance_function, - std::move(threadpool) + std::move(threadpool), + std::move(logger) ); }); } else { @@ -792,7 +814,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/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); } From d53ef443e5e896596aadb569b35ba6ad6464ed40 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Tue, 13 Oct 2026 05:56:16 -0700 Subject: [PATCH 3/9] refactor(orchestrators): split stream assemble into StreamLoader overloads Vamana, DynamicVamana and ConcurrentDynamicVamana now have a dedicated assemble(std::istream&, StreamLoader graph, StreamLoader data, ...) overload instead of branching inside the file-layout assemble. The file-layout overload is constrained to exclude streams and StreamLoader, so a mismatched call finds no viable overload rather than hitting a static_assert. The stream body moved unchanged; call sites are unaffected. Co-Authored-By: Claude Opus 5.5 --- .../orchestrators/concurrent_dynamic_vamana.h | 87 +++---- include/svs/orchestrators/dynamic_vamana.h | 244 +++++++++--------- include/svs/orchestrators/vamana.h | 234 +++++++++-------- 3 files changed, 271 insertions(+), 294 deletions(-) diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index bb20a93ed..cb914021d 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -131,76 +131,59 @@ class ConcurrentDynamicVamana : public DynamicVamana { Distance distance, ThreadPoolProto threadpool_proto, bool debug_load_from_static = false - ) { - if constexpr (std::is_base_of_v>) { - static_assert( - is_stream_loader_v> && - is_stream_loader_v>, - "Assembling from a std::istream requires svs::StreamLoader graph and " - "data loaders." - ); - return assemble_from_stream( - config_proto, - std::forward(graph_loader), - std::forward(data_loader), - std::move(distance), - std::move(threadpool_proto), - debug_load_from_static + ) + 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( + AssembleTag{}, + manager::as_typelist(), + index::vamana::concurrent::auto_dynamic_assemble( + std::forward(config_proto), + std::forward(graph_loader), + std::forward(data_loader), + std::move(distance_function), + std::move(threadpool), + debug_load_from_static + ) ); + }; + if constexpr (std::is_same_v, DistanceType>) { + return DistanceDispatcher(distance)(make); } else { - static_assert( - !is_stream_loader_v>, - "StreamLoader requires a std::istream config." - ); - static_assert( - !is_stream_loader_v>, - "StreamLoader requires a std::istream config." - ); - auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); - auto make = [&](auto distance_function) { - return ConcurrentDynamicVamana( - AssembleTag{}, - manager::as_typelist(), - index::vamana::concurrent::auto_dynamic_assemble( - std::forward(config_proto), - std::forward(graph_loader), - std::forward(data_loader), - std::move(distance_function), - std::move(threadpool), - debug_load_from_static - ) - ); - }; - if constexpr (std::is_same_v, DistanceType>) { - return DistanceDispatcher(distance)(make); - } else { - return make(std::move(distance)); - } + return make(std::move(distance)); } } - private: /// /// @brief Reload a ConcurrentDynamicVamana index saved with ``save(std::ostream&)``. /// /// Accepts both the native stream format and the directory-archive format. /// + /// @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 GraphLoader, - typename DataLoader, + typename Graph, + typename... GraphArgs, + typename Data, + typename... DataArgs, typename Distance, typename ThreadPoolProto> - static ConcurrentDynamicVamana assemble_from_stream( + static ConcurrentDynamicVamana assemble( std::istream& stream, - GraphLoader&& graph_loader, - DataLoader&& data_loader, + const StreamLoader& graph_loader, + const StreamLoader& data_loader, Distance distance, ThreadPoolProto threadpool_proto, - bool debug_load_from_static + bool debug_load_from_static = false ) { - using Graph = typename std::remove_cvref_t::return_type; - using Data = typename std::remove_cvref_t::return_type; static_assert( index::vamana::concurrent::is_segmented_blocked_v< typename Data::allocator_type>, diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h index 7e5cf08f1..fe5345264 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, @@ -374,50 +375,134 @@ class DynamicVamana : public manager::IndexManager { bool debug_load_from_static = false, svs::logging::logger_ptr logger = svs::logging::get() ) { - if constexpr (std::is_base_of_v>) { - static_assert( - is_stream_loader_v>, - "Assembling from a stream requires a StreamLoader for the graph." - ); - static_assert( - is_stream_loader_v>, - "Assembling from a stream requires a StreamLoader for the data." - ); - return assemble_from_stream( - config_proto, - graph_loader, - data_loader, + return DynamicVamana( + AssembleTag(), + manager::as_typelist(), + index::vamana::auto_dynamic_assemble( + std::forward(config_proto), + std::forward(graph_loader), + std::forward(data_loader), distance, - std::move(threadpool_proto), + threads::as_threadpool(std::move(threadpool_proto)), debug_load_from_static, std::move(logger) - ); + ) + ); + } + + // Assembly from stream + /// + /// @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 Graph The graph type to load. + /// @tparam GraphArgs Types of extra arguments forwarded to the graph loader. + /// @tparam Data The dataset type to load. + /// @tparam DataArgs Types of extra arguments forwarded to the data loader. + /// @tparam Distance Distance functor or ``svs::DistanceType`` enum. + /// @tparam ThreadPoolProto Thread pool type or size_t. + /// + /// @param stream Stream containing the serialized index. + /// @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 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 Graph, + typename... GraphArgs, + typename Data, + typename... DataArgs, + typename Distance, + typename ThreadPoolProto> + static DynamicVamana assemble( + std::istream& stream, + const StreamLoader& graph_loader, + const StreamLoader& data_loader, + const Distance& distance, + ThreadPoolProto threadpool_proto, + bool debug_load_from_static = false, + svs::logging::logger_ptr logger = svs::logging::get() + ) { + auto deserializer = svs::lib::detail::Deserializer::build(stream); + if (deserializer.is_native()) { + auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); + if constexpr (std::is_same_v, DistanceType>) { + auto dispatcher = DistanceDispatcher(distance); + return dispatcher([&](auto distance_function) { + return make_dynamic_vamana>( + index::vamana::auto_dynamic_assemble( + stream, + // lazy graph loader + [&]() -> Graph { return graph_loader.load(stream); }, + // lazy data loader + [&]() -> Data { return data_loader.load(stream); }, + distance_function, + std::move(threadpool), + debug_load_from_static, + logger + ) + ); + }); + } else { + return make_dynamic_vamana>( + index::vamana::auto_dynamic_assemble( + stream, + // lazy graph loader + [&]() -> Graph { return graph_loader.load(stream); }, + // lazy data loader + [&]() -> Data { return data_loader.load(stream); }, + distance, + std::move(threadpool), + debug_load_from_static, + logger + ) + ); + } } else { - static_assert( - !is_stream_loader_v>, - "A StreamLoader requires a std::istream config to assemble from." - ); - static_assert( - !is_stream_loader_v>, - "A StreamLoader requires a std::istream config to assemble from." - ); - return DynamicVamana( - AssembleTag(), - manager::as_typelist(), - index::vamana::auto_dynamic_assemble( - std::forward(config_proto), - std::forward(graph_loader), - std::forward(data_loader), - distance, - threads::as_threadpool(std::move(threadpool_proto)), - debug_load_from_static, - std::move(logger) - ) + namespace fs = std::filesystem; + lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; + lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); + + const auto config_path = tempdir.get() / "config"; + if (!fs::is_directory(config_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" + ); + } + + const auto graph_path = tempdir.get() / "graph"; + if (!fs::is_directory(graph_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" + ); + } + + const auto data_path = tempdir.get() / "data"; + if (!fs::is_directory(data_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); + } + + return assemble( + config_path, + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), + distance, + threads::as_threadpool(std::move(threadpool_proto)), + debug_load_from_static, + logger ); } } - // Assembly from stream + // Assembly from stream (released wrapper) /// /// @brief Assemble a DynamicVamana index from a serialized stream. /// @@ -525,93 +610,6 @@ class DynamicVamana : public manager::IndexManager { /// @brief Check if the index has been consolidated - no `Deleted` entries remain. bool is_consolidated() const { return impl_->is_consolidated(); } - - private: - template < - manager::QueryTypeDefinition QueryTypes, - typename GraphLoader, - typename DataLoader, - typename Distance, - typename ThreadPoolProto> - static DynamicVamana assemble_from_stream( - std::istream& stream, - const GraphLoader& graph_loader, - const DataLoader& data_loader, - const Distance& distance, - ThreadPoolProto threadpool_proto, - bool debug_load_from_static, - svs::logging::logger_ptr logger - ) { - using Data = typename DataLoader::return_type; - using GraphType = typename GraphLoader::return_type; - auto deserializer = svs::lib::detail::Deserializer::build(stream); - if (deserializer.is_native()) { - auto threadpool = threads::as_threadpool(std::move(threadpool_proto)); - if constexpr (std::is_same_v, DistanceType>) { - auto dispatcher = DistanceDispatcher(distance); - return dispatcher([&](auto distance_function) { - return make_dynamic_vamana>( - index::vamana::auto_dynamic_assemble( - stream, - // lazy graph loader - [&]() -> GraphType { return graph_loader.load(stream); }, - // lazy data loader - [&]() -> Data { return data_loader.load(stream); }, - distance_function, - std::move(threadpool), - debug_load_from_static, - logger - ) - ); - }); - } else { - return make_dynamic_vamana>( - index::vamana::auto_dynamic_assemble( - stream, - // lazy graph loader - [&]() -> GraphType { return graph_loader.load(stream); }, - // lazy data loader - [&]() -> Data { return data_loader.load(stream); }, - distance, - std::move(threadpool), - debug_load_from_static, - logger - ) - ); - } - } else { - namespace fs = std::filesystem; - lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; - lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); - - const auto config_path = tempdir.get() / "config"; - if (!fs::is_directory(config_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" - ); - } - - const auto graph_path = tempdir.get() / "graph"; - if (!fs::is_directory(graph_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" - ); - } - - const auto data_path = tempdir.get() / "data"; - if (!fs::is_directory(data_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); - } - - return assemble( - config_path, - SVS_LAZY(graph_loader.load(graph_path)), - data_loader.load(data_path), - distance, - threads::as_threadpool(std::move(threadpool_proto)), - debug_load_from_static, - logger - ); - } - } }; /// diff --git a/include/svs/orchestrators/vamana.h b/include/svs/orchestrators/vamana.h index bcd02865d..53a6985e6 100644 --- a/include/svs/orchestrators/vamana.h +++ b/include/svs/orchestrators/vamana.h @@ -460,70 +460,135 @@ class Vamana : public manager::IndexManager { const Distance& distance, ThreadPoolProto threadpool_proto, svs::logging::logger_ptr logger = svs::logging::get() - ) { - if constexpr (std::is_base_of_v< - std::istream, - std::remove_cvref_t>) { - static_assert( - is_stream_loader_v>, - "Assembling from a stream requires a StreamLoader for the graph." - ); - static_assert( - is_stream_loader_v>, - "Assembling from a stream requires a StreamLoader for the data." - ); - return assemble_from_stream( - config_proto, - graph_loader, - data_loader, - distance, - std::move(threadpool_proto), - std::move(logger) - ); - } else { - static_assert( - !is_stream_loader_v>, - "A StreamLoader requires a std::istream config to assemble from." - ); - static_assert( - !is_stream_loader_v>, - "A StreamLoader requires a std::istream config to assemble from." - ); - // 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)); - if constexpr (std::is_same_v) { - auto dispatcher = DistanceDispatcher(distance); - return dispatcher([&](auto distance_function) { - return make_vamana>( - AssembleTag(), - std::forward(config_proto), - graph_loader, - std::forward(data_loader), - distance_function, - std::move(threadpool), - std::move(logger) - ); - }); - } else { + ) + // 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)); + if constexpr (std::is_same_v) { + auto dispatcher = DistanceDispatcher(distance); + return dispatcher([&](auto distance_function) { return make_vamana>( AssembleTag(), std::forward(config_proto), graph_loader, std::forward(data_loader), - distance, + distance_function, std::move(threadpool), std::move(logger) ); - } + }); + } else { + return make_vamana>( + AssembleTag(), + std::forward(config_proto), + graph_loader, + std::forward(data_loader), + distance, + std::move(threadpool), + std::move(logger) + ); } } /// /// @brief Assemble a Vamana index from a stream. /// + /// @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 logger The logger to use for this index. Defaults to the global SVS logger + /// (``svs::logging::get()``). + /// + /// @copydoc threadpool_requirements + /// + /// @sa save, build + /// + template < + manager::QueryTypeDefinition QueryTypes, + typename Graph, + typename... GraphArgs, + typename Data, + typename... DataArgs, + typename Distance, + typename ThreadPoolProto> + static Vamana assemble( + std::istream& stream, + const StreamLoader& graph_loader, + const StreamLoader& data_loader, + const Distance& distance, + ThreadPoolProto threadpool_proto, + svs::logging::logger_ptr logger = svs::logging::get() + ) { + auto deserializer = svs::lib::detail::Deserializer::build(stream); + if (deserializer.is_native()) { + 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), + 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; + lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; + lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); + + const auto config_path = tempdir.get() / "config"; + if (!fs::is_directory(config_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" + ); + } + + const auto graph_path = tempdir.get() / "graph"; + if (!fs::is_directory(graph_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" + ); + } + + const auto data_path = tempdir.get() / "data"; + if (!fs::is_directory(data_path)) { + throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); + } + + return assemble( + config_path, + SVS_LAZY(graph_loader.load(graph_path)), + data_loader.load(data_path), + distance, + 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. @@ -710,75 +775,6 @@ class Vamana : public manager::IndexManager { } private: - template < - manager::QueryTypeDefinition QueryTypes, - typename GraphLoaderType, - typename DataLoader, - typename Distance, - typename ThreadPoolProto> - static Vamana assemble_from_stream( - std::istream& stream, - const GraphLoaderType& graph_loader, - const DataLoader& data_loader, - const Distance& distance, - ThreadPoolProto threadpool_proto, - svs::logging::logger_ptr logger - ) { - using Data = typename DataLoader::return_type; - using GraphType = typename GraphLoaderType::return_type; - - auto deserializer = svs::lib::detail::Deserializer::build(stream); - if (deserializer.is_native()) { - auto load_graph = [&]() -> GraphType { 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), - 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; - lib::UniqueTempDirectory tempdir{"svs_vamana_load"}; - lib::DirectoryArchiver::unpack(stream, tempdir, deserializer.magic()); - - const auto config_path = tempdir.get() / "config"; - if (!fs::is_directory(config_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing config directory!" - ); - } - - const auto graph_path = tempdir.get() / "graph"; - if (!fs::is_directory(graph_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing graph directory!" - ); - } - - const auto data_path = tempdir.get() / "data"; - if (!fs::is_directory(data_path)) { - throw ANNEXCEPTION("Invalid Vamana index archive: missing data directory!"); - } - - return assemble( - config_path, - SVS_LAZY(graph_loader.load(graph_path)), - data_loader.load(data_path), - distance, - threads::as_threadpool(std::move(threadpool_proto)), - std::move(logger) - ); - } - } - template < manager::QueryTypeDefinition QueryTypes, typename GraphLoaderFn, From bdf6a1bf27840944458004bfb73fb7cb7e87bbd6 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Wed, 14 Oct 2026 00:37:04 -0700 Subject: [PATCH 4/9] build(c-api): bump LTO shared library for StreamLoader The static-library C API job builds against the headers shipped in the prebuilt archive. The previous archive predates `svs::StreamLoader` and the StreamLoader assemble overloads, so the dispatchers failed to compile. Co-Authored-By: Claude Opus 5.5 --- bindings/c/CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 334b84706..f5835ca26 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -145,7 +145,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-14-0734.tar.gz" CACHE STRING "URL to download SVS shared library") else() # The fallback is correct but slower, so nothing downstream fails and CI From b880a31e250847e7d271da08f6e4901f7f4f76e5 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Wed, 14 Oct 2026 00:51:25 -0700 Subject: [PATCH 5/9] build(cpp-runtime): bump LTO shared library for StreamLoader The static-library C++ runtime job builds against the headers shipped in the prebuilt archive. The previous archive predates `svs::StreamLoader` and `Vamana::default_graph_type`, so the runtime impls failed to compile. Co-Authored-By: Claude Opus 5.5 --- bindings/cpp/CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bindings/cpp/CMakeLists.txt b/bindings/cpp/CMakeLists.txt index fffc8f037..e738fc412 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-14-0734.tar.gz" CACHE STRING "URL to download SVS shared library") else() if(SVS_REQUIRE_LTO_ARCHIVE) From acde405fd5a9ec1b5b31602869860b4a3b42c955 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Thu, 15 Oct 2026 00:17:03 -0700 Subject: [PATCH 6/9] next set of review comments --- bindings/c/src/dispatcher_dynamic_vamana.cpp | 7 ++- bindings/c/src/dispatcher_vamana.cpp | 7 ++- bindings/cpp/src/vamana_index_impl.h | 3 +- include/svs/core/loading.h | 13 +++++ .../orchestrators/concurrent_dynamic_vamana.h | 16 ++++--- include/svs/orchestrators/dynamic_vamana.h | 21 ++++---- include/svs/orchestrators/vamana.h | 18 +++---- tests/svs/core/loading.cpp | 48 +++++++++++++++++++ 8 files changed, 97 insertions(+), 36 deletions(-) diff --git a/bindings/c/src/dispatcher_dynamic_vamana.cpp b/bindings/c/src/dispatcher_dynamic_vamana.cpp index 407d366d6..31aba8d2b 100644 --- a/bindings/c/src/dispatcher_dynamic_vamana.cpp +++ b/bindings/c/src/dispatcher_dynamic_vamana.cpp @@ -120,10 +120,9 @@ svs::DynamicVamana load_stream_dynamic_vamana_index( // assemble copies data out; a view allocator here would leave the index dangling. return svs::DynamicVamana::assemble( *stream, - svs::StreamLoader< - svs::graphs::SimpleGraph, - decltype(graph_allocator)>{graph_allocator}, - svs::StreamLoader{allocator}, + 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 09d5e54d6..2a541dfd6 100644 --- a/bindings/c/src/dispatcher_vamana.cpp +++ b/bindings/c/src/dispatcher_vamana.cpp @@ -98,10 +98,9 @@ svs::Vamana load_stream_vamana_index( // assemble copies data out; a view allocator here would leave the index dangling. return svs::Vamana::assemble( *stream, - svs::StreamLoader< - svs::graphs::SimpleGraph, - decltype(graph_allocator)>{graph_allocator}, - svs::StreamLoader{data_allocator}, + svs::make_stream_loader< + svs::graphs::SimpleGraph>(graph_allocator), + svs::make_stream_loader(data_allocator), distance, std::move(pool) ); diff --git a/bindings/cpp/src/vamana_index_impl.h b/bindings/cpp/src/vamana_index_impl.h index c558c5049..07998fb4f 100644 --- a/bindings/cpp/src/vamana_index_impl.h +++ b/bindings/cpp/src/vamana_index_impl.h @@ -505,8 +505,7 @@ class VamanaIndexImpl { return new svs::Vamana(svs::Vamana::assemble( stream, svs::StreamLoader>{}, - svs::StreamLoader...>{ - std::forward(args)...}, + svs::make_stream_loader(std::forward(args)...), to_svs_distance(metric), std::move(threadpool) )); diff --git a/include/svs/core/loading.h b/include/svs/core/loading.h index 28b56b848..4ab117b4f 100644 --- a/include/svs/core/loading.h +++ b/include/svs/core/loading.h @@ -23,6 +23,7 @@ #include #include #include +#include #include namespace svs { @@ -60,6 +61,18 @@ template struct StreamLoader { 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 diff --git a/include/svs/orchestrators/concurrent_dynamic_vamana.h b/include/svs/orchestrators/concurrent_dynamic_vamana.h index cb914021d..3ec385b91 100644 --- a/include/svs/orchestrators/concurrent_dynamic_vamana.h +++ b/include/svs/orchestrators/concurrent_dynamic_vamana.h @@ -170,20 +170,22 @@ class ConcurrentDynamicVamana : public DynamicVamana { /// template < manager::QueryTypeDefinition QueryTypes, - typename Graph, - typename... GraphArgs, - typename Data, - typename... DataArgs, + typename GraphLoader, + typename DataLoader, typename Distance, typename ThreadPoolProto> static ConcurrentDynamicVamana assemble( std::istream& stream, - const StreamLoader& graph_loader, - const StreamLoader& data_loader, + const GraphLoader& graph_loader, + const DataLoader& data_loader, Distance distance, ThreadPoolProto threadpool_proto, 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>, diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h index fe5345264..b30477d91 100644 --- a/include/svs/orchestrators/dynamic_vamana.h +++ b/include/svs/orchestrators/dynamic_vamana.h @@ -399,10 +399,8 @@ class DynamicVamana : public manager::IndexManager { /// /// @tparam QueryTypes The set of query element types supported by the resulting /// index. - /// @tparam Graph The graph type to load. - /// @tparam GraphArgs Types of extra arguments forwarded to the graph loader. - /// @tparam Data The dataset type to load. - /// @tparam DataArgs Types of extra arguments forwarded to the data loader. + /// @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. /// @@ -418,21 +416,22 @@ class DynamicVamana : public manager::IndexManager { /// template < manager::QueryTypeDefinition QueryTypes, - typename Graph, - typename... GraphArgs, - typename Data, - typename... DataArgs, + typename GraphLoader, + typename DataLoader, typename Distance, typename ThreadPoolProto> + requires(is_stream_loader_v && is_stream_loader_v) static DynamicVamana assemble( std::istream& stream, - const StreamLoader& graph_loader, - const StreamLoader& data_loader, + const GraphLoader& graph_loader, + const DataLoader& data_loader, const Distance& distance, ThreadPoolProto threadpool_proto, 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)); @@ -502,7 +501,7 @@ class DynamicVamana : public manager::IndexManager { } } - // Assembly from stream (released wrapper) + // Assembly from stream wrapper /// /// @brief Assemble a DynamicVamana index from a serialized stream. /// diff --git a/include/svs/orchestrators/vamana.h b/include/svs/orchestrators/vamana.h index 53a6985e6..5787e8a7a 100644 --- a/include/svs/orchestrators/vamana.h +++ b/include/svs/orchestrators/vamana.h @@ -520,20 +520,22 @@ class Vamana : public manager::IndexManager { /// template < manager::QueryTypeDefinition QueryTypes, - typename Graph, - typename... GraphArgs, - typename Data, - typename... DataArgs, + typename GraphLoader, + typename DataLoader, typename Distance, typename ThreadPoolProto> static Vamana assemble( std::istream& stream, - const StreamLoader& graph_loader, - const StreamLoader& data_loader, + const GraphLoader& graph_loader, + const DataLoader& data_loader, const Distance& distance, ThreadPoolProto threadpool_proto, 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()) { auto load_graph = [&]() -> Graph { return graph_loader.load(stream); }; @@ -619,7 +621,7 @@ class Vamana : public manager::IndexManager { return assemble( stream, StreamLoader>{}, - StreamLoader...>{SVS_FWD(data_args)...}, + make_stream_loader(SVS_FWD(data_args)...), distance, std::move(threadpool_proto) ); 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); + } +} From f990121e1a2499c57bf4a9cbccaa912b66feb76b Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Thu, 15 Oct 2026 06:04:41 -0700 Subject: [PATCH 7/9] build(c-api): bump LTO shared library for make_stream_loader The static-library C API job builds against the headers shipped in the prebuilt archive. The previous archive predates `svs::make_stream_loader`, which the stream dispatchers now call. Co-Authored-By: Claude Opus 5.5 --- bindings/c/CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 6fc574a1a..736ba1734 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -146,7 +146,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-14-0734.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 From 4ae0098c76c0d7de53d6b664cf4acaf8e767efd5 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Thu, 15 Oct 2026 06:04:43 -0700 Subject: [PATCH 8/9] build(cpp-runtime): bump LTO shared library for make_stream_loader The static-library C++ runtime job builds against the headers shipped in the prebuilt archive. The previous archive predates `svs::make_stream_loader`, which the runtime impls now call. Co-Authored-By: Claude Opus 5.5 --- bindings/cpp/CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bindings/cpp/CMakeLists.txt b/bindings/cpp/CMakeLists.txt index e738fc412..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-14-0734.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) From dc3287b7277db42a4d6ccd2186563f9fb9555111 Mon Sep 17 00:00:00 2001 From: Andreas Huber Date: Fri, 16 Oct 2026 04:46:14 -0700 Subject: [PATCH 9/9] Use make_stream_loader() factory --- include/svs/orchestrators/dynamic_vamana.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/svs/orchestrators/dynamic_vamana.h b/include/svs/orchestrators/dynamic_vamana.h index b30477d91..067c7c486 100644 --- a/include/svs/orchestrators/dynamic_vamana.h +++ b/include/svs/orchestrators/dynamic_vamana.h @@ -534,7 +534,7 @@ class DynamicVamana : public manager::IndexManager { return assemble( stream, StreamLoader{}, - StreamLoader...>{SVS_FWD(data_args)...}, + make_stream_loader(SVS_FWD(data_args)...), distance, std::move(threadpool_proto) );