1use crate::Vector;
10use anyhow::{anyhow, Result};
11use oxicode::{Decode, Encode};
12use serde::{Deserialize, Serialize};
13use std::fs::File;
14use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
15use std::path::Path;
16
17fn bincode_config() -> oxicode::config::Configuration<
20 oxicode::config::LittleEndian,
21 oxicode::config::Fixint,
22 oxicode::config::NoLimit,
23> {
24 oxicode::config::standard().with_fixed_int_encoding()
25}
26
27#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, Encode, Decode)]
29pub enum CompressionType {
30 None,
32 Lz4,
34 Zstd,
36 Brotli,
38 Gzip,
40 VectorQuantization,
42}
43
44pub fn compress_bytes(data: &[u8], compression: CompressionType, level: u8) -> Result<Vec<u8>> {
56 match compression {
57 CompressionType::None => Ok(data.to_vec()),
58 CompressionType::Lz4 => {
59 oxiarc_lz4::compress_bytes(data).map_err(|e| anyhow!("LZ4 compression failed: {e}"))
60 }
61 CompressionType::Zstd => {
62 let zstd_level = (level as i32 * 2).clamp(1, 22);
63 oxiarc_zstd::compress_with_level(data, zstd_level)
64 .map_err(|e| anyhow!("Zstd compression failed: {e}"))
65 }
66 CompressionType::Brotli => {
67 let quality = (level as u32).min(11);
68 oxiarc_brotli::compress(data, quality)
69 .map_err(|e| anyhow!("Brotli compression failed: {e}"))
70 }
71 CompressionType::Gzip => oxiarc_deflate::gzip_compress(data, level.min(9))
72 .map_err(|e| anyhow!("Gzip compression failed: {e}")),
73 CompressionType::VectorQuantization => Err(anyhow!(
74 "CompressionType::VectorQuantization is not implemented as a raw-byte \
75 compressor; use CompressionType::None or a real byte codec \
76 (Lz4/Zstd/Brotli/Gzip) instead of silently storing uncompressed data"
77 )),
78 }
79}
80
81pub fn decompress_bytes(
88 data: &[u8],
89 compression: CompressionType,
90 original_size: usize,
91) -> Result<Vec<u8>> {
92 match compression {
93 CompressionType::None => Ok(data.to_vec()),
94 CompressionType::Lz4 => oxiarc_lz4::decompress_bytes(data, original_size)
95 .map_err(|e| anyhow!("LZ4 decompression failed: {e}")),
96 CompressionType::Zstd => {
97 oxiarc_zstd::decompress(data).map_err(|e| anyhow!("Zstd decompression failed: {e}"))
98 }
99 CompressionType::Brotli => {
100 oxiarc_brotli::decompress(data).map_err(|e| anyhow!("Brotli decompression failed: {e}"))
101 }
102 CompressionType::Gzip => oxiarc_deflate::gzip_decompress(data)
103 .map_err(|e| anyhow!("Gzip decompression failed: {e}")),
104 CompressionType::VectorQuantization => Err(anyhow!(
105 "CompressionType::VectorQuantization is not implemented as a raw-byte \
106 compressor"
107 )),
108 }
109}
110
111#[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode)]
113pub struct StorageConfig {
114 pub compression: CompressionType,
116 pub compression_level: u8,
118 pub buffer_size: usize,
120 pub enable_mmap: bool,
122 pub block_size: usize,
124 pub enable_checksums: bool,
126 pub format_version: u32,
128}
129
130impl Default for StorageConfig {
131 fn default() -> Self {
132 Self {
133 compression: CompressionType::Zstd,
134 compression_level: 3,
135 buffer_size: 1024 * 1024, enable_mmap: true,
137 block_size: 64 * 1024, enable_checksums: true,
139 format_version: 1,
140 }
141 }
142}
143
144#[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode)]
146pub struct VectorFileHeader {
147 pub magic: [u8; 8],
149 pub version: u32,
151 pub vector_count: u64,
153 pub dimensions: usize,
155 pub compression: CompressionType,
157 pub compression_level: u8,
159 pub block_size: usize,
161 pub header_checksum: u32,
163 pub data_offset: u64,
165 pub data_size: u64,
167 pub original_size: u64,
169 pub reserved: [u8; 32],
171}
172
173impl Default for VectorFileHeader {
174 fn default() -> Self {
175 Self {
176 magic: *b"OXIRSVEC",
177 version: 1,
178 vector_count: 0,
179 dimensions: 0,
180 compression: CompressionType::None,
181 compression_level: 0,
182 block_size: 64 * 1024,
183 header_checksum: 0,
184 data_offset: 0,
185 data_size: 0,
186 original_size: 0,
187 reserved: [0; 32],
188 }
189 }
190}
191
192impl VectorFileHeader {
193 pub fn calculate_checksum(&mut self) {
195 let mut checksum = 0u32;
197 checksum ^=
198 u32::from_le_bytes([self.magic[0], self.magic[1], self.magic[2], self.magic[3]]);
199 checksum ^=
200 u32::from_le_bytes([self.magic[4], self.magic[5], self.magic[6], self.magic[7]]);
201 checksum ^= self.version;
202 checksum ^= self.vector_count as u32;
203 checksum ^= self.dimensions as u32;
204 checksum ^= self.compression as u8 as u32;
205 checksum ^= self.compression_level as u32;
206 self.header_checksum = checksum;
207 }
208
209 pub fn verify_checksum(&self) -> bool {
211 let mut temp_header = self.clone();
212 temp_header.header_checksum = 0;
213 temp_header.calculate_checksum();
214 temp_header.header_checksum == self.header_checksum
215 }
216}
217
218#[derive(Debug, Clone)]
220pub struct VectorBlock {
221 pub block_id: u32,
223 pub vector_count: u32,
225 pub data: Vec<u8>,
227 pub original_size: u32,
229 pub checksum: u32,
231}
232
233pub struct VectorWriter {
235 writer: BufWriter<File>,
236 config: StorageConfig,
237 header: VectorFileHeader,
238 current_block: Vec<Vector>,
239 blocks_written: u32,
240 total_vectors: u64,
241}
242
243impl VectorWriter {
244 pub fn new<P: AsRef<Path>>(path: P, config: StorageConfig) -> Result<Self> {
246 let file = File::create(path)?;
247 let mut writer = BufWriter::new(file);
248
249 let header = VectorFileHeader {
250 compression: config.compression,
251 compression_level: config.compression_level,
252 block_size: config.block_size,
253 ..Default::default()
254 };
255
256 let placeholder_header_bytes = oxicode::serde::encode_to_vec(&header, bincode_config())
258 .map_err(|e| anyhow!("Failed to serialize placeholder header: {}", e))?;
259 let _header_size = (4 + placeholder_header_bytes.len()) as u64;
260
261 writer.write_all(&(placeholder_header_bytes.len() as u32).to_le_bytes())?;
263 writer.write_all(&placeholder_header_bytes)?;
264 writer.flush()?;
265
266 Ok(Self {
267 writer,
268 config,
269 header,
270 current_block: Vec::new(),
271 blocks_written: 0,
272 total_vectors: 0,
273 })
274 }
275
276 pub fn write_vector(&mut self, vector: Vector) -> Result<()> {
278 if self.header.dimensions == 0 {
280 self.header.dimensions = vector.dimensions;
281 } else if self.header.dimensions != vector.dimensions {
282 return Err(anyhow!(
283 "Vector dimension mismatch: expected {}, got {}",
284 self.header.dimensions,
285 vector.dimensions
286 ));
287 }
288
289 self.current_block.push(vector);
290 self.total_vectors += 1;
291
292 let block_size_estimate = self.current_block.len() * self.header.dimensions * 4; if block_size_estimate >= self.config.block_size {
295 self.flush_block()?;
296 }
297
298 Ok(())
299 }
300
301 pub fn write_vectors(&mut self, vectors: &[Vector]) -> Result<()> {
303 for vector in vectors {
304 self.write_vector(vector.clone())?;
305 }
306 Ok(())
307 }
308
309 fn flush_block(&mut self) -> Result<()> {
311 if self.current_block.is_empty() {
312 return Ok(());
313 }
314
315 if self.config.compression == CompressionType::None {
317 for vector in &self.current_block {
318 let vector_bytes = vector.as_f32();
319 for value in vector_bytes {
320 self.writer.write_all(&value.to_le_bytes())?;
321 }
322 }
323 self.current_block.clear();
324 return Ok(());
325 }
326
327 let mut block_data = Vec::new();
329 for vector in &self.current_block {
330 let vector_bytes = vector.as_f32();
331 for value in vector_bytes {
332 block_data.extend_from_slice(&value.to_le_bytes());
333 }
334 }
335
336 let compressed_data = self.compress_data(&block_data)?;
338
339 let block = VectorBlock {
341 block_id: self.blocks_written,
342 vector_count: self.current_block.len() as u32,
343 original_size: block_data.len() as u32,
344 checksum: self.calculate_data_checksum(&compressed_data),
345 data: compressed_data,
346 };
347
348 self.write_block(&block)?;
350
351 self.current_block.clear();
352 self.blocks_written += 1;
353
354 Ok(())
355 }
356
357 fn compress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
359 compress_bytes(data, self.config.compression, self.config.compression_level)
360 }
361
362 fn calculate_data_checksum(&self, data: &[u8]) -> u32 {
364 data.iter().fold(0u32, |acc, &b| acc.wrapping_add(b as u32))
366 }
367
368 fn write_block(&mut self, block: &VectorBlock) -> Result<()> {
370 self.writer.write_all(&block.block_id.to_le_bytes())?;
372 self.writer.write_all(&block.vector_count.to_le_bytes())?;
373 self.writer.write_all(&block.original_size.to_le_bytes())?;
374 self.writer.write_all(&block.checksum.to_le_bytes())?;
375 self.writer
376 .write_all(&(block.data.len() as u32).to_le_bytes())?;
377
378 self.writer.write_all(&block.data)?;
380
381 Ok(())
382 }
383
384 pub fn finalize(mut self) -> Result<()> {
386 self.flush_block()?;
388
389 self.header.vector_count = self.total_vectors;
391
392 let mut temp_header = self.header.clone();
396 temp_header.calculate_checksum();
397
398 let temp_header_bytes = oxicode::serde::encode_to_vec(&temp_header, bincode_config())
399 .map_err(|e| anyhow!("Failed to serialize header for size calculation: {}", e))?;
400
401 self.header.data_offset = 4 + temp_header_bytes.len() as u64;
403 self.header.calculate_checksum();
404
405 self.writer.flush()?;
407
408 self.writer.get_mut().seek(SeekFrom::Start(0))?;
410
411 let header_bytes = oxicode::serde::encode_to_vec(&self.header, bincode_config())
412 .map_err(|e| anyhow!("Failed to serialize header: {}", e))?;
413
414 let header_size = header_bytes.len() as u32;
416 self.writer.write_all(&header_size.to_le_bytes())?;
417 self.writer.write_all(&header_bytes)?;
418
419 self.writer.flush()?;
421
422 drop(self.writer);
424
425 Ok(())
426 }
427}
428
429pub struct VectorReader {
431 reader: BufReader<File>,
432 header: VectorFileHeader,
433 current_position: u64,
434 vectors_read: u64,
435}
436
437impl VectorReader {
438 pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
440 let file = File::open(path)?;
441 let mut reader = BufReader::new(file);
442
443 let header = Self::read_header(&mut reader)?;
445 let data_offset = header.data_offset;
446
447 reader.get_mut().seek(SeekFrom::Start(data_offset))?;
449
450 Ok(Self {
451 reader,
452 header,
453 current_position: data_offset,
454 vectors_read: 0,
455 })
456 }
457
458 fn read_header(reader: &mut BufReader<File>) -> Result<VectorFileHeader> {
460 let mut size_bytes = [0u8; 4];
462 reader.read_exact(&mut size_bytes)?;
463 let header_size = u32::from_le_bytes(size_bytes) as usize;
464
465 let mut header_data = vec![0u8; header_size];
467 reader.read_exact(&mut header_data)?;
468
469 let (header, _): (VectorFileHeader, _) =
470 oxicode::serde::decode_from_slice(&header_data, bincode_config())
471 .map_err(|e| anyhow!("Failed to deserialize header: {}", e))?;
472
473 if &header.magic != b"OXIRSVEC" {
475 return Err(anyhow!("Invalid file format: magic number mismatch"));
476 }
477
478 if !header.verify_checksum() {
480 return Err(anyhow!("Header checksum verification failed"));
481 }
482
483 Ok(header)
484 }
485
486 pub fn metadata(&self) -> &VectorFileHeader {
488 &self.header
489 }
490
491 pub fn read_vector(&mut self) -> Result<Option<Vector>> {
493 if self.vectors_read >= self.header.vector_count {
494 return Ok(None);
495 }
496
497 let mut vector_data = vec![0f32; self.header.dimensions];
500
501 for vector_item in vector_data.iter_mut().take(self.header.dimensions) {
502 let mut bytes = [0u8; 4];
503 self.reader.read_exact(&mut bytes)?;
504 *vector_item = f32::from_le_bytes(bytes);
505 }
506
507 self.vectors_read += 1;
508 self.current_position += (self.header.dimensions * 4) as u64;
509 Ok(Some(Vector::new(vector_data)))
510 }
511
512 pub fn read_vectors(&mut self, count: usize) -> Result<Vec<Vector>> {
514 let mut vectors = Vec::with_capacity(count);
515
516 for _ in 0..count {
517 if let Some(vector) = self.read_vector()? {
518 vectors.push(vector);
519 } else {
520 break;
521 }
522 }
523
524 Ok(vectors)
525 }
526
527 pub fn read_all(&mut self) -> Result<Vec<Vector>> {
529 let remaining = (self.header.vector_count - self.vectors_read) as usize;
530 self.read_vectors(remaining)
531 }
532
533 pub fn seek_to_vector(&mut self, index: u64) -> Result<()> {
535 if index >= self.header.vector_count {
536 return Err(anyhow!("Vector index {} out of bounds", index));
537 }
538
539 let byte_offset = self.header.data_offset + (index * self.header.dimensions as u64 * 4);
540 self.reader.get_mut().seek(SeekFrom::Start(byte_offset))?;
541 self.vectors_read = index;
542
543 Ok(())
544 }
545}
546
547pub struct MmapVectorFile {
549 _file: File,
550 mmap: memmap2::Mmap,
551 header: VectorFileHeader,
552}
553
554impl MmapVectorFile {
555 pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
557 let file = File::open(path)?;
558 let mmap = unsafe { memmap2::Mmap::map(&file)? };
559
560 if mmap.len() < 4 {
562 return Err(anyhow!("File too small to contain header"));
563 }
564 let header_size = u32::from_le_bytes([mmap[0], mmap[1], mmap[2], mmap[3]]) as usize;
565 if mmap.len() < 4 + header_size {
566 return Err(anyhow!("File too small to contain full header"));
567 }
568 let header_bytes = &mmap[4..4 + header_size];
569 let (header, _): (VectorFileHeader, _) =
570 oxicode::serde::decode_from_slice(header_bytes, bincode_config())
571 .map_err(|e| anyhow!("Failed to deserialize header: {}", e))?;
572
573 if &header.magic != b"OXIRSVEC" {
575 return Err(anyhow!("Invalid file format"));
576 }
577
578 if !header.verify_checksum() {
579 return Err(anyhow!("Header checksum verification failed"));
580 }
581
582 Ok(Self {
583 _file: file,
584 mmap,
585 header,
586 })
587 }
588
589 pub fn get_vector(&self, index: u64) -> Result<Vector> {
591 if index >= self.header.vector_count {
592 return Err(anyhow!("Vector index out of bounds"));
593 }
594
595 let offset =
596 self.header.data_offset as usize + (index as usize * self.header.dimensions * 4);
597 let end_offset = offset + (self.header.dimensions * 4);
598
599 if end_offset > self.mmap.len() {
600 return Err(anyhow!("Vector data extends beyond file"));
601 }
602
603 let vector_bytes = &self.mmap[offset..end_offset];
604 let mut vector_data = vec![0f32; self.header.dimensions];
605
606 for (i, chunk) in vector_bytes.chunks_exact(4).enumerate() {
607 vector_data[i] = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
608 }
609
610 Ok(Vector::new(vector_data))
611 }
612
613 pub fn get_vectors(&self, start: u64, count: usize) -> Result<Vec<Vector>> {
615 let mut vectors = Vec::with_capacity(count);
616
617 for i in 0..count {
618 let index = start + i as u64;
619 if index >= self.header.vector_count {
620 break;
621 }
622 vectors.push(self.get_vector(index)?);
623 }
624
625 Ok(vectors)
626 }
627
628 pub fn vector_count(&self) -> u64 {
630 self.header.vector_count
631 }
632
633 pub fn dimensions(&self) -> usize {
635 self.header.dimensions
636 }
637}
638
639pub struct StorageUtils;
641
642impl StorageUtils {
643 pub fn vectors_to_binary(vectors: &[Vector]) -> Result<Vec<u8>> {
645 let mut data = Vec::new();
646
647 for vector in vectors {
648 let vector_f32 = vector.as_f32();
649 for value in vector_f32 {
650 data.extend_from_slice(&value.to_le_bytes());
651 }
652 }
653
654 Ok(data)
655 }
656
657 pub fn binary_to_vectors(data: &[u8], dimensions: usize) -> Result<Vec<Vector>> {
659 if data.len() % (dimensions * 4) != 0 {
660 return Err(anyhow!("Invalid binary data length for given dimensions"));
661 }
662
663 let vector_count = data.len() / (dimensions * 4);
664 let mut vectors = Vec::with_capacity(vector_count);
665
666 for i in 0..vector_count {
667 let start = i * dimensions * 4;
668 let end = start + dimensions * 4;
669 let vector_bytes = &data[start..end];
670
671 let mut vector_data = vec![0f32; dimensions];
672 for (j, chunk) in vector_bytes.chunks_exact(4).enumerate() {
673 vector_data[j] = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
674 }
675
676 vectors.push(Vector::new(vector_data));
677 }
678
679 Ok(vectors)
680 }
681
682 pub fn estimate_storage_size(
691 vector_count: usize,
692 dimensions: usize,
693 compression: CompressionType,
694 ) -> usize {
695 let raw_size = vector_count * dimensions * 4; let header_size = std::mem::size_of::<VectorFileHeader>();
697
698 let compressed_size = match compression {
699 CompressionType::None => raw_size,
700 CompressionType::Lz4 => (raw_size as f64 * 0.6) as usize, CompressionType::Zstd => (raw_size as f64 * 0.5) as usize, CompressionType::Brotli => (raw_size as f64 * 0.4) as usize, CompressionType::Gzip => (raw_size as f64 * 0.5) as usize, CompressionType::VectorQuantization => raw_size,
708 };
709
710 header_size + compressed_size
711 }
712
713 pub fn benchmark_compression(vectors: &[Vector]) -> Result<Vec<(CompressionType, f64, usize)>> {
717 let binary_data = Self::vectors_to_binary(vectors)?;
718
719 let algorithms = [
720 CompressionType::None,
721 CompressionType::Lz4,
722 CompressionType::Zstd,
723 CompressionType::Brotli,
724 CompressionType::Gzip,
725 ];
726
727 let mut results = Vec::new();
728
729 for &algorithm in &algorithms {
730 let start_time = std::time::Instant::now();
731 let compressed = compress_bytes(&binary_data, algorithm, 3)?;
732 let duration = start_time.elapsed().as_secs_f64();
733 results.push((algorithm, duration, compressed.len()));
734 }
735
736 Ok(results)
737 }
738}
739
740#[cfg(test)]
741mod tests {
742 use super::*;
743 use anyhow::Result;
744 use tempfile::NamedTempFile;
745
746 #[test]
747 fn test_vector_file_header() {
748 let mut header = VectorFileHeader {
749 vector_count: 1000,
750 dimensions: 128,
751 ..Default::default()
752 };
753 header.calculate_checksum();
754
755 assert!(header.verify_checksum());
756
757 header.vector_count = 2000;
759 assert!(!header.verify_checksum());
760 }
761
762 #[test]
763 fn test_storage_utils() -> Result<()> {
764 let vectors = vec![
765 Vector::new(vec![1.0, 2.0, 3.0]),
766 Vector::new(vec![4.0, 5.0, 6.0]),
767 ];
768
769 let binary_data = StorageUtils::vectors_to_binary(&vectors)?;
770 let restored_vectors = StorageUtils::binary_to_vectors(&binary_data, 3)?;
771
772 assert_eq!(vectors.len(), restored_vectors.len());
773 for (original, restored) in vectors.iter().zip(restored_vectors.iter()) {
774 assert_eq!(original.as_f32(), restored.as_f32());
775 }
776 Ok(())
777 }
778
779 #[test]
780 fn test_vector_writer_reader() -> Result<()> {
781 let temp_file = NamedTempFile::new()?;
782 let file_path = temp_file.path();
783
784 {
786 let config = StorageConfig {
787 compression: CompressionType::None,
788 ..Default::default()
789 };
790 let mut writer = VectorWriter::new(file_path, config)?;
791
792 let vectors = vec![
793 Vector::new(vec![1.0, 2.0, 3.0, 4.0]),
794 Vector::new(vec![5.0, 6.0, 7.0, 8.0]),
795 Vector::new(vec![9.0, 10.0, 11.0, 12.0]),
796 ];
797
798 writer.write_vectors(&vectors)?;
799 writer.finalize()?;
800 }
801
802 {
804 let mut reader = VectorReader::open(file_path)?;
805 let metadata = reader.metadata();
806
807 assert_eq!(metadata.vector_count, 3);
808 assert_eq!(metadata.dimensions, 4);
809
810 let vectors = reader.read_all()?;
811 assert_eq!(vectors.len(), 3);
812
813 assert_eq!(vectors[0].as_f32(), &[1.0, 2.0, 3.0, 4.0]);
814 assert_eq!(vectors[1].as_f32(), &[5.0, 6.0, 7.0, 8.0]);
815 assert_eq!(vectors[2].as_f32(), &[9.0, 10.0, 11.0, 12.0]);
816 }
817
818 Ok(())
819 }
820
821 #[test]
822 fn test_compression_benchmark() -> Result<()> {
823 let vectors = vec![
824 Vector::new(vec![1.0; 128]),
825 Vector::new(vec![2.0; 128]),
826 Vector::new(vec![3.0; 128]),
827 ];
828
829 let results = StorageUtils::benchmark_compression(&vectors)?;
830 assert_eq!(results.len(), 5); let none_size = results
834 .iter()
835 .find(|(t, _, _)| *t == CompressionType::None)
836 .expect("None compression type should be present")
837 .2;
838 let zstd_size = results
839 .iter()
840 .find(|(t, _, _)| *t == CompressionType::Zstd)
841 .expect("Zstd compression type should be present")
842 .2;
843
844 assert!(zstd_size < none_size);
845 Ok(())
846 }
847
848 #[test]
854 fn test_compress_bytes_round_trip_all_algorithms() -> Result<()> {
855 let data: Vec<u8> = (0..8192u32).map(|i| (i % 7) as u8).collect();
857
858 for &algorithm in &[
859 CompressionType::None,
860 CompressionType::Lz4,
861 CompressionType::Zstd,
862 CompressionType::Brotli,
863 CompressionType::Gzip,
864 ] {
865 let compressed = compress_bytes(&data, algorithm, 5)?;
866 let restored = decompress_bytes(&compressed, algorithm, data.len())?;
867 assert_eq!(
868 restored, data,
869 "{:?} round-trip should reproduce the original bytes exactly",
870 algorithm
871 );
872
873 if algorithm != CompressionType::None {
874 assert!(
875 compressed.len() < data.len(),
876 "{:?} should actually shrink highly-repetitive data (no-op regression check)",
877 algorithm
878 );
879 }
880 }
881
882 Ok(())
883 }
884
885 #[test]
889 fn test_compress_bytes_vector_quantization_errors_loudly() {
890 let data = vec![1u8, 2, 3, 4, 5];
891 let result = compress_bytes(&data, CompressionType::VectorQuantization, 3);
892 assert!(
893 result.is_err(),
894 "VectorQuantization must error instead of silently passing data through"
895 );
896 }
897}