use std::sync::Arc;
use papaya::HashMap as ConcurrentMap;
use parking_lot::Mutex;
use super::{
hnsw::{HnswConfig, HnswIndex, decode_native, decode_values, distance},
vector_types::{VectorDistanceMetricType, VectorQuantType},
};
pub mod term {
pub const FULL_VECTOR: u64 = 0;
pub const NEIGHBOR_LIST: u64 = 1;
pub const QUANTIZED_VECTOR: u64 = 2;
pub const ATTRIBUTES: u64 = 3;
pub const METADATA: u64 = 4;
pub const INTERNAL_ID_MAP: u64 = 5;
pub const EXTERNAL_ID_MAP: u64 = 6;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DiskAnnInsertResult {
False = 0,
True = 1,
QuantizationRequested = 2,
}
#[derive(Debug, Clone, PartialEq)]
pub struct SearchHit {
pub external_id: Vec<u8>,
pub distance: f32,
}
struct DiskAnnIndex {
hnsw: Mutex<HnswIndex>,
external_to_internal: ConcurrentMap<Vec<u8>, u32>,
internal_to_external: ConcurrentMap<u32, Vec<u8>>,
attributes: ConcurrentMap<Vec<u8>, Vec<u8>>,
}
#[derive(Default)]
pub struct DiskANNService {
indexes: ConcurrentMap<u64, Arc<DiskAnnIndex>>,
}
impl DiskANNService {
#[allow(clippy::too_many_arguments)]
pub fn create_index(
&self,
context: u64,
dims: u32,
reduce_dims: u32,
quant: VectorQuantType,
build_exploration_factor: u32,
num_links: u32,
metric: VectorDistanceMetricType,
) -> bool {
let config = HnswConfig {
dims,
reduce_dims,
quant,
metric,
build_exploration_factor: build_exploration_factor.max(1),
num_links: num_links.max(1),
};
match self.indexes.pin().get(&context).cloned() {
Some(existing) => {
existing.hnsw.lock().clear_for_recreate(config);
true
}
None => {
let index = DiskAnnIndex {
hnsw: Mutex::new(HnswIndex::new(config)),
external_to_internal: ConcurrentMap::new(),
internal_to_external: ConcurrentMap::new(),
attributes: ConcurrentMap::new(),
};
self.indexes.pin().insert(context, Arc::new(index));
true
}
}
}
pub fn drop_index(&self, context: u64) {
self.indexes.pin().remove(&context);
}
pub fn insert(
&self,
context: u64,
external_id: &[u8],
vector: &[u8],
attributes: &[u8],
) -> DiskAnnInsertResult {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return DiskAnnInsertResult::False;
};
let mut hnsw = index.hnsw.lock();
let dims = hnsw.config().dims as usize;
if index.external_to_internal.pin().contains_key(external_id)
|| decode_native_len(vector, hnsw.config().quant) != dims
{
return DiskAnnInsertResult::False;
}
let needs_quantization = !hnsw.quant_table_ready();
let internal_id = {
let mut rng = fastrand::Rng::new();
hnsw.insert(vector, &mut rng)
};
index
.external_to_internal
.pin()
.insert(external_id.to_vec(), internal_id);
index
.internal_to_external
.pin()
.insert(internal_id, external_id.to_vec());
if !attributes.is_empty() {
index
.attributes
.pin()
.insert(external_id.to_vec(), attributes.to_vec());
}
if needs_quantization {
DiskAnnInsertResult::QuantizationRequested
} else {
DiskAnnInsertResult::True
}
}
pub fn remove(&self, context: u64, external_id: &[u8]) -> bool {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return false;
};
let internal_id = match index.external_to_internal.pin().remove(external_id) {
Some(&id) => id,
None => return false,
};
index.internal_to_external.pin().remove(&internal_id);
index.attributes.pin().remove(external_id);
index.hnsw.lock().remove(internal_id)
}
pub fn needs_quantization(&self, context: u64) -> bool {
self
.indexes
.pin()
.get(&context)
.is_some_and(|i| !i.hnsw.lock().quant_table_ready())
}
pub fn build_quantization_table(&self, context: u64) -> bool {
self
.indexes
.pin()
.get(&context)
.is_some_and(|i| i.hnsw.lock().build_quant_table())
}
pub fn backfill_quantized_vectors(&self, context: u64, task_index: usize, task_count: usize) {
if task_count == 0 || !task_index.is_multiple_of(task_count) {
return;
}
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return;
};
index.hnsw.lock().build_quant_table();
}
pub fn backfill_quant_vectors(&self, context: u64, task_index: usize, task_count: usize) {
self.backfill_quantized_vectors(context, task_index, task_count)
}
pub fn search_vector(
&self,
context: u64,
vector: &[u8],
count: usize,
search_exploration_factor: usize,
predicate: &mut dyn FnMut(&[u8]) -> bool,
) -> Result<Vec<SearchHit>, i32> {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return Err(-1);
};
let query = {
let hnsw = index.hnsw.lock();
if decode_native_len(vector, hnsw.config().quant) != hnsw.config().dims as usize {
return Err(-1);
}
decode_native(vector, hnsw.config().quant)
};
Self::search_values(&index, &query, count, search_exploration_factor, predicate)
}
pub fn search_element(
&self,
context: u64,
external_id: &[u8],
count: usize,
search_exploration_factor: usize,
predicate: &mut dyn FnMut(&[u8]) -> bool,
) -> Result<Vec<SearchHit>, i32> {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return Err(-1);
};
let internal = match index.external_to_internal.pin().get(external_id).copied() {
Some(id) => id,
None => return Err(-1),
};
let query = {
let hnsw = index.hnsw.lock();
let Some(bytes) = hnsw.vector_of(internal) else {
return Err(-1);
};
decode_values(bytes, hnsw.config().quant, hnsw.quant_table())
};
Self::search_values(&index, &query, count, search_exploration_factor, predicate)
}
fn search_values(
index: &Arc<DiskAnnIndex>,
query: &[f32],
count: usize,
search_exploration_factor: usize,
predicate: &mut dyn FnMut(&[u8]) -> bool,
) -> Result<Vec<SearchHit>, i32> {
let internal = index.internal_to_external.pin();
let mut internal_predicate = |id: u32| internal.get(&id).is_some_and(|eid| predicate(eid));
let hits = index.hnsw.lock().search(
query,
count,
search_exploration_factor,
&mut internal_predicate,
);
drop(internal);
let ext = index.internal_to_external.pin();
Ok(
hits
.into_iter()
.filter_map(|(id, dist)| {
ext.get(&id).map(|eid| SearchHit {
external_id: eid.clone(),
distance: dist,
})
})
.collect(),
)
}
pub fn continue_search(&self, _context: u64, _continuation: u64) -> Result<Vec<SearchHit>, i32> {
Err(-1)
}
pub fn check_internal_id_valid(&self, context: u64, internal_id: u32) -> bool {
self
.indexes
.pin()
.get(&context)
.is_some_and(|i| i.hnsw.lock().is_internal_id_valid(internal_id))
}
pub fn check_external_id_valid(&self, context: u64, external_id: &[u8]) -> bool {
self
.indexes
.pin()
.get(&context)
.is_some_and(|i| i.external_to_internal.pin().contains_key(external_id))
}
pub fn set_attribute(&self, context: u64, external_id: &[u8], attribute: &[u8]) -> bool {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return false;
};
if !index.external_to_internal.pin().contains_key(external_id) {
return false;
}
index
.attributes
.pin()
.insert(external_id.to_vec(), attribute.to_vec());
true
}
pub fn get_attribute(&self, context: u64, external_id: &[u8]) -> Option<Vec<u8>> {
let index = self.indexes.pin().get(&context).cloned()?;
index.attributes.pin().get(external_id).cloned()
}
pub fn get_full_vector(&self, context: u64, external_id: &[u8]) -> Option<Vec<u8>> {
let index = self.indexes.pin().get(&context).cloned()?;
let internal = *index.external_to_internal.pin().get(external_id)?;
let hnsw = index.hnsw.lock();
hnsw.vector_of(internal).map(|v| v.to_vec())
}
pub fn card(&self, context: u64) -> u64 {
self
.indexes
.pin()
.get(&context)
.map_or(0, |i| i.hnsw.lock().len() as u64)
}
pub fn links_of(&self, context: u64, external_id: &[u8]) -> Option<Vec<Vec<u8>>> {
let index = self.indexes.pin().get(&context).cloned()?;
let internal = *index.external_to_internal.pin().get(external_id)?;
let hnsw = index.hnsw.lock();
let links = hnsw.links_of(internal)?;
let ext = index.internal_to_external.pin();
Some(links.iter().filter_map(|id| ext.get(id).cloned()).collect())
}
pub fn sample(&self, context: u64, count: usize) -> Vec<Vec<u8>> {
let Some(index) = self.indexes.pin().get(&context).cloned() else {
return Vec::new();
};
let ids = index.hnsw.lock().sample(count);
let ext = index.internal_to_external.pin();
ids.iter().filter_map(|id| ext.get(id).cloned()).collect()
}
pub fn internal_id_of(&self, context: u64, external_id: &[u8]) -> Option<u32> {
let index = self.indexes.pin().get(&context).cloned()?;
index.external_to_internal.pin().get(external_id).copied()
}
pub fn distance_between(&self, context: u64, a: &[u8], b: &[u8]) -> Option<f32> {
let index = self.indexes.pin().get(&context).cloned()?;
let va = self.embedding_of(context, a)?;
let vb = self.embedding_of(context, b)?;
let metric = index.hnsw.lock().config().metric;
Some(distance(&va, &vb, metric))
}
pub fn dims_of(&self, context: u64) -> Option<u32> {
self
.indexes
.pin()
.get(&context)
.map(|i| i.hnsw.lock().config().dims)
}
pub fn quant_of(&self, context: u64) -> Option<VectorQuantType> {
self
.indexes
.pin()
.get(&context)
.map(|i| i.hnsw.lock().config().quant)
}
pub fn embedding_of(&self, context: u64, external_id: &[u8]) -> Option<Vec<f32>> {
let index = self.indexes.pin().get(&context).cloned()?;
let internal = *index.external_to_internal.pin().get(external_id)?;
let hnsw = index.hnsw.lock();
let bytes = hnsw.vector_of(internal)?;
Some(decode_values(
bytes,
hnsw.config().quant,
hnsw.quant_table(),
))
}
}
fn decode_native_len(bytes: &[u8], quant: VectorQuantType) -> usize {
match quant {
VectorQuantType::XNoQuant_U8
| VectorQuantType::XBin_U8
| VectorQuantType::XNoQuant_I8
| VectorQuantType::XBin_I8 => bytes.len(),
_ => bytes.len() / 4,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn svc() -> DiskANNService {
DiskANNService::default()
}
fn f32_bytes(vals: &[f32]) -> Vec<u8> {
vals.iter().flat_map(|v| v.to_le_bytes()).collect()
}
#[test]
fn create_insert_search_roundtrip() {
let service = svc();
assert!(service.create_index(
8,
2,
0,
VectorQuantType::NoQuant,
64,
8,
VectorDistanceMetricType::L2
));
let res = service.insert(8, b"a", &f32_bytes(&[0.0, 0.0]), b"{\"k\":1}");
assert_eq!(res, DiskAnnInsertResult::True);
let res = service.insert(8, b"b", &f32_bytes(&[1.0, 1.0]), b"");
assert_eq!(res, DiskAnnInsertResult::True);
let res = service.insert(8, b"a", &f32_bytes(&[5.0, 5.0]), b"");
assert_eq!(res, DiskAnnInsertResult::False);
let res = service.insert(8, b"c", &f32_bytes(&[1.0]), b"");
assert_eq!(res, DiskAnnInsertResult::False);
assert_eq!(service.card(8), 2);
assert!(service.check_external_id_valid(8, b"a"));
assert!(!service.check_external_id_valid(8, b"zz"));
let hits = service
.search_vector(8, &f32_bytes(&[0.1, 0.1]), 2, 32, &mut |_| true)
.unwrap();
assert_eq!(hits.len(), 2);
assert_eq!(hits[0].external_id, b"a".to_vec());
let hits = service
.search_element(8, b"b", 1, 32, &mut |_| true)
.unwrap();
assert_eq!(hits[0].external_id, b"b".to_vec());
let hits = service
.search_vector(8, &f32_bytes(&[0.0, 0.0]), 2, 32, &mut |id| id == b"b")
.unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].external_id, b"b".to_vec());
assert_eq!(
service.search_vector(8, &f32_bytes(&[1.0]), 1, 8, &mut |_| true),
Err(-1)
);
}
#[test]
fn attributes_and_embeddings() {
let service = svc();
service.create_index(
3,
2,
0,
VectorQuantType::NoQuant,
32,
4,
VectorDistanceMetricType::Cosine,
);
service.insert(3, b"x", &f32_bytes(&[1.0, 0.0]), b"{\"t\":\"a\"}");
service.insert(3, b"y", &f32_bytes(&[0.0, 1.0]), b"{\"t\":\"b\"}");
assert_eq!(
service.get_attribute(3, b"x").unwrap(),
b"{\"t\":\"a\"}".to_vec()
);
assert!(service.set_attribute(3, b"x", b"{\"t\":\"c\"}"));
assert_eq!(
service.get_attribute(3, b"x").unwrap(),
b"{\"t\":\"c\"}".to_vec()
);
assert!(!service.set_attribute(3, b"nope", b"{}"));
let emb = service.embedding_of(3, b"y").unwrap();
assert_eq!(emb, vec![0.0, 1.0]);
let links = service.links_of(3, b"x").unwrap();
assert!(!links.is_empty());
assert_eq!(service.sample(3, 5).len(), 2);
let d = service.distance_between(3, b"x", b"y").unwrap();
assert!((d - 1.0).abs() < 1e-5);
assert_eq!(service.dims_of(3), Some(2));
assert_eq!(service.quant_of(3), Some(VectorQuantType::NoQuant));
}
#[test]
fn quantization_request_lifecycle() {
let service = svc();
service.create_index(
5,
2,
0,
VectorQuantType::Q8,
32,
4,
VectorDistanceMetricType::L2,
);
assert!(service.needs_quantization(5));
let res = service.insert(5, b"p", &f32_bytes(&[0.0, 100.0]), b"");
assert_eq!(res, DiskAnnInsertResult::QuantizationRequested);
let res = service.insert(5, b"q", &f32_bytes(&[100.0, 0.0]), b"");
assert_eq!(res, DiskAnnInsertResult::QuantizationRequested);
assert!(service.build_quantization_table(5));
assert!(!service.needs_quantization(5));
let res = service.insert(5, b"r", &f32_bytes(&[50.0, 50.0]), b"");
assert_eq!(res, DiskAnnInsertResult::True);
assert_eq!(service.get_full_vector(5, b"r").unwrap().len(), 2);
service.backfill_quantized_vectors(5, 0, 4);
service.backfill_quantized_vectors(5, 1, 4);
assert!(service.build_quantization_table(5));
let hits = service
.search_vector(5, &f32_bytes(&[0.0, 100.0]), 1, 32, &mut |_| true)
.unwrap();
assert_eq!(hits[0].external_id, b"p".to_vec());
let hits = service
.search_element(5, b"q", 1, 32, &mut |_| true)
.unwrap();
assert_eq!(hits[0].external_id, b"q".to_vec());
let emb = service.embedding_of(5, b"p").unwrap();
assert_eq!(emb.len(), 2);
let d_self = service.distance_between(5, b"p", b"p").unwrap();
assert!(d_self.abs() < 1.0);
let d_cross = service.distance_between(5, b"p", b"q").unwrap();
assert!(d_cross > d_self);
}
#[test]
fn remove_and_drop() {
let service = svc();
service.create_index(
9,
1,
0,
VectorQuantType::XNoQuant_U8,
32,
4,
VectorDistanceMetricType::L2,
);
assert_eq!(
service.insert(9, b"u", &[7u8], b""),
DiskAnnInsertResult::True
);
assert_eq!(
service.insert(9, b"v", &[9u8], b""),
DiskAnnInsertResult::True
);
assert_eq!(service.card(9), 2);
assert!(service.remove(9, b"u"));
assert!(!service.remove(9, b"u"));
assert_eq!(service.card(9), 1);
assert!(service.get_full_vector(9, b"u").is_none());
service.drop_index(9);
assert_eq!(service.card(9), 0);
assert_eq!(service.search_vector(9, &[1], 1, 8, &mut |_| true), Err(-1));
assert!(!service.check_internal_id_valid(9, 0));
}
#[test]
fn continue_search_unsupported_and_id_channel() {
let service = svc();
assert_eq!(service.continue_search(1, 7), Err(-1));
service.create_index(
2,
1,
0,
VectorQuantType::NoQuant,
16,
2,
VectorDistanceMetricType::L2,
);
service.insert(2, b"s", &f32_bytes(&[1.0]), b"");
assert_eq!(service.internal_id_of(2, b"s"), Some(0));
assert_eq!(service.internal_id_of(2, b"none"), None);
assert!(service.check_internal_id_valid(2, 0));
assert!(!service.check_internal_id_valid(2, 42));
}
}