use std::cmp::Ordering;
use std::sync::Arc;
use durability::{Directory, PersistenceResult};
use segstore::{SegmentedStore, Store};
use crate::{AxisBox, IndexParams, RegionIndex, SearchParams};
struct BoxBacking;
impl Store for BoxBacking {
type Id = u32;
type Item = AxisBox;
type Segment = Vec<(u32, AxisBox)>;
fn build_segment(&self, batch: &[(u32, AxisBox)]) -> Vec<(u32, AxisBox)> {
batch.to_vec()
}
fn merge_segments(
&self,
segs: &[Vec<(u32, AxisBox)>],
live: &dyn Fn(&u32) -> bool,
) -> Vec<(u32, AxisBox)> {
segs.iter()
.flatten()
.filter(|(id, _)| live(id))
.cloned()
.collect()
}
}
pub struct UpdatableIndex {
inner: SegmentedStore<BoxBacking>,
dim: usize,
m: usize,
m_max: usize,
ef_construction: usize,
}
impl UpdatableIndex {
pub fn open(
dir: Arc<dyn Directory>,
flush_threshold: usize,
dim: usize,
params: IndexParams,
) -> PersistenceResult<Self> {
Ok(Self {
inner: SegmentedStore::open(dir, BoxBacking, flush_threshold)?,
dim,
m: params.m,
m_max: params.m_max,
ef_construction: params.ef_construction,
})
}
pub fn add(&mut self, id: u32, region: AxisBox) -> PersistenceResult<()> {
self.inner.add(id, region)
}
pub fn delete(&mut self, id: u32) -> PersistenceResult<()> {
self.inner.delete(id)
}
pub fn compact(&mut self) -> PersistenceResult<()> {
self.inner.compact()
}
pub fn checkpoint(&mut self) -> PersistenceResult<()> {
self.inner.checkpoint()
}
pub fn search(&self, query: &[f32], k: usize, params: SearchParams) -> Vec<(u32, f32)> {
let (ef, overretrieve) = (params.ef, params.overretrieve);
let mut cand: Vec<(u32, f32)> = Vec::new();
for seg in self.inner.segments() {
cand.extend(self.search_batch(seg, query, k, ef, overretrieve));
}
let buffered = self.inner.buffer().to_vec();
cand.extend(self.search_batch(&buffered, query, k, ef, overretrieve));
cand.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(Ordering::Equal));
cand.truncate(k);
cand
}
fn search_batch(
&self,
batch: &[(u32, AxisBox)],
query: &[f32],
k: usize,
ef: usize,
overretrieve: usize,
) -> Vec<(u32, f32)> {
let params = IndexParams {
m: self.m,
m_max: self.m_max,
ef_construction: self.ef_construction,
};
let mut idx = match RegionIndex::<AxisBox>::new(self.dim, params) {
Ok(i) => i,
Err(_) => return Vec::new(),
};
let mut any = false;
for (id, region) in batch {
if self.inner.is_live(id) && idx.add(*id, region.clone()).is_ok() {
any = true;
}
}
if !any || idx.build().is_err() {
return Vec::new();
}
idx.search(query, k, SearchParams { ef, overretrieve })
.unwrap_or_default()
}
}
#[cfg(test)]
mod tests {
use super::*;
use durability::MemoryDirectory;
fn b(lo: f32, hi: f32) -> AxisBox {
AxisBox::new(vec![lo, lo], vec![hi, hi])
}
#[test]
fn add_delete_compact_recover_through_real_region_index() {
let dir = MemoryDirectory::arc();
{
let mut store =
UpdatableIndex::open(dir.clone(), 2, 2, IndexParams::default()).unwrap();
store.add(0, b(0.0, 1.0)).unwrap(); store.add(1, b(5.0, 6.0)).unwrap(); store.add(2, b(10.0, 11.0)).unwrap();
let top: Vec<u32> = store
.search(&[0.5, 0.5], 1, SearchParams::default())
.into_iter()
.map(|(id, _)| id)
.collect();
assert_eq!(top, vec![0], "point inside box 0 retrieves region 0");
store.delete(0).unwrap();
let top: Vec<u32> = store
.search(&[0.5, 0.5], 1, SearchParams::default())
.into_iter()
.map(|(id, _)| id)
.collect();
assert_eq!(top, vec![1], "after deleting 0, nearest region is 1");
store.compact().unwrap();
assert_eq!(
store
.search(&[0.5, 0.5], 1, SearchParams::default())
.first()
.map(|(id, _)| *id),
Some(1)
);
}
let store = UpdatableIndex::open(dir, 2, 2, IndexParams::default()).unwrap();
let top: Vec<u32> = store
.search(&[0.5, 0.5], 1, SearchParams::default())
.into_iter()
.map(|(id, _)| id)
.collect();
assert_eq!(top, vec![1], "recovery preserves the search");
}
}