Repository: antgroup/vsag (language: python)

Issue:
Add a new Layer "DataCell" to manage data access 
I think high level index should ignore the specific implement of quantization and io type, So the quantization and io components can be managed by new layer DataCell which provide different data access interface to the index layer. just like the following show

|Layer| Description|Implement|
|-------|--------|---------|
|Index Layer | ANN algorithm & architecture.          |                       Graph (HNSW,NNDescent) + Reorder,  IVF(PQ,SQ,fastscan), HNSW_IVF, MultiIndex ......|
|DataCell Layer| data access, use quantization, io and others    | FlattenDC, GraphNeighborhoodDC, MixtureDC .......|
|Operator Layer| quantizations, IO and other basic components                                                 | PQ,SQ,MemoryIO,SSDIO,......|

this abstract layered approach may facilitate scalability and code reuse
currently we only have index layer, so the issue will provide Datacell Layer and Operator Layer in this feature.
the related pull requests:

- #25 
- #34 
- #37 
- #41 
- #35 
- #59 
- #60 
- #61 
- #62 
- #79 
- #80 
- #96 
- #111 
- #114 
- #125 
todo tasks:
- support mixture DataCell
- adapt Graph DataCell for HNSW
- implement HNSW + SQ4 + SQ8 Index 




Discussion / hints:

Fix patch:
diff --git a/include/vsag/constants.h b/include/vsag/constants.h
index 86b8a832..32889a2a 100644
--- a/include/vsag/constants.h
+++ b/include/vsag/constants.h
@@ -16,6 +16,7 @@
 
 namespace vsag {
 
+extern const char* const INDEX_HGRAPH;
 extern const char* const INDEX_DISKANN;
 extern const char* const INDEX_HNSW;
 extern const char* const INDEX_FRESH_HNSW;
@@ -91,4 +92,37 @@ extern const char* const MESSAGE_PARAMETER;
 extern const char* const SERIALIZE_MAGIC_NUM;
 extern const char* const SERIALIZE_VERSION;
 
+// Index Type
+extern const char* const INDEX_TYPE_HGRAPH;
+
+// Parameter key for hgraph
+extern const char* const HGRAPH_USE_REORDER_KEY;
+extern const char* const HGRAPH_GRAPH_KEY;
+extern const char* const HGRAPH_BASE_CODES_KEY;
+extern const char* const HGRAPH_PRECISE_CODES_KEY;
+
+// IO type
+extern const char* const IO_TYPE_KEY;
+extern const char* const IO_TYPE_VALUE_MEMORY_IO;
+extern const char* const IO_TYPE_VALUE_BLOCK_MEMORY_IO;
+
+// IO param key
+extern const char* const IO_PARAMS_KEY;
+extern const char* const BLOCK_IO_BLOCK_SIZE_KEY;
+
+// quantization type
+extern const char* const QUANTIZATION_TYPE_KEY;
+extern const char* const QUANTIZATION_TYPE_VALUE_SQ8;
+extern const char* const QUANTIZATION_TYPE_VALUE_FP32;
+
+// quantization param key
+extern const char* const QUANTIZATION_PARAMS_KEY;
+
+// graph param key
+extern const char* const GRAPH_PARAMS_KEY;
+
+// graph param value
+extern const char* const GRAPH_PARAM_MAX_DEGREE;
+extern const char* const GRAPH_PARAM_INIT_MAX_CAPACITY;
+
 }  // namespace vsag
diff --git a/src/constants.cpp b/src/constants.cpp
index bc781994..aec320f9 100644
--- a/src/constants.cpp
+++ b/src/constants.cpp
@@ -17,6 +17,7 @@
 
 namespace vsag {
 
+const char* const INDEX_HGRAPH = "hgraph";
 const char* const INDEX_DISKANN = "diskann";
 const char* const INDEX_HNSW = "hnsw";
 const char* const INDEX_FRESH_HNSW = "fresh_hnsw";
@@ -92,4 +93,37 @@ const char* const MESSAGE_PARAMETER = "invalid parameter";
 const char* const SERIALIZE_MAGIC_NUM = "MAGIC_NUM";
 const char* const SERIALIZE_VERSION = "VERSION";
 
+// Index Type
+const char* const INDEX_TYPE_HGRAPH = "hgraph";
+
+// Parameter key for hgraph
+const char* const HGRAPH_USE_REORDER_KEY = "use_reorder";
+const char* const HGRAPH_GRAPH_KEY = "graph";
+const char* const HGRAPH_BASE_CODES_KEY = "base_codes";
+const char* const HGRAPH_PRECISE_CODES_KEY = "precise_codes";
+
+// IO type
+const char* const IO_TYPE_KEY = "io_type";
+const char* const IO_TYPE_VALUE_MEMORY_IO = "memory";
+const char* const IO_TYPE_VALUE_BLOCK_MEMORY_IO = "block_memory";
+
+// IO param key
+const char* const IO_PARAMS_KEY = "io_params";
+const char* const BLOCK_IO_BLOCK_SIZE_KEY = "block_size";
+
+// quantization type
+const char* const QUANTIZATION_TYPE_KEY = "quantization_type";
+const char* const QUANTIZATION_TYPE_VALUE_SQ8 = "sq8";
+const char* const QUANTIZATION_TYPE_VALUE_FP32 = "fp32";
+
+// quantization params key
+const char* const QUANTIZATION_PARAMS_KEY = "quantization_params";
+
+// graph param key
+const char* const GRAPH_PARAMS_KEY = "graph_params";
+
+// graph param value
+const char* const GRAPH_PARAM_MAX_DEGREE = "max_degree";
+const char* const GRAPH_PARAM_INIT_MAX_CAPACITY = "init_capacity";
+
 };  // namespace vsag
diff --git a/src/data_type.h b/src/data_type.h
new file mode 100644
index 00000000..39aa5802
--- /dev/null
+++ b/src/data_type.h
@@ -0,0 +1,21 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+namespace vsag {
+enum class DataTypes { DATA_TYPE_FLOAT = 0, DATA_TYPE_INT8 = 1, DATA_TYPE_FP16 = 2 };
+
+}  // namespace vsag
diff --git a/src/index/index_common_param.h b/src/index/index_common_param.h
new file mode 100644
index 00000000..ffc6b57f
--- /dev/null
+++ b/src/index/index_common_param.h
@@ -0,0 +1,32 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <cstdint>
+
+#include "data_type.h"
+#include "metric_type.h"
+#include "vsag/allocator.h"
+
+namespace vsag {
+class IndexCommonParam {
+public:
+    MetricType metric_{MetricType::METRIC_TYPE_L2SQR};
+    DataTypes data_type_{DataTypes::DATA_TYPE_FLOAT};
+    int64_t dim_{0};
+    Allocator* allocator_{nullptr};
+};
+}  // namespace vsag
diff --git a/src/io/basic_io.h b/src/io/basic_io.h
index 1cc757cb..df501da8 100644
--- a/src/io/basic_io.h
+++ b/src/io/basic_io.h
@@ -17,6 +17,9 @@
 
 #include <cstdint>
 
+#include "stream_reader.h"
+#include "stream_writer.h"
+
 namespace vsag {
 
 template <typename IOTmpl>
@@ -24,7 +27,7 @@ class BasicIO {
 public:
     BasicIO<IOTmpl>() = default;
 
-    ~BasicIO() = default;
+    virtual ~BasicIO() = default;
 
     inline void
     Write(const uint8_t* data, uint64_t size, uint64_t offset) {
@@ -51,6 +54,16 @@ class BasicIO {
         return cast().PrefetchImpl(offset, cache_line);
     }
 
+    inline void
+    Serialize(StreamWriter& writer) {
+        return cast().SerializeImpl(writer);
+    }
+
+    inline void
+    Deserialize(StreamReader& reader) {
+        return cast().DeserializeImpl(reader);
+    }
+
 private:
     inline IOTmpl&
     cast() {
diff --git a/src/io/io_headers.h b/src/io/io_headers.h
new file mode 100644
index 00000000..4639735a
--- /dev/null
+++ b/src/io/io_headers.h
@@ -0,0 +1,20 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "basic_io.h"
+#include "memory_block_io.h"
+#include "memory_io.h"
diff --git a/src/io/memory_block_io.h b/src/io/memory_block_io.h
new file mode 100644
index 00000000..3e045254
--- /dev/null
+++ b/src/io/memory_block_io.h
@@ -0,0 +1,218 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#if defined(ENABLE_SSE)
+#include <xmmintrin.h>  //todo
+#endif
+
+#include <algorithm>
+#include <cmath>
+#include <cstring>
+#include <nlohmann/json.hpp>
+#include <stdexcept>
+#include <vector>
+
+#include "basic_io.h"
+#include "common.h"
+#include "index/index_common_param.h"
+#include "vsag/allocator.h"
+#include "vsag/constants.h"
+
+namespace vsag {
+
+class MemoryBlockIO : public BasicIO<MemoryBlockIO> {
+public:
+    explicit MemoryBlockIO(Allocator* allocator, uint64_t block_size = DEFAULT_BLOCK_SIZE)
+        : block_size_(block_size), allocator_(allocator) {
+    }
+
+    MemoryBlockIO(const nlohmann::json& io_obj, const IndexCommonParam& common_param)
+        : allocator_(common_param.allocator_) {
+        CHECK_ARGUMENT(
+            io_obj.contains(BLOCK_IO_BLOCK_SIZE_KEY),
+            fmt::format("memory block io parameters must contains {}", BLOCK_IO_BLOCK_SIZE_KEY));
+
+        this->block_size_ = io_obj[BLOCK_IO_BLOCK_SIZE_KEY];  // TODO(LHT): trans str to uint64_t
+    }
+
+    ~MemoryBlockIO() override {
+        for (auto* block : blocks_) {
+            allocator_->Deallocate(block);
+        }
+    }
+
+    inline void
+    WriteImpl(const uint8_t* data, uint64_t size, uint64_t offset);
+
+    inline bool
+    ReadImpl(uint8_t* data, uint64_t size, uint64_t offset) const;
+
+    [[nodiscard]] inline const uint8_t*
+    ReadImpl(uint64_t size, uint64_t offset) const;
+
+    inline bool
+    MultiReadImpl(uint8_t* datas, uint64_t* sizes, uint64_t* offsets, uint64_t count) const;
+
+    inline void
+    PrefetchImpl(uint64_t offset, uint64_t cache_line = 64);
+
+    inline void
+    SerializeImpl(StreamWriter& writer);
+
+    inline void
+    DeserializeImpl(StreamReader& reader);
+
+private:
+    [[nodiscard]] inline bool
+    check_valid_offset(uint64_t size) const {
+        return size <= blocks_.size() * block_size_;
+    }
+
+    inline void
+    check_and_realloc(uint64_t size);
+
+    [[nodiscard]] inline const uint8_t*
+    get_data_ptr(uint64_t offset) const {
+        auto block_no = offset / block_size_;
+        auto block_off = offset % block_size_;
+        return blocks_[block_no] + block_off;
+    }
+
+    [[nodiscard]] inline bool
+    check_in_one_block(uint64_t off1, uint64_t off2) const {
+        return (off1 / block_size_) == (off2 / block_size_);
+    }
+
+private:
+    uint64_t block_size_{DEFAULT_BLOCK_SIZE};
+
+    std::vector<uint8_t*> blocks_{};
+
+    Allocator* const allocator_{nullptr};
+
+    static const uint64_t DEFAULT_BLOCK_SIZE = 128 * 1024 * 1024;  // 128MB
+};
+
+void
+MemoryBlockIO::WriteImpl(const uint8_t* data, uint64_t size, uint64_t offset) {
+    check_and_realloc(size + offset);
+    uint64_t cur_size = 0;
+    auto start_no = offset / block_size_;
+    auto start_off = offset % block_size_;
+    auto max_size = block_size_ - start_off;
+    while (cur_size < size) {
+        uint8_t* cur_write = blocks_[start_no] + start_off;
+        auto cur_length = std::min(size - cur_size, max_size);
+        memcpy(cur_write, data + cur_size, cur_length);
+        cur_size += cur_length;
+        max_size = block_size_;
+        ++start_no;
+        start_off = 0;
+    }
+}
+
+bool
+MemoryBlockIO::ReadImpl(uint8_t* data, uint64_t size, uint64_t offset) const {
+    bool ret = check_valid_offset(size + offset);
+    if (ret) {
+        uint64_t cur_size = 0;
+        auto start_no = offset / block_size_;
+        auto start_off = offset % block_size_;
+        auto max_size = block_size_ - start_off;
+        while (cur_size < size) {
+            const uint8_t* cur_write = blocks_[start_no] + start_off;
+            auto cur_length = std::min(size - cur_size, max_size);
+            memcpy(data + cur_size, cur_write, cur_length);
+            cur_size += cur_length;
+            max_size = block_size_;
+            ++start_no;
+            start_off = 0;
+        }
+    }
+    return ret;
+}
+
+const uint8_t*
+MemoryBlockIO::ReadImpl(uint64_t size, uint64_t offset) const {
+    if (check_valid_offset(size + offset)) {
+        if (check_in_one_block(offset, size + offset)) {
+            return this->get_data_ptr(offset);
+        } else {
+            throw std::runtime_error(
+                "Memory Block IO don't support read cross-block data without passing data "
+                "pointer");
+        }
+    }
+    return nullptr;
+}
+bool
+MemoryBlockIO::MultiReadImpl(uint8_t* datas,
+                             uint64_t* sizes,
+                             uint64_t* offsets,
+                             uint64_t count) const {
+    bool ret = true;
+    for (uint64_t i = 0; i < count; ++i) {
+        ret &= this->ReadImpl(datas, sizes[i], offsets[i]);
+        datas += sizes[i];
+    }
+    return ret;
+}
+void
+MemoryBlockIO::PrefetchImpl(uint64_t offset, uint64_t cache_line) {
+#if defined(ENABLE_SSE)
+    _mm_prefetch(get_data_ptr(offset), _MM_HINT_T0);  // todo
+#endif
+}
+
+void
+MemoryBlockIO::check_and_realloc(uint64_t size) {
+    if (check_valid_offset(size)) {
+        return;
+    }
+    const uint64_t new_block_count = (size + this->block_size_ - 1) / block_size_;
+    auto cur_block_size = this->blocks_.size();
+    while (cur_block_size < new_block_count) {
+        this->blocks_.emplace_back((uint8_t*)(allocator_->Allocate(block_size_)));
+        ++cur_block_size;
+    }
+}
+void
+MemoryBlockIO::SerializeImpl(StreamWriter& writer) {
+    StreamWriter::WriteObj(writer, this->block_size_);
+    uint64_t block_count = this->blocks_.size();
+    StreamWriter::WriteObj(writer, block_count);
+    for (uint64_t i = 0; i < block_count; ++i) {
+        writer.Write(reinterpret_cast<char*>(this->blocks_[i]), block_size_);
+    }
+}
+
+void
+MemoryBlockIO::DeserializeImpl(StreamReader& reader) {
+    for (auto* block : blocks_) {
+        allocator_->Deallocate(block);
+    }
+    StreamReader::ReadObj(reader, this->block_size_);
+    uint64_t block_count;
+    StreamReader::ReadObj(reader, block_count);
+    this->blocks_.resize(block_count);
+    for (uint64_t i = 0; i < block_count; ++i) {
+        blocks_[i] = static_cast<unsigned char*>(allocator_->Allocate(this->block_size_));
+        reader.Read(reinterpret_cast<char*>(blocks_[i]), block_size_);
+    }
+}
+
+}  // namespace vsag
diff --git a/src/io/memory_io.h b/src/io/memory_io.h
index 39741da3..81d3434f 100644
--- a/src/io/memory_io.h
+++ b/src/io/memory_io.h
@@ -20,8 +20,10 @@
 #endif
 
 #include <cstring>
+#include <nlohmann/json.hpp>
 
 #include "basic_io.h"
+#include "index/index_common_param.h"
 #include "vsag/allocator.h"
 
 namespace vsag {
@@ -33,6 +35,12 @@ class MemoryIO : public BasicIO<MemoryIO> {
         current_size_ = MIN_SIZE;
     }
 
+    MemoryIO(const nlohmann::json& io_obj, const IndexCommonParam& common_param)
+        : allocator_(common_param.allocator_) {
+        start_ = reinterpret_cast<uint8_t*>(allocator_->Allocate(MIN_SIZE));
+        current_size_ = MIN_SIZE;
+    }
+
     ~MemoryIO() {
         allocator_->Deallocate(start_);
     }
@@ -52,6 +60,12 @@ class MemoryIO : public BasicIO<MemoryIO> {
     inline void
     PrefetchImpl(uint64_t offset, uint64_t cache_line = 64);
 
+    inline void
+    SerializeImpl(StreamWriter& writer);
+
+    inline void
+    DeserializeImpl(StreamReader& reader);
+
 private:
     [[nodiscard]] inline bool
     checkValidOffset(uint64_t size) const {
@@ -68,7 +82,7 @@ class MemoryIO : public BasicIO<MemoryIO> {
     }
 
 private:
-    Allocator* allocator_{nullptr};
+    Allocator* const allocator_{nullptr};
     uint8_t* start_{nullptr};
     uint64_t current_size_{0};
     static const uint64_t MIN_SIZE = 1024;
@@ -111,5 +125,18 @@ MemoryIO::PrefetchImpl(uint64_t offset, uint64_t cache_line) {
     _mm_prefetch(this->start_ + offset, _MM_HINT_T0);  // todo
 #endif
 }
+void
+MemoryIO::SerializeImpl(StreamWriter& writer) {
+    StreamWriter::WriteObj(writer, this->current_size_);
+    writer.Write(reinterpret_cast<char*>(this->start_), current_size_);
+}
+
+void
+MemoryIO::DeserializeImpl(StreamReader& reader) {
+    allocator_->Deallocate(this->start_);
+    StreamReader::ReadObj(reader, this->current_size_);
+    this->start_ = static_cast<uint8_t*>(allocator_->Allocate(this->current_size_));
+    reader.Read(reinterpret_cast<char*>(this->start_), current_size_);
+}
 
 }  // namespace vsag
diff --git a/src/metric_type.h b/src/metric_type.h
index 3215d3fd..99abf7cb 100644
--- a/src/metric_type.h
+++ b/src/metric_type.h
@@ -14,6 +14,7 @@
 // limitations under the License.
 
 #pragma once
+
 namespace vsag {
 enum class MetricType { METRIC_TYPE_L2SQR = 0, METRIC_TYPE_IP = 1, METRIC_TYPE_COSINE = 2 };
 
diff --git a/src/algorithm/hnswlib/stream_reader.cpp b/src/stream_reader.cpp
similarity index 100%
rename from src/algorithm/hnswlib/stream_reader.cpp
rename to src/stream_reader.cpp
diff --git a/src/algorithm/hnswlib/stream_reader.h b/src/stream_reader.h
similarity index 65%
rename from src/algorithm/hnswlib/stream_reader.h
rename to src/stream_reader.h
index deb461a2..7579bc66 100644
--- a/src/algorithm/hnswlib/stream_reader.h
+++ b/src/stream_reader.h
@@ -19,12 +19,38 @@
 #include <functional>
 #include <istream>
 
+#include "utils.h"
+
 class StreamReader {
 public:
     StreamReader() = default;
 
     virtual void
     Read(char* data, uint64_t size) = 0;
+
+    template <typename T>
+    static void
+    ReadObj(StreamReader& reader, T& val) {
+        reader.Read(reinterpret_cast<char*>(&val), sizeof(val));
+    }
+
+    template <typename T>
+    static void
+    ReadVector(StreamReader& reader, std::vector<T>& val) {
+        uint64_t size;
+        ReadObj(reader, size);
+        val.resize(size);
+        reader.Read(reinterpret_cast<char*>(val.data()), size * sizeof(T));
+    }
+
+    template <typename T>
+    static void
+    ReadVector(StreamReader& reader, vsag::Vector<T>& val) {
+        uint64_t size;
+        ReadObj(reader, size);
+        val.resize(size);
+        reader.Read(reinterpret_cast<char*>(val.data()), size * sizeof(T));
+    }
 };
 
 class ReadFuncStreamReader : public StreamReader {
diff --git a/src/algorithm/hnswlib/stream_writer.cpp b/src/stream_writer.cpp
similarity index 100%
rename from src/algorithm/hnswlib/stream_writer.cpp
rename to src/stream_writer.cpp
diff --git a/src/algorithm/hnswlib/stream_writer.h b/src/stream_writer.h
similarity index 69%
rename from src/algorithm/hnswlib/stream_writer.h
rename to src/stream_writer.h
index aa24ec9a..ca38cd2e 100644
--- a/src/algorithm/hnswlib/stream_writer.h
+++ b/src/stream_writer.h
@@ -19,12 +19,36 @@
 #include <functional>
 #include <ostream>
 
+#include "utils.h"
+
 class StreamWriter {
 public:
     StreamWriter() = default;
 
     virtual void
     Write(char* data, uint64_t size) = 0;
+
+    template <typename T>
+    static void
+    WriteObj(StreamWriter& writer, T& val) {
+        writer.Write(reinterpret_cast<char*>(&val), sizeof(val));
+    }
+
+    template <typename T>
+    static void
+    WriteVector(StreamWriter& writer, std::vector<T>& val) {
+        uint64_t size = val.size();
+        WriteObj(writer, size);
+        writer.Write(reinterpret_cast<char*>(val.data()), size * sizeof(T));
+    }
+
+    template <typename T>
+    static void
+    WriteVector(StreamWriter& writer, vsag::Vector<T>& val) {
+        uint64_t size = val.size();
+        WriteObj(writer, size);
+        writer.Write(reinterpret_cast<char*>(val.data()), size * sizeof(T));
+    }
 };
 
 class BufferStreamWriter : public StreamWriter {


Test patch:
diff --git a/src/io/basic_io_test.h b/src/io/basic_io_test.h
new file mode 100644
index 00000000..ec6db016
--- /dev/null
+++ b/src/io/basic_io_test.h
@@ -0,0 +1,79 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <catch2/catch_test_macros.hpp>
+#include <filesystem>
+#include <fstream>
+#include <memory>
+
+#include "basic_io.h"
+#include "fixtures.h"
+
+using namespace vsag;
+
+template <typename T>
+void
+TestBasicReadWrite(BasicIO<T>& io) {
+    std::vector<uint64_t> counts = {200, 500};
+    std::vector<uint64_t> max_lengths = {2, 20, 37, 64, 128, 260, 999, 4097};
+    for (auto count : counts) {
+        for (auto max_length : max_lengths) {
+            auto vecs = fixtures::GenTestItems(count, max_length);
+            for (auto& item : vecs) {
+                io.Write(item.data_, item.length_, item.start_);
+            }
+            for (auto& item : vecs) {
+                std::vector<uint8_t> data(item.length_);
+                io.Read(data.data(), item.length_, item.start_);
+                REQUIRE(memcmp(data.data(), item.data_, item.length_) == 0);
+            }
+        }
+    }
+}
+
+template <typename T>
+void
+TestSerializeAndDeserialize(BasicIO<T>& wio, BasicIO<T>& rio) {
+    std::vector<uint64_t> counts = {200, 500};
+    std::vector<uint64_t> max_lengths = {2, 20, 37, 64, 128, 260, 999, 4097};
+    srandom(time(nullptr));
+    fixtures::temp_dir dirname("TestSerializeAndDeserialize");
+    for (auto count : counts) {
+        for (auto max_length : max_lengths) {
+            auto filename = dirname.path + "/file_" + std::to_string(random());
+            auto vecs = fixtures::GenTestItems(count, max_length);
+            for (auto& item : vecs) {
+                wio.Write(item.data_, item.length_, item.start_);
+            }
+            std::ofstream outfile(filename.c_str(), std::ios::binary);
+            IOStreamWriter writer(outfile);
+            wio.Serialize(writer);
+            outfile.close();
+
+            std::ifstream infile(filename.c_str(), std::ios::binary);
+            IOStreamReader reader(infile);
+            rio.Deserialize(reader);
+
+            for (auto& item : vecs) {
+                std::vector<uint8_t> data(item.length_);
+                rio.Read(data.data(), item.length_, item.start_);
+                REQUIRE(memcmp(data.data(), item.data_, item.length_) == 0);
+            }
+            infile.close();
+        }
+    }
+}
diff --git a/src/io/memory_block_io_test.cpp b/src/io/memory_block_io_test.cpp
new file mode 100644
index 00000000..b5f940ce
--- /dev/null
+++ b/src/io/memory_block_io_test.cpp
@@ -0,0 +1,36 @@
+
+// Copyright 2024-present the vsag project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "memory_block_io.h"
+
+#include <catch2/catch_test_macros.hpp>
+#include <memory>
+
+#include "basic_io_test.h"
+#include "default_allocator.h"
+using namespace vsag;
+
+TEST_CASE("read&write [ut][memory_block_io]") {
+    auto allocator = std::make_unique<DefaultAllocator>();
+    auto io = std::make_unique<MemoryBlockIO>(allocator.get());
+    TestBasicReadWrite(*io);
+}
+
+TEST_CASE("serialize&deserialize [ut][memory_block_io]") {
+    auto allocator = std::make_unique<DefaultAllocator>();
+    auto wio = std::make_unique<MemoryBlockIO>(allocator.get());
+    auto rio = std::make_unique<MemoryBlockIO>(allocator.get());
+    TestSerializeAndDeserialize(*wio, *rio);
+}
diff --git a/src/io/memory_io_test.cpp b/src/io/memory_io_test.cpp
index 0a8e4143..be4af4bb 100644
--- a/src/io/memory_io_test.cpp
+++ b/src/io/memory_io_test.cpp
@@ -18,34 +18,19 @@
 #include <catch2/catch_test_macros.hpp>
 #include <memory>
 
+#include "basic_io_test.h"
 #include "default_allocator.h"
-#include "fixtures.h"
 using namespace vsag;
 
-template <typename T>
-void
-TestReadWrite(BasicIO<T>* basicIo) {
-    int dim = 32;
-    auto vector = fixtures::generate_vectors(100, dim);
-    auto code_size = dim * sizeof(float);
-    std::unordered_map<uint64_t, float*> maps;
-    for (int i = 0; i < 100; ++i) {
-        auto offset = random() % 10000000 * code_size;
-        basicIo->Write((uint8_t*)(vector.data() + i * dim), code_size, offset);
-        maps[offset] = vector.data() + i * dim;
-    }
-
-    for (auto& iter : maps) {
-        const auto* result = (const float*)(basicIo->Read(code_size, iter.first));
-        auto* gt = iter.second;
-        for (int i = 0; i < dim; ++i) {
-            REQUIRE(result[i] == gt[i]);
-        }
-    }
+TEST_CASE("read&write [ut][memory_io]") {
+    auto allocator = std::make_unique<DefaultAllocator>();
+    auto io = std::make_unique<MemoryIO>(allocator.get());
+    TestBasicReadWrite(*io);
 }
 
-TEST_CASE("read&write[ut][memory_io]") {
+TEST_CASE("serialize&deserialize [ut][memory_io]") {
     auto allocator = std::make_unique<DefaultAllocator>();
-    auto io = std::make_unique<MemoryIO>(allocator.get());
-    TestReadWrite(io.get());
+    auto wio = std::make_unique<MemoryIO>(allocator.get());
+    auto rio = std::make_unique<MemoryIO>(allocator.get());
+    TestSerializeAndDeserialize(*wio, *rio);
 }
diff --git a/tests/fixtures/fixtures.cpp b/tests/fixtures/fixtures.cpp
index 5e3ee591..1a3106b0 100644
--- a/tests/fixtures/fixtures.cpp
+++ b/tests/fixtures/fixtures.cpp
@@ -17,6 +17,7 @@
 
 #include <cstdint>
 #include <string>
+#include <unordered_set>
 
 #include "fmt/format.h"
 #include "vsag/dataset.h"
@@ -200,4 +201,25 @@ brute_force(const vsag::DatasetPtr& query,
 
     return std::move(result);
 }
+
+std::vector<IOItem>
+GenTestItems(uint64_t count, uint64_t max_length, uint64_t max_index) {
+    std::vector<IOItem> result(count);
+    std::unordered_set<uint64_t> maps;
+    for (auto& item : result) {
+        while (true) {
+            item.start_ = (random() % max_index) * max_length;
+            if (not maps.count(item.start_)) {
+                maps.insert(item.start_);
+                break;
+            }
+        };
+        item.length_ = random() % max_length;
+        item.data_ = new uint8_t[item.length_];
+        auto vec = fixtures::generate_vectors(1, max_length, false, random());
+        memcpy(item.data_, vec.data(), item.length_);
+    }
+    return result;
+}
+
 }  // namespace fixtures
diff --git a/tests/fixtures/fixtures.h b/tests/fixtures/fixtures.h
index e563d708..e52f83c8 100644
--- a/tests/fixtures/fixtures.h
+++ b/tests/fixtures/fixtures.h
@@ -68,7 +68,7 @@ brute_force(const vsag::DatasetPtr& query,
             const std::string& metric_type);
 
 struct temp_dir {
-    temp_dir(const std::string& name) {
+    explicit temp_dir(const std::string& name) {
         auto epoch_time = std::chrono::system_clock::now().time_since_epoch();
         auto seconds = std::chrono::duration_cast<std::chrono::seconds>(epoch_time).count();
 
@@ -117,4 +117,17 @@ using dist_t = comparable_float_t;
 using time_t = comparable_float_t;
 using recall_t = comparable_float_t;
 
+struct IOItem {
+    uint64_t start_;
+    uint64_t length_;
+    uint8_t* data_;
+
+    ~IOItem() {
+        delete[] data_;
+    }
+};
+
+std::vector<IOItem>
+GenTestItems(uint64_t count, uint64_t max_length, uint64_t max_index = 100000);
+
 }  // Namespace fixtures