Native serialization for Indexes. (#300)

This PR adds native stream serialization to all SVS index types, as an
alternative to the existing (legacy) directory-based serialization. It
allow to avoid filesystem round-trips of the data. The native
serialization doesn't require from the stream to be seek able, so no
additional restrictions were introduced.

See the following PR for details:
#280,
#281,
#285,
#286,
#289,
#292,
#294,
#296,
#299

Main changes are:
1. A CRTP base `Archiver` extracts binary I/O primitives (`write_size`,
`read_size`, `write_name`, `read_name`, `read_from_istream`) from
`DirectoryArchiver`. `DirectoryArchiver` and new `StreamArchiver` class
inherit from `Archiver`. `StreamArchiver` has its own magic number
("SVS_STRM") to distinguish native streams from directory archives.
2. The monolithic `Writer` is split via CRTP with two derived classes:
`FileWriter` owns an `std::ofstream`, writes a header, flushes on
destructor, `StreamWriter` wraps an external `std::ostream&`, no
header/lifecycle management. This allows `io::save(data, os)` to write
vector data directly to any stream.
3. The `save(stream)` in orchestrator `Impl` classes no longer does
temp-dir->pack. Instead it directly calls `impl().save(stream)`.
4. The dispatching between new (native) and old (legacy) deserialization
is made at the orchestrators. `Deserializer::build(is)` reads the magic
number, exposes `is_native()` to choose path.

---------

Co-authored-by: Dmitry Razdoburdin <drazdobu@intel.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Rafik Saliev <rafik.f.saliev@intel.com>
Co-authored-by: ethanglaser <42726565+ethanglaser@users.noreply.github.com>

Author: 5 people

bindings/cpp/src/dynamic_vamana_index_impl.h

@@ -316,15 +316,7 @@ class DynamicVamanaIndexImpl {
                 ErrorCode::NOT_INITIALIZED, "Cannot serialize: SVS index not initialized."};
         }
 
-        lib::UniqueTempDirectory tempdir{"svs_vamana_save"};
-        const auto config_dir = tempdir.get() / "config";
-        const auto graph_dir = tempdir.get() / "graph";
-        const auto data_dir = tempdir.get() / "data";
-        std::filesystem::create_directories(config_dir);
-        std::filesystem::create_directories(graph_dir);
-        std::filesystem::create_directories(data_dir);
-        impl_->save(config_dir, graph_dir, data_dir);
-        lib::DirectoryArchiver::pack(tempdir, out);
+        impl_->save(out);
     }
 
   protected:
@@ -469,49 +461,32 @@ class DynamicVamanaIndexImpl {
             impl_->get_full_search_history()};
     }
 
-    template <typename Tag>
-    static svs::DynamicVamana*
-    load_impl_t(Tag&& tag, std::istream& stream, MetricType metric) {
-        namespace fs = std::filesystem;
-        lib::UniqueTempDirectory tempdir{"svs_vamana_load"};
-        lib::DirectoryArchiver::unpack(stream, tempdir);
-
-        const auto config_path = tempdir.get() / "config";
-        if (!fs::is_directory(config_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing config directory!"};
-        }
-
-        const auto graph_path = tempdir.get() / "graph";
-        if (!fs::is_directory(graph_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing graph directory!"};
-        }
-
-        const auto data_path = tempdir.get() / "data";
-        if (!fs::is_directory(data_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing data directory!"};
+    template <StorageKind Kind, typename Alloc>
+    static svs::DynamicVamana* load_impl_t(
+        storage::StorageType<Kind, Alloc>&& SVS_UNUSED(tag),
+        std::istream& stream,
+        MetricType metric
+    ) {
+        if constexpr (!storage::is_supported_storage_kind_v<Kind>) {
+            throw StatusException(
+                ErrorCode::NOT_IMPLEMENTED, "Requested storage kind is not supported"
+            );
+        } else {
+            using storage_type = storage::StorageType_t<Kind, Alloc>;
+            auto threadpool = default_threadpool();
+
+            svs::DistanceDispatcher distance_dispatcher(to_svs_distance(metric));
+
+            return distance_dispatcher([&](auto&& distance) {
+                return new svs::DynamicVamana(
+                    svs::DynamicVamana::assemble<float, storage_type>(
+                        stream,
+                        std::forward<decltype(distance)>(distance),
+                        std::move(threadpool)
+                    )
+                );
+            });
         }
-
-        auto storage = storage::load_storage(std::forward<Tag>(tag), data_path);
-        auto threadpool = default_threadpool();
-
-        svs::DistanceDispatcher distance_dispatcher(to_svs_distance(metric));
-
-        return distance_dispatcher([&](auto&& distance) {
-            return new svs::DynamicVamana(svs::DynamicVamana::assemble<float>(
-                config_path,
-                svs::GraphLoader{graph_path},
-                std::move(storage),
-                std::forward<decltype(distance)>(distance),
-                std::move(threadpool),
-                false
-            ));
-        });
     }
 
   public:

bindings/cpp/src/svs_runtime_utils.h

@@ -203,6 +203,11 @@ template <typename Alloc> struct StorageType<StorageKind::SQI8, Alloc> {
     using type = SQDatasetType<std::int8_t, allocator_type>;
 };
 
+template <StorageKind Kind>
+inline constexpr bool is_supported_storage_kind_v = !std::is_same_v<
+    typename StorageType<Kind, svs::lib::Allocator<float>>::type,
+    UnsupportedStorageType<svs::lib::Allocator<float>>>;
+
 // Storage factory
 template <typename T> struct StorageFactory;
 
@@ -222,14 +227,6 @@ template <typename Alloc> struct StorageFactory<UnsupportedStorageType<Alloc>> {
             ErrorCode::NOT_IMPLEMENTED, "Requested storage kind is not supported"
         );
     }
-
-    template <typename... Args>
-    static StorageType
-    load(const std::filesystem::path& SVS_UNUSED(path), Args&&... SVS_UNUSED(args)) {
-        throw StatusException(
-            ErrorCode::NOT_IMPLEMENTED, "Requested storage kind is not supported"
-        );
-    }
 };
 
 template <typename T, size_t Extent, typename Alloc>
@@ -254,11 +251,6 @@ struct StorageFactory<svs::data::SimpleData<T, Extent, Alloc>> {
         );
         return result;
     }
-
-    template <typename... Args>
-    static StorageType load(const std::filesystem::path& path, Args&&... args) {
-        return svs::lib::load_from_disk<StorageType>(path, SVS_FWD(args)...);
-    }
 };
 
 // SQ Storage support
@@ -274,11 +266,6 @@ struct StorageFactory<svs::quantization::scalar::SQDataset<T, Extent, Alloc>> {
     ) {
         return StorageType::compress(data, pool, alloc);
     }
-
-    template <typename... Args>
-    static StorageType load(const std::filesystem::path& path, Args&&... args) {
-        return svs::lib::load_from_disk<StorageType>(path, SVS_FWD(args)...);
-    }
 };
 
 // LVQ Storage support
@@ -327,11 +314,6 @@ struct StorageFactory<LVQStorageType> {
     ) {
         return StorageType::compress(data, pool, 0, alloc);
     }
-
-    template <typename... Args>
-    static StorageType load(const std::filesystem::path& path, Args&&... args) {
-        return svs::lib::load_from_disk<StorageType>(path, SVS_FWD(args)...);
-    }
 };
 
 // LeanVec Storage support
@@ -381,11 +363,6 @@ struct StorageFactory<LeanVecStorageType> {
             data, std::move(matrices), pool, 0, svs::lib::MaybeStatic{leanvec_d}, alloc
         );
     }
-
-    template <typename... Args>
-    static StorageType load(const std::filesystem::path& path, Args&&... args) {
-        return svs::lib::load_from_disk<StorageType>(path, SVS_FWD(args)...);
-    }
 };
 #endif // SVS_RUNTIME_HAVE_LVQ_LEANVEC
 
@@ -394,11 +371,6 @@ auto make_storage(StorageType<Kind, Alloc> SVS_UNUSED(tag), Args&&... args) {
     return StorageFactory<StorageType_t<Kind, Alloc>>::init(std::forward<Args>(args)...);
 }
 
-template <StorageKind Kind, typename Alloc, typename... Args>
-auto load_storage(StorageType<Kind, Alloc> SVS_UNUSED(tag), Args&&... args) {
-    return StorageFactory<StorageType_t<Kind, Alloc>>::load(std::forward<Args>(args)...);
-}
-
 template <typename Alloc, typename F, typename... Args>
 auto dispatch_storage_kind(StorageKind kind, F&& f, Args&&... args) {
     if (!is_supported_storage_kind(kind)) {

bindings/cpp/src/vamana_index_impl.h

@@ -271,17 +271,7 @@ class VamanaIndexImpl {
 
     void reset() { impl_.reset(); }
 
-    void save(std::ostream& out) const {
-        lib::UniqueTempDirectory tempdir{"svs_vamana_save"};
-        const auto config_dir = tempdir.get() / "config";
-        const auto graph_dir = tempdir.get() / "graph";
-        const auto data_dir = tempdir.get() / "data";
-        std::filesystem::create_directories(config_dir);
-        std::filesystem::create_directories(graph_dir);
-        std::filesystem::create_directories(data_dir);
-        get_impl()->save(config_dir, graph_dir, data_dir);
-        lib::DirectoryArchiver::pack(tempdir, out);
-    }
+    void save(std::ostream& out) const { get_impl()->save(out); }
 
   protected:
     // Utility functions
@@ -420,47 +410,24 @@ class VamanaIndexImpl {
         }
     }
 
-    template <typename Tag>
-    static svs::Vamana* load_impl_t(Tag&& tag, std::istream& stream, MetricType metric) {
-        namespace fs = std::filesystem;
-        lib::UniqueTempDirectory tempdir{"svs_vamana_load"};
-        lib::DirectoryArchiver::unpack(stream, tempdir);
-
-        const auto config_path = tempdir.get() / "config";
-        if (!fs::is_directory(config_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing config directory!"};
-        }
-
-        const auto graph_path = tempdir.get() / "graph";
-        if (!fs::is_directory(graph_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing graph directory!"};
-        }
-
-        const auto data_path = tempdir.get() / "data";
-        if (!fs::is_directory(data_path)) {
-            throw StatusException{
-                ErrorCode::RUNTIME_ERROR,
-                "Invalid Vamana index archive: missing data directory!"};
-        }
-
-        auto storage = storage::load_storage(std::forward<Tag>(tag), data_path);
-        auto threadpool = default_threadpool();
-
-        svs::DistanceDispatcher distance_dispatcher(to_svs_distance(metric));
+    template <StorageKind Kind, typename Alloc>
+    static svs::Vamana* load_impl_t(
+        storage::StorageType<Kind, Alloc>&& SVS_UNUSED(tag),
+        std::istream& stream,
+        MetricType metric
+    ) {
+        if constexpr (!storage::is_supported_storage_kind_v<Kind>) {
+            throw StatusException(
+                ErrorCode::NOT_IMPLEMENTED, "Requested storage kind is not supported"
+            );
+        } else {
+            using storage_type = storage::StorageType_t<Kind, Alloc>;
+            auto threadpool = default_threadpool();
 
-        return distance_dispatcher([&](auto&& distance) {
-            return new svs::Vamana(svs::Vamana::assemble<float>(
-                config_path,
-                svs::GraphLoader{graph_path},
-                std::move(storage),
-                std::forward<decltype(distance)>(distance),
-                std::move(threadpool)
+            return new svs::Vamana(svs::Vamana::assemble<float, storage_type>(
+                stream, to_svs_distance(metric), std::move(threadpool)
             ));
-        });
+        }
     }
 
   public:

include/svs/core/data/io.h

@@ -79,6 +79,22 @@ void populate_impl(
     }
 }
 
+template <data::MemoryDataset Data> void populate(std::istream& is, Data& data) {
+    auto accessor = DefaultWriteAccessor();
+
+    size_t num_vectors = data.size();
+    size_t dims = data.dimensions();
+
+    auto max_lines = Dynamic;
+    auto nvectors = std::min(num_vectors, max_lines);
+
+    auto reader = lib::VectorReader<typename Data::element_type>(dims);
+    for (size_t i = 0; i < nvectors; ++i) {
+        reader.read(is);
+        accessor.set(data, i, reader.data());
+    }
+}
+
 // Intercept the native file to perform dispatch on the actual file type.
 template <data::MemoryDataset Data, typename WriteAccessor>
 void populate_impl(
@@ -120,6 +136,15 @@ void save(const Dataset& data, const File& file, const lib::UUID& uuid = lib::Ze
     return save(data, accessor, file, uuid);
 }
 
+template <data::ImmutableMemoryDataset Dataset>
+void save(const Dataset& data, std::ostream& os) {
+    auto accessor = DefaultReadAccessor();
+    auto writer = svs::io::v1::StreamWriter<void>(os);
+    for (size_t i = 0; i < data.size(); ++i) {
+        writer << accessor.get(data, i);
+    }
+}
+
 ///
 /// @brief Save the dataset as a "*vecs" file.
 ///
@@ -169,6 +194,14 @@ lib::lazy_result_t<F, size_t, size_t> load_dataset(const File& file, const F& la
     return load_impl(detail::to_native(file), default_accessor, lazy);
 }
 
+template <lib::LazyInvocable<size_t, size_t> F>
+lib::lazy_result_t<F, size_t, size_t>
+load_dataset(std::istream& is, const F& lazy, size_t num_vectors, size_t dims) {
+    auto data = lazy(num_vectors, dims);
+    populate(is, data);
+    return data;
+}
+
 // Return whether or not a file is directly loadable via file-extension.
 inline bool special_by_file_extension(std::string_view path) {
     return (path.ends_with("svs") || path.ends_with("vecs") || path.ends_with("bin"));