1#![warn(missing_docs)]
140#![warn(rustdoc::missing_crate_level_docs)]
141
142#[cfg(feature = "wasm")]
144#[global_allocator]
145static ALLOC: wee_alloc::WeeAlloc = wee_alloc::WeeAlloc::INIT;
146
147pub mod distance;
148pub mod e8;
149pub mod error;
150pub mod filter;
151pub mod h4;
152pub mod hnsw;
153pub mod metadata;
154#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
155pub mod persistence;
156pub mod quantization;
157pub mod storage;
158
159#[cfg(feature = "python")]
160pub mod python;
161
162pub use distance::Distance;
164pub use e8::{E8Codec, HadamardTransform};
165pub use error::{EmbedVecError, Result};
166pub use filter::FilterExpr;
167pub use h4::{H4Codec, hadamard4_inplace};
168pub use hnsw::HnswIndex;
169pub use metadata::Metadata;
170#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
171pub use persistence::{BackendConfig, BackendType, PersistenceBackend};
172pub use quantization::Quantization;
173pub use storage::VectorStorage;
174
175use ordered_float::OrderedFloat;
176use parking_lot::RwLock;
177use serde::{Deserialize, Serialize};
178use std::collections::HashSet;
179use std::sync::Arc;
180
181#[derive(Debug, Clone, Serialize, Deserialize)]
183pub struct Hit {
184 pub id: usize,
186 pub score: f32,
188 pub payload: Metadata,
190}
191
192impl Hit {
193 pub fn new(id: usize, score: f32, payload: Metadata) -> Self {
195 Self { id, score, payload }
196 }
197}
198
199#[derive(Debug, Clone)]
201pub struct EmbedVecBuilder {
202 dimension: usize,
203 distance: Distance,
204 m: usize,
205 ef_construction: usize,
206 quantization: Quantization,
207 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
208 persistence_config: Option<persistence::BackendConfig>,
209}
210
211impl EmbedVecBuilder {
212 pub fn new(dimension: usize) -> Self {
214 Self {
215 dimension,
216 distance: Distance::Cosine,
217 m: 16,
218 ef_construction: 200,
219 quantization: Quantization::None,
220 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
221 persistence_config: None,
222 }
223 }
224
225 pub fn dimension(mut self, dim: usize) -> Self {
227 self.dimension = dim;
228 self
229 }
230
231 pub fn metric(mut self, distance: Distance) -> Self {
233 self.distance = distance;
234 self
235 }
236
237 pub fn m(mut self, m: usize) -> Self {
239 self.m = m;
240 self
241 }
242
243 pub fn ef_construction(mut self, ef: usize) -> Self {
245 self.ef_construction = ef;
246 self
247 }
248
249 pub fn quantization(mut self, quant: Quantization) -> Self {
251 self.quantization = quant;
252 self
253 }
254
255 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
257 pub fn persistence(mut self, path: impl Into<String>) -> Self {
258 self.persistence_config = Some(persistence::BackendConfig::new(path));
259 self
260 }
261
262 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
264 pub fn persistence_config(mut self, config: persistence::BackendConfig) -> Self {
265 self.persistence_config = Some(config);
266 self
267 }
268
269 #[cfg(feature = "async")]
271 pub async fn build(self) -> Result<EmbedVec> {
272 EmbedVec::from_builder(self).await
273 }
274
275 #[cfg(not(feature = "async"))]
277 pub fn build(self) -> Result<EmbedVec> {
278 EmbedVec::new_internal(
279 self.dimension,
280 self.distance,
281 self.m,
282 self.ef_construction,
283 self.quantization,
284 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
285 self.persistence_config,
286 )
287 }
288}
289
290pub struct EmbedVec {
295 dimension: usize,
297 distance: Distance,
299 pub index: Arc<RwLock<HnswIndex>>,
301 pub storage: Arc<RwLock<VectorStorage>>,
303 pub metadata: Arc<RwLock<Vec<Metadata>>>,
305 quantization: Quantization,
307 e8_codec: Option<E8Codec>,
309 h4_codec: Option<H4Codec>,
311 deleted: Arc<RwLock<HashSet<usize>>>,
314 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
316 backend: Option<Box<dyn persistence::PersistenceBackend>>,
317}
318
319#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
325const PERSIST_HEADER_KEY: &[u8] = b"__embedvec_header__";
326
327#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
330const PERSIST_REC_PREFIX: &[u8] = b"rec:";
331
332#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
334const PERSIST_VERSION: u32 = 1;
335
336#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
339#[derive(Serialize, Deserialize)]
340struct PersistHeader {
341 version: u32,
342 dimension: usize,
343 distance: Distance,
344 quantization: Quantization,
345 m: usize,
346 ef_construction: usize,
347 #[serde(default)]
350 high_water_mark: usize,
351}
352
353#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
355#[derive(Serialize, Deserialize)]
356struct PersistedRecord {
357 stored: storage::StoredVector,
358 meta: Metadata,
359}
360
361impl EmbedVec {
362 #[cfg(all(feature = "async", any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb")))]
380 pub async fn new(
381 dim: usize,
382 distance: Distance,
383 m: usize,
384 ef_construction: usize,
385 ) -> Result<Self> {
386 Self::new_internal(dim, distance, m, ef_construction, Quantization::None, None)
387 }
388
389 #[cfg(all(feature = "async", not(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))))]
390 pub async fn new(
391 dim: usize,
392 distance: Distance,
393 m: usize,
394 ef_construction: usize,
395 ) -> Result<Self> {
396 Self::new_internal(dim, distance, m, ef_construction, Quantization::None)
397 }
398
399 #[cfg(all(feature = "async", any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb")))]
401 pub async fn with_persistence(
402 path: impl AsRef<std::path::Path>,
403 dim: usize,
404 distance: Distance,
405 m: usize,
406 ef_construction: usize,
407 ) -> Result<Self> {
408 let path_str = path.as_ref().to_string_lossy().to_string();
409 let config = persistence::BackendConfig::new(path_str);
410 Self::new_internal(
411 dim,
412 distance,
413 m,
414 ef_construction,
415 Quantization::None,
416 Some(config),
417 )
418 }
419
420 #[cfg(all(feature = "async", any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb")))]
422 pub async fn with_backend(
423 config: persistence::BackendConfig,
424 dim: usize,
425 distance: Distance,
426 m: usize,
427 ef_construction: usize,
428 ) -> Result<Self> {
429 Self::new_internal(
430 dim,
431 distance,
432 m,
433 ef_construction,
434 Quantization::None,
435 Some(config),
436 )
437 }
438
439 #[cfg(feature = "async")]
441 async fn from_builder(builder: EmbedVecBuilder) -> Result<Self> {
442 Self::new_internal(
443 builder.dimension,
444 builder.distance,
445 builder.m,
446 builder.ef_construction,
447 builder.quantization,
448 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
449 builder.persistence_config,
450 )
451 }
452
453 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
455 pub fn new_internal(
456 dim: usize,
457 distance: Distance,
458 m: usize,
459 ef_construction: usize,
460 quantization: Quantization,
461 persistence_config: Option<persistence::BackendConfig>,
462 ) -> Result<Self> {
463 if dim == 0 {
464 return Err(EmbedVecError::InvalidDimension(dim));
465 }
466
467 let mut index = HnswIndex::new(m, ef_construction, distance);
468 let storage = VectorStorage::new(dim, quantization.clone());
469
470 let e8_codec = match &quantization {
471 Quantization::E8 { bits_per_block, use_hadamard, random_seed } =>
472 Some(E8Codec::new(dim, *bits_per_block, *use_hadamard, *random_seed)),
473 _ => None,
474 };
475
476 let h4_codec = match &quantization {
477 Quantization::H4 { use_hadamard, random_seed } =>
478 Some(H4Codec::new(dim, *use_hadamard, *random_seed)),
479 _ => None,
480 };
481
482 index.set_h4_codec(h4_codec.clone());
485
486 let backend = if let Some(config) = persistence_config {
487 Some(persistence::create_backend(&config)?)
488 } else {
489 None
490 };
491
492 let mut db = Self {
493 dimension: dim,
494 distance,
495 index: Arc::new(RwLock::new(index)),
496 storage: Arc::new(RwLock::new(storage)),
497 metadata: Arc::new(RwLock::new(Vec::new())),
498 quantization,
499 e8_codec,
500 h4_codec,
501 deleted: Arc::new(RwLock::new(HashSet::new())),
502 backend,
503 };
504
505 db.load_from_backend()?;
508
509 Ok(db)
510 }
511
512 #[cfg(not(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb")))]
514 pub fn new_internal(
515 dim: usize,
516 distance: Distance,
517 m: usize,
518 ef_construction: usize,
519 quantization: Quantization,
520 ) -> Result<Self> {
521 if dim == 0 {
522 return Err(EmbedVecError::InvalidDimension(dim));
523 }
524
525 let mut index = HnswIndex::new(m, ef_construction, distance);
526 let storage = VectorStorage::new(dim, quantization.clone());
527
528 let e8_codec = match &quantization {
529 Quantization::E8 { bits_per_block, use_hadamard, random_seed } =>
530 Some(E8Codec::new(dim, *bits_per_block, *use_hadamard, *random_seed)),
531 _ => None,
532 };
533
534 let h4_codec = match &quantization {
535 Quantization::H4 { use_hadamard, random_seed } =>
536 Some(H4Codec::new(dim, *use_hadamard, *random_seed)),
537 _ => None,
538 };
539
540 index.set_h4_codec(h4_codec.clone());
543
544 Ok(Self {
545 dimension: dim,
546 distance,
547 index: Arc::new(RwLock::new(index)),
548 storage: Arc::new(RwLock::new(storage)),
549 metadata: Arc::new(RwLock::new(Vec::new())),
550 quantization,
551 e8_codec,
552 h4_codec,
553 deleted: Arc::new(RwLock::new(HashSet::new())),
554 })
555 }
556
557 pub fn builder() -> EmbedVecBuilder {
559 EmbedVecBuilder::new(768) }
561
562 #[cfg(feature = "async")]
571 pub async fn add(&mut self, vector: &[f32], payload: impl Into<Metadata>) -> Result<usize> {
572 self.add_internal(vector, payload.into())
573 }
574
575 #[cfg(feature = "async")]
581 pub async fn add_many(
582 &mut self,
583 vectors: &[Vec<f32>],
584 payloads: Vec<impl Into<Metadata>>,
585 ) -> Result<()> {
586 if vectors.len() != payloads.len() {
587 return Err(EmbedVecError::MismatchedLengths {
588 vectors: vectors.len(),
589 payloads: payloads.len(),
590 });
591 }
592
593 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
594 let mut new_ids: Vec<usize> = Vec::with_capacity(vectors.len());
595
596 for (vector, payload) in vectors.iter().zip(payloads.into_iter()) {
597 let _id = self.insert_in_memory(vector, payload.into())?;
598 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
599 new_ids.push(_id);
600 }
601
602 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
604 self.persist_records(&new_ids)?;
605
606 Ok(())
607 }
608
609 pub fn add_internal(&mut self, vector: &[f32], payload: Metadata) -> Result<usize> {
611 let id = self.insert_in_memory(vector, payload)?;
612
613 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
615 self.persist_record(id)?;
616
617 Ok(id)
618 }
619
620 fn insert_in_memory(&mut self, vector: &[f32], payload: Metadata) -> Result<usize> {
623 if vector.len() != self.dimension {
624 return Err(EmbedVecError::DimensionMismatch {
625 expected: self.dimension,
626 got: vector.len(),
627 });
628 }
629
630 let processed_vector = if self.distance == Distance::Cosine {
632 normalize_vector(vector)
633 } else {
634 vector.to_vec()
635 };
636
637 let id = {
639 let mut storage = self.storage.write();
640 storage.add(&processed_vector, self.e8_codec.as_ref(), self.h4_codec.as_ref())?
641 };
642
643 {
645 let mut meta = self.metadata.write();
646 if id >= meta.len() {
647 meta.resize(id + 1, Metadata::default());
648 }
649 meta[id] = payload;
650 }
651
652 {
654 let mut index = self.index.write();
655 let storage = self.storage.read();
656 index.insert(id, &processed_vector, &storage, self.e8_codec.as_ref())?;
657 }
658
659 Ok(id)
660 }
661
662 #[cfg(feature = "async")]
673 pub async fn delete(&mut self, id: usize) -> Result<bool> {
674 self.delete_internal(id)
675 }
676
677 #[cfg(feature = "async")]
679 pub async fn delete_many(&mut self, ids: &[usize]) -> Result<usize> {
680 let mut removed = 0;
681 for &id in ids {
682 if self.delete_internal(id)? {
683 removed += 1;
684 }
685 }
686 Ok(removed)
687 }
688
689 pub fn delete_internal(&mut self, id: usize) -> Result<bool> {
691 if id >= self.storage.read().len() {
693 return Ok(false);
694 }
695 if !self.deleted.write().insert(id) {
697 return Ok(false);
698 }
699 {
701 let mut meta = self.metadata.write();
702 if let Some(slot) = meta.get_mut(id) {
703 *slot = Metadata::default();
704 }
705 }
706 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
708 {
709 if let Some(backend) = &self.backend {
710 backend.delete(&Self::rec_key(id))?;
711 }
712 }
713 Ok(true)
714 }
715
716 #[cfg(feature = "async")]
727 pub async fn search(
728 &self,
729 query: &[f32],
730 k: usize,
731 ef_search: usize,
732 filter: Option<FilterExpr>,
733 ) -> Result<Vec<Hit>> {
734 self.search_internal(query, k, ef_search, filter)
735 }
736
737 pub fn search_internal(
739 &self,
740 query: &[f32],
741 k: usize,
742 ef_search: usize,
743 filter: Option<FilterExpr>,
744 ) -> Result<Vec<Hit>> {
745 if query.len() != self.dimension {
746 return Err(EmbedVecError::DimensionMismatch {
747 expected: self.dimension,
748 got: query.len(),
749 });
750 }
751
752 let processed_query = if self.distance == Distance::Cosine {
754 normalize_vector(query)
755 } else {
756 query.to_vec()
757 };
758
759 let fetch_k = ef_search.max(k);
762
763 let candidates = {
765 let index = self.index.read();
766 let storage = self.storage.read();
767 index.search(
768 &processed_query,
769 fetch_k,
770 ef_search,
771 &storage,
772 self.e8_codec.as_ref(),
773 )?
774 };
775
776 let metadata = self.metadata.read();
778 let deleted = self.deleted.read();
779 let mut results: Vec<Hit> = candidates
780 .into_iter()
781 .filter_map(|(id, score)| {
782 if deleted.contains(&id) {
784 return None;
785 }
786
787 let payload = metadata.get(id)?.clone();
788
789 if let Some(ref f) = filter {
791 if !f.matches(&payload) {
792 return None;
793 }
794 }
795
796 Some(Hit::new(id, score, payload))
797 })
798 .take(k)
799 .collect();
800
801 results.sort_by_key(|h| OrderedFloat(h.score));
803
804 Ok(results)
805 }
806
807 #[cfg(feature = "async")]
813 pub async fn search_many(
814 &self,
815 queries: &[Vec<f32>],
816 k: usize,
817 ef_search: usize,
818 filter: Option<FilterExpr>,
819 ) -> Result<Vec<Vec<Hit>>> {
820 self.search_many_internal(queries, k, ef_search, filter)
821 }
822
823 pub fn search_many_internal(
825 &self,
826 queries: &[Vec<f32>],
827 k: usize,
828 ef_search: usize,
829 filter: Option<FilterExpr>,
830 ) -> Result<Vec<Vec<Hit>>> {
831 use rayon::prelude::*;
832 queries
833 .par_iter()
834 .map(|q| self.search_internal(q, k, ef_search, filter.clone()))
835 .collect()
836 }
837
838 pub fn payload(&self, id: usize) -> Option<Metadata> {
840 if self.deleted.read().contains(&id) {
841 return None;
842 }
843 self.metadata.read().get(id).cloned()
844 }
845
846 pub fn contains_id(&self, id: usize) -> bool {
848 id < self.storage.read().len() && !self.deleted.read().contains(&id)
849 }
850
851 pub fn entries(&self) -> Vec<(usize, Metadata)> {
857 let metadata = self.metadata.read();
858 let deleted = self.deleted.read();
859 metadata
860 .iter()
861 .enumerate()
862 .filter(|(id, _)| !deleted.contains(id))
863 .map(|(id, m)| (id, m.clone()))
864 .collect()
865 }
866
867 pub fn live_count(&self) -> usize {
869 self.storage
870 .read()
871 .len()
872 .saturating_sub(self.deleted.read().len())
873 }
874
875 #[cfg(feature = "async")]
877 pub async fn len(&self) -> usize {
878 self.live_count()
879 }
880
881 #[cfg(feature = "async")]
883 pub async fn is_empty(&self) -> bool {
884 self.len().await == 0
885 }
886
887 #[cfg(feature = "async")]
889 pub async fn clear(&mut self) -> Result<()> {
890 self.clear_sync()
891 }
892
893 pub fn clear_sync(&mut self) -> Result<()> {
895 {
896 let mut storage = self.storage.write();
897 storage.clear();
898 }
899 {
900 let mut metadata = self.metadata.write();
901 metadata.clear();
902 }
903 {
904 let mut index = self.index.write();
905 index.clear();
906 }
907 {
908 let mut deleted = self.deleted.write();
909 deleted.clear();
910 }
911
912 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
915 self.clear_backend()?;
916
917 Ok(())
918 }
919
920 #[cfg(all(feature = "async", any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb")))]
922 pub async fn flush(&mut self) -> Result<()> {
923 self.persist_header()?;
926 if let Some(ref backend) = self.backend {
927 backend.flush()?;
928 }
929 Ok(())
930 }
931
932 pub fn quantization(&self) -> &Quantization {
934 &self.quantization
935 }
936
937 #[cfg(feature = "async")]
939 pub async fn set_quantization(&mut self, quant: Quantization) -> Result<()> {
940 self.e8_codec = match &quant {
941 Quantization::E8 { bits_per_block, use_hadamard, random_seed } =>
942 Some(E8Codec::new(self.dimension, *bits_per_block, *use_hadamard, *random_seed)),
943 _ => None,
944 };
945 self.h4_codec = match &quant {
946 Quantization::H4 { use_hadamard, random_seed } =>
947 Some(H4Codec::new(self.dimension, *use_hadamard, *random_seed)),
948 _ => None,
949 };
950 self.quantization = quant.clone();
951
952 self.index.write().set_h4_codec(self.h4_codec.clone());
955
956 {
959 let mut storage = self.storage.write();
960 storage.set_quantization(quant, self.e8_codec.as_ref(), self.h4_codec.as_ref())?;
961 }
962
963 #[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
965 self.persist_all()?;
966
967 Ok(())
968 }
969
970 pub fn dimension(&self) -> usize {
972 self.dimension
973 }
974
975 pub fn distance(&self) -> Distance {
977 self.distance
978 }
979}
980
981#[cfg(any(feature = "persistence-fjall", feature = "persistence-sled", feature = "persistence-rocksdb"))]
986impl EmbedVec {
987 fn rec_key(id: usize) -> Vec<u8> {
989 format!("rec:{:020}", id).into_bytes()
990 }
991
992 fn encode_record(&self, id: usize) -> Result<Vec<u8>> {
994 let stored = self
995 .storage
996 .read()
997 .get_stored(id)
998 .cloned()
999 .ok_or(EmbedVecError::VectorNotFound(id))?;
1000 let meta = self.metadata.read().get(id).cloned().unwrap_or_default();
1001 let record = PersistedRecord { stored, meta };
1002 serde_json::to_vec(&record)
1003 .map_err(|e| EmbedVecError::SerializationError(e.to_string()))
1004 }
1005
1006 fn persist_header(&self) -> Result<()> {
1008 let backend = match &self.backend {
1009 Some(b) => b,
1010 None => return Ok(()),
1011 };
1012 let header = PersistHeader {
1013 version: PERSIST_VERSION,
1014 dimension: self.dimension,
1015 distance: self.distance,
1016 quantization: self.quantization.clone(),
1017 m: self.index.read().m(),
1018 ef_construction: self.index.read().ef_construction(),
1019 high_water_mark: self.storage.read().len(),
1020 };
1021 let bytes = serde_json::to_vec(&header)
1022 .map_err(|e| EmbedVecError::SerializationError(e.to_string()))?;
1023 backend.set(PERSIST_HEADER_KEY, &bytes)
1024 }
1025
1026 fn persist_record(&self, id: usize) -> Result<()> {
1028 let backend = match &self.backend {
1029 Some(b) => b,
1030 None => return Ok(()),
1031 };
1032 let value = self.encode_record(id)?;
1033 backend.set(&Self::rec_key(id), &value)
1034 }
1035
1036 fn persist_records(&self, ids: &[usize]) -> Result<()> {
1038 let backend = match &self.backend {
1039 Some(b) => b,
1040 None => return Ok(()),
1041 };
1042 if ids.is_empty() {
1043 return Ok(());
1044 }
1045 let mut batch: Vec<(Vec<u8>, Vec<u8>)> = Vec::with_capacity(ids.len());
1046 for &id in ids {
1047 batch.push((Self::rec_key(id), self.encode_record(id)?));
1048 }
1049 backend.set_batch(&batch)
1050 }
1051
1052 fn persist_all(&self) -> Result<()> {
1054 if self.backend.is_none() {
1055 return Ok(());
1056 }
1057 self.persist_header()?;
1058 let n = self.storage.read().len();
1059 let ids: Vec<usize> = (0..n).collect();
1060 self.persist_records(&ids)
1061 }
1062
1063 fn clear_backend(&self) -> Result<()> {
1065 let backend = match &self.backend {
1066 Some(b) => b,
1067 None => return Ok(()),
1068 };
1069 let keys: Vec<Vec<u8>> = backend
1070 .scan_prefix(PERSIST_REC_PREFIX)?
1071 .into_iter()
1072 .map(|(k, _)| k)
1073 .collect();
1074 for key in keys {
1075 backend.delete(&key)?;
1076 }
1077 Ok(())
1078 }
1079
1080 fn load_from_backend(&mut self) -> Result<()> {
1091 let header_bytes = match &self.backend {
1092 Some(backend) => backend.get(PERSIST_HEADER_KEY)?,
1093 None => return Ok(()),
1094 };
1095
1096 let header: PersistHeader = match header_bytes {
1097 Some(bytes) => serde_json::from_slice(&bytes)
1098 .map_err(|e| EmbedVecError::SerializationError(e.to_string()))?,
1099 None => return self.persist_header(),
1101 };
1102
1103 if header.dimension != self.dimension {
1104 return Err(EmbedVecError::DimensionMismatch {
1105 expected: header.dimension,
1106 got: self.dimension,
1107 });
1108 }
1109
1110 self.distance = header.distance;
1112 self.quantization = header.quantization.clone();
1113 self.e8_codec = match &self.quantization {
1114 Quantization::E8 { bits_per_block, use_hadamard, random_seed } =>
1115 Some(E8Codec::new(self.dimension, *bits_per_block, *use_hadamard, *random_seed)),
1116 _ => None,
1117 };
1118 self.h4_codec = match &self.quantization {
1119 Quantization::H4 { use_hadamard, random_seed } =>
1120 Some(H4Codec::new(self.dimension, *use_hadamard, *random_seed)),
1121 _ => None,
1122 };
1123 *self.storage.write() = VectorStorage::new(self.dimension, self.quantization.clone());
1124 let mut rebuilt = HnswIndex::new(header.m, header.ef_construction, self.distance);
1125 rebuilt.set_h4_codec(self.h4_codec.clone());
1126 *self.index.write() = rebuilt;
1127 self.metadata.write().clear();
1128 self.deleted.write().clear();
1129
1130 let records = match &self.backend {
1134 Some(backend) => backend.scan_prefix(PERSIST_REC_PREFIX)?,
1135 None => return Ok(()),
1136 };
1137
1138 let mut deleted_ids: HashSet<usize> = HashSet::new();
1139 {
1140 let mut storage = self.storage.write();
1141 let mut metadata = self.metadata.write();
1142 let mut index = self.index.write();
1143 let mut next_id = 0usize;
1144
1145 for (key, value) in records {
1146 let id: usize = std::str::from_utf8(&key[PERSIST_REC_PREFIX.len()..])
1147 .ok()
1148 .and_then(|s| s.parse().ok())
1149 .ok_or_else(|| {
1150 EmbedVecError::PersistenceError(format!(
1151 "invalid record key: {}",
1152 String::from_utf8_lossy(&key)
1153 ))
1154 })?;
1155
1156 while next_id < id {
1158 storage.push_stored(crate::storage::StoredVector::Tombstone);
1159 metadata.push(Metadata::default());
1160 deleted_ids.insert(next_id);
1161 next_id += 1;
1162 }
1163
1164 let record: PersistedRecord = serde_json::from_slice(&value)
1165 .map_err(|e| EmbedVecError::SerializationError(e.to_string()))?;
1166 let decoded = record
1168 .stored
1169 .to_f32(self.e8_codec.as_ref(), self.h4_codec.as_ref());
1170 storage.push_stored(record.stored);
1171 metadata.push(record.meta);
1172 index.insert(id, &decoded, &*storage, self.e8_codec.as_ref())?;
1173 next_id += 1;
1174 }
1175
1176 while next_id < header.high_water_mark {
1178 storage.push_stored(crate::storage::StoredVector::Tombstone);
1179 metadata.push(Metadata::default());
1180 deleted_ids.insert(next_id);
1181 next_id += 1;
1182 }
1183 }
1184
1185 *self.deleted.write() = deleted_ids;
1186
1187 Ok(())
1188 }
1189}
1190
1191fn normalize_vector(v: &[f32]) -> Vec<f32> {
1193 let norm: f32 = v.iter().map(|x| x * x).sum::<f32>().sqrt();
1194 if norm > 1e-10 {
1195 v.iter().map(|x| x / norm).collect()
1196 } else {
1197 v.to_vec()
1198 }
1199}
1200
1201#[cfg(test)]
1202mod tests {
1203 use super::*;
1204
1205 #[tokio::test]
1206 async fn test_basic_operations() {
1207 let mut db = EmbedVec::new(4, Distance::Cosine, 16, 100).await.unwrap();
1208
1209 let id = db
1210 .add(&[1.0, 0.0, 0.0, 0.0], serde_json::json!({"test": "value"}))
1211 .await
1212 .unwrap();
1213 assert_eq!(id, 0);
1214
1215 let results = db.search(&[1.0, 0.0, 0.0, 0.0], 1, 50, None).await.unwrap();
1216 assert_eq!(results.len(), 1);
1217 assert_eq!(results[0].id, 0);
1218 }
1219
1220 #[tokio::test]
1221 async fn test_dimension_mismatch() {
1222 let mut db = EmbedVec::new(4, Distance::Cosine, 16, 100).await.unwrap();
1223
1224 let result = db
1225 .add(&[1.0, 0.0, 0.0], serde_json::json!({}))
1226 .await;
1227 assert!(result.is_err());
1228 }
1229
1230 #[tokio::test]
1231 async fn test_h4_search_recall_multi() {
1232 let dim = 128;
1237 let n = 16;
1238 let mut db = EmbedVec::builder()
1239 .dimension(dim)
1240 .metric(Distance::Cosine)
1241 .quantization(Quantization::h4_default())
1242 .build()
1243 .await
1244 .unwrap();
1245
1246 let mut seed = 0x1234_5678u64;
1248 let mut rng = || {
1249 seed ^= seed << 13;
1250 seed ^= seed >> 7;
1251 seed ^= seed << 17;
1252 (seed as f32 / u64::MAX as f32) * 2.0 - 1.0
1253 };
1254 let mut vectors = Vec::new();
1255 for i in 0..n {
1256 let v: Vec<f32> = (0..dim).map(|_| rng()).collect();
1257 db.add(&v, serde_json::json!({ "i": i })).await.unwrap();
1258 vectors.push(v);
1259 }
1260
1261 let mut self_hits = 0;
1263 for (i, v) in vectors.iter().enumerate() {
1264 let hits = db.search(v, 1, 64, None).await.unwrap();
1265 if hits[0].id == i {
1266 self_hits += 1;
1267 }
1268 }
1269 assert!(
1270 self_hits >= n - 1,
1271 "H4 self-recall too low: {self_hits}/{n} (index not decoding H4?)"
1272 );
1273
1274 let h0 = db.search(&vectors[0], 1, 64, None).await.unwrap()[0].id;
1276 let h_last = db.search(&vectors[n - 1], 1, 64, None).await.unwrap()[0].id;
1277 assert_ne!(h0, h_last, "distinct H4 queries returned the same top hit");
1278 }
1279
1280 #[tokio::test]
1281 async fn test_search_many_and_entries() {
1282 let mut db = EmbedVec::new(4, Distance::Cosine, 16, 100).await.unwrap();
1283 let a = db.add(&[1.0, 0.0, 0.0, 0.0], serde_json::json!({"doc": "a"})).await.unwrap();
1284 let b = db.add(&[0.0, 1.0, 0.0, 0.0], serde_json::json!({"doc": "b"})).await.unwrap();
1285 let _c = db.add(&[0.0, 0.0, 1.0, 0.0], serde_json::json!({"doc": "c"})).await.unwrap();
1286
1287 let queries = vec![vec![1.0, 0.0, 0.0, 0.0], vec![0.0, 1.0, 0.0, 0.0]];
1289 let results = db.search_many(&queries, 1, 50, None).await.unwrap();
1290 assert_eq!(results.len(), 2);
1291 assert_eq!(results[0][0].id, a);
1292 assert_eq!(results[1][0].id, b);
1293
1294 let entries = db.entries();
1296 assert_eq!(entries.len(), 3);
1297 assert_eq!(db.payload(a).unwrap()["doc"], "a");
1298 assert!(db.contains_id(b));
1299
1300 db.delete(b).await.unwrap();
1302 assert_eq!(db.entries().len(), 2);
1303 assert!(db.payload(b).is_none());
1304 assert!(!db.contains_id(b));
1305 }
1306
1307 #[tokio::test]
1308 async fn test_delete_in_memory() {
1309 let mut db = EmbedVec::new(4, Distance::Cosine, 16, 100).await.unwrap();
1310 let a = db.add(&[1.0, 0.0, 0.0, 0.0], serde_json::json!({"v": "a"})).await.unwrap();
1311 let b = db.add(&[0.0, 1.0, 0.0, 0.0], serde_json::json!({"v": "b"})).await.unwrap();
1312 let c = db.add(&[0.0, 0.0, 1.0, 0.0], serde_json::json!({"v": "c"})).await.unwrap();
1313 assert_eq!(db.len().await, 3);
1314
1315 assert!(db.delete(b).await.unwrap());
1317 assert_eq!(db.len().await, 2);
1318
1319 assert!(!db.delete(b).await.unwrap());
1321 assert!(!db.delete(999).await.unwrap());
1322
1323 let hits = db.search(&[0.0, 1.0, 0.0, 0.0], 3, 50, None).await.unwrap();
1325 assert!(hits.iter().all(|h| h.id != b));
1326 assert_eq!(db.search(&[1.0, 0.0, 0.0, 0.0], 1, 50, None).await.unwrap()[0].id, a);
1327 assert_eq!(db.search(&[0.0, 0.0, 1.0, 0.0], 1, 50, None).await.unwrap()[0].id, c);
1328
1329 let d = db.add(&[0.0, 0.0, 0.0, 1.0], serde_json::json!({"v": "d"})).await.unwrap();
1331 assert_eq!(d, 3);
1332 assert_ne!(d, b);
1333 assert_eq!(db.len().await, 3);
1334 }
1335
1336 #[tokio::test]
1337 async fn test_h4_quantization_end_to_end() {
1338 let mut db = EmbedVec::builder()
1339 .dimension(8)
1340 .metric(Distance::Cosine)
1341 .quantization(Quantization::h4_default())
1342 .build()
1343 .await
1344 .unwrap();
1345
1346 let id = db
1347 .add(&[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], serde_json::json!({"lattice": "h4"}))
1348 .await
1349 .unwrap();
1350 assert_eq!(id, 0);
1351
1352 let results = db
1353 .search(&[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], 1, 50, None)
1354 .await
1355 .unwrap();
1356 assert_eq!(results.len(), 1);
1357 assert_eq!(results[0].id, 0);
1358 }
1359}