use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::Path;
use std::sync::RwLock;
use hnsw_rs::prelude::*;
use crate::config::HnswConfig;
use crate::error::{PulseDBError, Result};
use crate::types::ExperienceId;
use super::VectorIndex;
const BRUTE_FORCE_THRESHOLD: usize = 128;
struct FilterBridge<'a>(&'a (dyn Fn(&usize) -> bool + Sync));
impl FilterT for FilterBridge<'_> {
fn hnsw_filter(&self, id: &DataId) -> bool {
(self.0)(id)
}
}
pub struct HnswIndex {
hnsw: Hnsw<'static, f32, DistCosine>,
state: RwLock<IndexState>,
#[allow(dead_code)]
config: HnswConfig,
dimension: usize,
}
#[derive(Debug)]
struct IndexState {
id_to_internal: HashMap<ExperienceId, usize>,
internal_to_id: Vec<ExperienceId>,
deleted: HashSet<usize>,
next_id: usize,
}
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct IndexMetadata {
pub(crate) dimension: usize,
pub(crate) next_id: usize,
pub(crate) id_map: Vec<(String, usize)>,
pub(crate) deleted: Vec<String>,
}
impl HnswIndex {
pub fn new(dimension: usize, config: &HnswConfig) -> Self {
let hnsw = Hnsw::new(
config.max_nb_connection,
config.max_elements,
config.max_layer,
config.ef_construction,
DistCosine,
);
Self {
hnsw,
state: RwLock::new(IndexState {
id_to_internal: HashMap::new(),
internal_to_id: Vec::new(),
deleted: HashSet::new(),
next_id: 0,
}),
config: config.clone(),
dimension,
}
}
pub fn insert_experience(&self, exp_id: ExperienceId, embedding: &[f32]) -> Result<()> {
if embedding.len() != self.dimension {
return Err(PulseDBError::vector(format!(
"Embedding dimension mismatch: expected {}, got {}",
self.dimension,
embedding.len()
)));
}
let mut state = self
.state
.write()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
if state.id_to_internal.contains_key(&exp_id) {
return Ok(());
}
let internal_id = state.next_id;
state.next_id += 1;
state.id_to_internal.insert(exp_id, internal_id);
state.internal_to_id.push(exp_id);
drop(state);
self.hnsw.insert((embedding, internal_id));
Ok(())
}
pub fn delete_experience(&self, exp_id: ExperienceId) -> Result<()> {
let mut state = self
.state
.write()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
if let Some(&internal_id) = state.id_to_internal.get(&exp_id) {
state.deleted.insert(internal_id);
}
Ok(())
}
pub fn search_experiences(
&self,
query: &[f32],
k: usize,
ef_search: usize,
) -> Result<Vec<(ExperienceId, f32)>> {
if query.len() != self.dimension {
return Err(PulseDBError::vector(format!(
"Query dimension mismatch: expected {}, got {}",
self.dimension,
query.len()
)));
}
let state = self
.state
.read()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
let active_count = state.next_id - state.deleted.len();
if active_count == 0 {
return Ok(vec![]);
}
let effective_k = k.min(active_count);
if active_count <= BRUTE_FORCE_THRESHOLD {
let dist_fn = DistCosine;
let mut all_distances: Vec<(ExperienceId, f32)> = Vec::with_capacity(active_count);
for point in self.hnsw.get_point_indexation().into_iter() {
let origin_id = point.get_origin_id();
if state.deleted.contains(&origin_id) {
continue;
}
let distance = dist_fn.eval(query, point.get_v());
if let Some(&exp_id) = state.internal_to_id.get(origin_id) {
all_distances.push((exp_id, distance));
}
}
all_distances
.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
all_distances.truncate(effective_k);
return Ok(all_distances);
}
let effective_ef = ef_search.max(effective_k);
let deleted_ref = &state.deleted;
let filter_fn = |id: &usize| -> bool { !deleted_ref.contains(id) };
let results = if state.deleted.is_empty() {
self.hnsw.search(query, effective_k, effective_ef)
} else {
self.hnsw
.search_filter(query, effective_k, effective_ef, Some(&filter_fn))
};
let mapped: Vec<(ExperienceId, f32)> = results
.into_iter()
.filter_map(|n| {
state
.internal_to_id
.get(n.d_id)
.map(|&exp_id| (exp_id, n.distance))
})
.collect();
Ok(mapped)
}
pub fn contains(&self, exp_id: ExperienceId) -> bool {
let state = self.state.read().ok();
state.is_some_and(|s| {
s.id_to_internal
.get(&exp_id)
.is_some_and(|id| !s.deleted.contains(id))
})
}
pub fn active_count(&self) -> usize {
let state = self.state.read().ok();
state.map_or(0, |s| s.id_to_internal.len() - s.deleted.len())
}
pub fn total_count(&self) -> usize {
self.hnsw.get_nb_point()
}
pub fn restore_deleted_set(&self, deleted_exp_ids: &[String]) -> Result<()> {
let mut state = self
.state
.write()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
for exp_id_str in deleted_exp_ids {
let uuid = uuid::Uuid::parse_str(exp_id_str)
.map_err(|e| PulseDBError::vector(format!("Invalid UUID in deleted set: {}", e)))?;
let exp_id = ExperienceId::from_bytes(*uuid.as_bytes());
if let Some(&internal_id) = state.id_to_internal.get(&exp_id) {
state.deleted.insert(internal_id);
}
}
Ok(())
}
pub fn save_to_dir(&self, dir: &Path, name: &str) -> Result<()> {
fs::create_dir_all(dir)
.map_err(|e| PulseDBError::vector(format!("Failed to create HNSW directory: {}", e)))?;
let state = self
.state
.read()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
let metadata = IndexMetadata {
dimension: self.dimension,
next_id: state.next_id,
id_map: state
.id_to_internal
.iter()
.map(|(exp_id, &internal_id)| (exp_id.to_string(), internal_id))
.collect(),
deleted: state
.deleted
.iter()
.filter_map(|&internal_id| {
state
.internal_to_id
.get(internal_id)
.map(|exp_id| exp_id.to_string())
})
.collect(),
};
let meta_path = dir.join(format!("{}.hnsw.meta", name));
let json = serde_json::to_string_pretty(&metadata).map_err(|e| {
PulseDBError::vector(format!("Failed to serialize HNSW metadata: {}", e))
})?;
fs::write(&meta_path, json)
.map_err(|e| PulseDBError::vector(format!("Failed to write HNSW metadata: {}", e)))?;
if state.id_to_internal.is_empty() {
return Ok(());
}
drop(state);
if let Err(e) = self.hnsw.file_dump(dir, name) {
tracing::warn!(error = %e, "Failed to dump HNSW graph (non-fatal, will rebuild on next open)");
}
Ok(())
}
#[allow(dead_code)] pub(crate) fn load_metadata(dir: &Path, name: &str) -> Result<Option<IndexMetadata>> {
let meta_path = dir.join(format!("{}.hnsw.meta", name));
if !meta_path.exists() {
return Ok(None);
}
let json = fs::read_to_string(&meta_path)
.map_err(|e| PulseDBError::vector(format!("Failed to read HNSW metadata: {}", e)))?;
let metadata: IndexMetadata = serde_json::from_str(&json)
.map_err(|e| PulseDBError::vector(format!("Failed to parse HNSW metadata: {}", e)))?;
Ok(Some(metadata))
}
pub fn rebuild_from_embeddings(
dimension: usize,
config: &HnswConfig,
embeddings: Vec<(ExperienceId, Vec<f32>)>,
) -> Result<Self> {
let index = Self::new(dimension, config);
if embeddings.is_empty() {
return Ok(index);
}
let mut state = index
.state
.write()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
let mut batch: Vec<(&Vec<f32>, usize)> = Vec::with_capacity(embeddings.len());
for (exp_id, embedding) in &embeddings {
let internal_id = state.next_id;
state.next_id += 1;
state.id_to_internal.insert(*exp_id, internal_id);
state.internal_to_id.push(*exp_id);
batch.push((embedding, internal_id));
}
drop(state);
index.hnsw.parallel_insert(&batch);
Ok(index)
}
pub fn remove_files(dir: &Path, name: &str) -> Result<()> {
let meta_path = dir.join(format!("{}.hnsw.meta", name));
if meta_path.exists() {
fs::remove_file(&meta_path).map_err(|e| {
PulseDBError::vector(format!("Failed to remove HNSW metadata: {}", e))
})?;
}
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
let file_name = entry.file_name();
let file_str = file_name.to_string_lossy();
if file_str.starts_with(name) && file_str.contains("hnswdump") {
let _ = fs::remove_file(entry.path());
}
}
}
Ok(())
}
}
impl VectorIndex for HnswIndex {
fn insert(&self, id: usize, embedding: &[f32]) -> Result<()> {
if embedding.len() != self.dimension {
return Err(PulseDBError::vector(format!(
"Embedding dimension mismatch: expected {}, got {}",
self.dimension,
embedding.len()
)));
}
self.hnsw.insert((embedding, id));
Ok(())
}
fn insert_batch(&self, items: &[(&Vec<f32>, usize)]) -> Result<()> {
self.hnsw.parallel_insert(items);
Ok(())
}
fn search(&self, query: &[f32], k: usize, ef_search: usize) -> Result<Vec<(usize, f32)>> {
let results = self.hnsw.search(query, k, ef_search);
Ok(results.into_iter().map(|n| (n.d_id, n.distance)).collect())
}
fn search_filtered(
&self,
query: &[f32],
k: usize,
ef_search: usize,
filter: &(dyn Fn(&usize) -> bool + Sync),
) -> Result<Vec<(usize, f32)>> {
let bridge = FilterBridge(filter);
let results = self.hnsw.search_filter(query, k, ef_search, Some(&bridge));
Ok(results.into_iter().map(|n| (n.d_id, n.distance)).collect())
}
fn delete(&self, id: usize) -> Result<()> {
let mut state = self
.state
.write()
.map_err(|_| PulseDBError::vector("Index state lock poisoned"))?;
state.deleted.insert(id);
Ok(())
}
fn is_deleted(&self, id: usize) -> bool {
self.state
.read()
.ok()
.is_some_and(|s| s.deleted.contains(&id))
}
fn len(&self) -> usize {
self.active_count()
}
fn save(&self, dir: &Path, name: &str) -> Result<()> {
self.save_to_dir(dir, name)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::HnswConfig;
fn test_config() -> HnswConfig {
HnswConfig {
max_nb_connection: 16,
ef_construction: 100,
ef_search: 50,
max_layer: 8,
max_elements: 1000,
}
}
fn make_embedding(seed: u64, dim: usize) -> Vec<f32> {
(0..dim)
.map(|i| (seed as f32 * 0.1 + i as f32 * 0.01).sin())
.collect()
}
#[test]
fn test_new_index_is_empty() {
let index = HnswIndex::new(384, &test_config());
assert_eq!(index.active_count(), 0);
assert_eq!(index.total_count(), 0);
assert!(index.is_empty());
}
#[test]
fn test_insert_and_search() {
let dim = 8;
let config = test_config();
let index = HnswIndex::new(dim, &config);
for i in 0..10u64 {
let exp_id = ExperienceId::new();
let embedding = make_embedding(i, dim);
index.insert_experience(exp_id, &embedding).unwrap();
}
assert_eq!(index.active_count(), 10);
let query = make_embedding(5, dim);
let results = index.search_experiences(&query, 3, 50).unwrap();
assert!(!results.is_empty());
assert!(results.len() <= 3);
for w in results.windows(2) {
assert!(w[0].1 <= w[1].1, "Results not sorted by distance");
}
}
#[test]
fn test_insert_idempotent() {
let dim = 4;
let index = HnswIndex::new(dim, &test_config());
let exp_id = ExperienceId::new();
let embedding = make_embedding(1, dim);
index.insert_experience(exp_id, &embedding).unwrap();
index.insert_experience(exp_id, &embedding).unwrap();
assert_eq!(index.active_count(), 1);
}
#[test]
fn test_dimension_mismatch_rejected() {
let index = HnswIndex::new(384, &test_config());
let exp_id = ExperienceId::new();
let wrong_dim = vec![1.0f32; 128];
let result = index.insert_experience(exp_id, &wrong_dim);
assert!(result.is_err());
assert!(result.unwrap_err().is_vector());
}
#[test]
fn test_delete_excludes_from_search() {
let dim = 8;
let index = HnswIndex::new(dim, &test_config());
let mut ids = Vec::new();
for i in 0..5u64 {
let exp_id = ExperienceId::new();
index
.insert_experience(exp_id, &make_embedding(i, dim))
.unwrap();
ids.push(exp_id);
}
assert_eq!(index.active_count(), 5);
index.delete_experience(ids[0]).unwrap();
assert_eq!(index.active_count(), 4);
assert!(!index.contains(ids[0]));
assert!(index.contains(ids[1]));
let query = make_embedding(0, dim); let results = index.search_experiences(&query, 10, 50).unwrap();
let result_ids: Vec<ExperienceId> = results.iter().map(|r| r.0).collect();
assert!(!result_ids.contains(&ids[0]));
}
#[test]
fn test_search_k_larger_than_index() {
let dim = 4;
let index = HnswIndex::new(dim, &test_config());
let exp_id = ExperienceId::new();
index
.insert_experience(exp_id, &make_embedding(1, dim))
.unwrap();
let results = index
.search_experiences(&make_embedding(1, dim), 100, 50)
.unwrap();
assert_eq!(results.len(), 1);
}
#[test]
fn test_search_empty_index() {
let dim = 4;
let index = HnswIndex::new(dim, &test_config());
let results = index
.search_experiences(&make_embedding(1, dim), 10, 50)
.unwrap();
assert!(results.is_empty());
}
#[test]
fn test_rebuild_from_embeddings() {
let dim = 8;
let config = test_config();
let embeddings: Vec<(ExperienceId, Vec<f32>)> = (0..20u64)
.map(|i| (ExperienceId::new(), make_embedding(i, dim)))
.collect();
let index = HnswIndex::rebuild_from_embeddings(dim, &config, embeddings.clone()).unwrap();
assert_eq!(index.active_count(), 20);
let query = make_embedding(10, dim);
let results = index.search_experiences(&query, 5, 50).unwrap();
assert!(!results.is_empty());
}
#[test]
fn test_rebuild_empty() {
let dim = 384;
let config = test_config();
let index = HnswIndex::rebuild_from_embeddings(dim, &config, vec![]).unwrap();
assert!(index.is_empty());
}
#[test]
fn test_save_and_load_metadata_roundtrip() {
let dim = 4;
let index = HnswIndex::new(dim, &test_config());
let mut exp_ids = Vec::new();
for i in 0..5u64 {
let exp_id = ExperienceId::new();
index
.insert_experience(exp_id, &make_embedding(i, dim))
.unwrap();
exp_ids.push(exp_id);
}
index.delete_experience(exp_ids[2]).unwrap();
let dir = tempfile::tempdir().unwrap();
index.save_to_dir(dir.path(), "test_collective").unwrap();
let metadata = HnswIndex::load_metadata(dir.path(), "test_collective")
.unwrap()
.expect("Metadata should exist");
assert_eq!(metadata.dimension, dim);
assert_eq!(metadata.next_id, 5);
assert_eq!(metadata.id_map.len(), 5);
assert_eq!(metadata.deleted.len(), 1);
assert_eq!(metadata.deleted[0], exp_ids[2].to_string());
}
#[test]
fn test_remove_files() {
let dim = 4;
let index = HnswIndex::new(dim, &test_config());
index
.insert_experience(ExperienceId::new(), &make_embedding(1, dim))
.unwrap();
let dir = tempfile::tempdir().unwrap();
index.save_to_dir(dir.path(), "test_coll").unwrap();
let meta_path = dir.path().join("test_coll.hnsw.meta");
assert!(meta_path.exists());
HnswIndex::remove_files(dir.path(), "test_coll").unwrap();
assert!(!meta_path.exists());
}
#[test]
fn test_brute_force_search_returns_all_items() {
let dim = 8;
let config = test_config();
let index = HnswIndex::new(dim, &config);
let mut ids = Vec::new();
for i in 0..20u64 {
let exp_id = ExperienceId::new();
index
.insert_experience(exp_id, &make_embedding(i, dim))
.unwrap();
ids.push(exp_id);
}
let query = make_embedding(10, dim);
let results = index.search_experiences(&query, 20, 50).unwrap();
assert_eq!(results.len(), 20, "Brute-force must return all 20 items");
for w in results.windows(2) {
assert!(
w[0].1 <= w[1].1,
"Brute-force results not sorted: {} > {}",
w[0].1,
w[1].1
);
}
assert_eq!(results[0].0, ids[10]);
assert!(
results[0].1 < 0.001,
"Expected near-zero distance for exact match, got {}",
results[0].1
);
}
#[test]
fn test_brute_force_excludes_deleted() {
let dim = 8;
let index = HnswIndex::new(dim, &test_config());
let mut ids = Vec::new();
for i in 0..5u64 {
let exp_id = ExperienceId::new();
index
.insert_experience(exp_id, &make_embedding(i, dim))
.unwrap();
ids.push(exp_id);
}
index.delete_experience(ids[2]).unwrap();
let query = make_embedding(2, dim);
let results = index.search_experiences(&query, 10, 50).unwrap();
assert_eq!(results.len(), 4, "Should return 4 after deleting 1 of 5");
let result_ids: Vec<ExperienceId> = results.iter().map(|r| r.0).collect();
assert!(
!result_ids.contains(&ids[2]),
"Deleted item must be excluded"
);
}
#[test]
fn test_cosine_distance_identical_vectors() {
let dim = 8;
let index = HnswIndex::new(dim, &test_config());
let embedding = make_embedding(42, dim);
let exp_id = ExperienceId::new();
index.insert_experience(exp_id, &embedding).unwrap();
let results = index.search_experiences(&embedding, 1, 50).unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].0, exp_id);
assert!(
results[0].1 < 0.001,
"Expected near-zero distance for identical vectors, got {}",
results[0].1
);
}
}