use std::collections::{HashMap, HashSet};
use yo_common::{Code, Error, Result};
use crate::coarse::Coarse;
use crate::dist::sqdist;
use crate::rabitq::{Bits, Coded, Quantizer};
pub trait Vectors {
fn get(&self, id: u64, into: &mut [f32]) -> bool;
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Tuning {
pub posting: usize,
pub probe: usize,
pub rerank: usize,
pub sweep: usize,
pub widen: usize,
pub patience: usize,
pub spill: usize,
pub slack: f32,
}
impl Default for Tuning {
fn default() -> Tuning {
Tuning {
posting: 256,
probe: 8,
rerank: 4,
sweep: 4,
widen: 8,
spill: 4,
slack: 0.10,
patience: 8,
}
}
}
const FLOOR: usize = 32;
pub trait Filter {
fn allows(&self, tag: u64) -> bool;
fn exact(&self, _id: u64) -> bool {
true
}
fn narrowing(&self) -> bool {
true
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct Any;
impl Filter for Any {
fn allows(&self, _tag: u64) -> bool {
true
}
fn narrowing(&self) -> bool {
false
}
}
impl<F: Fn(u64) -> bool> Filter for F {
fn allows(&self, tag: u64) -> bool {
self(tag)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Signature(u64);
impl Signature {
#[must_use]
pub fn of(values: &[(&str, &[u8])]) -> Signature {
let mut got = Signature(0);
for (attribute, value) in values {
got.insert(attribute, value);
}
got
}
pub fn insert(&mut self, attribute: &str, value: &[u8]) {
self.insert_bytes(attribute.as_bytes(), value);
}
pub fn insert_bytes(&mut self, attribute: &[u8], value: &[u8]) {
self.0 |= 1u64 << (hash(attribute, value) % 64);
}
#[must_use]
pub fn bits(self) -> u64 {
self.0
}
#[must_use]
pub fn from_bits(bits: u64) -> Signature {
Signature(bits)
}
#[must_use]
pub fn covers(self, want: Signature) -> bool {
self.0 & want.0 == want.0
}
}
impl Filter for Signature {
fn allows(&self, tag: u64) -> bool {
Signature(tag).covers(*self)
}
}
fn hash(attribute: &[u8], value: &[u8]) -> u64 {
let mut h = 0xcbf2_9ce4_8422_2325u64;
for byte in attribute.iter().chain(b":").chain(value) {
h ^= u64::from(*byte);
h = h.wrapping_mul(0x1000_0000_01b3);
}
h
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Hit {
pub id: u64,
pub distance: f32,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Work {
pub probed: usize,
pub scanned: usize,
}
#[derive(Debug, Clone, Copy)]
struct Place {
partition: u32,
slot: u32,
next: u32,
}
const END: u32 = u32::MAX;
#[derive(Debug, Default)]
struct Posting {
ids: Vec<u64>,
tags: Vec<u64>,
codes: Vec<u8>,
meta: Vec<Coded>,
stuck: usize,
}
impl Posting {
fn len(&self) -> usize {
self.ids.len()
}
}
#[derive(Debug)]
pub struct Partitions {
quant: Quantizer,
tuning: Tuning,
centroids: Vec<f32>,
postings: Vec<Posting>,
at: HashMap<u64, u32>,
places: Vec<Place>,
free: u32,
coarse: Coarse,
scratch: Vec<u32>,
spare: Vec<Place>,
spill: Vec<(usize, f32)>,
big: Vec<u32>,
small: Vec<u32>,
}
impl Partitions {
#[must_use]
pub fn new(dim: usize, bits: Bits, seed: u64, tuning: Tuning) -> Partitions {
Partitions {
quant: Quantizer::new(dim, bits, seed),
tuning,
centroids: Vec::new(),
postings: Vec::new(),
at: HashMap::new(),
places: Vec::new(),
free: END,
coarse: Coarse::default(),
scratch: Vec::new(),
spare: Vec::new(),
spill: Vec::new(),
big: Vec::new(),
small: Vec::new(),
}
}
#[must_use]
pub fn dim(&self) -> usize {
self.quant.dim()
}
#[must_use]
pub fn len(&self) -> usize {
self.at.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.at.is_empty()
}
#[must_use]
pub fn partitions(&self) -> usize {
self.postings.len()
}
#[must_use]
pub fn entries(&self) -> usize {
self.postings.iter().map(Posting::len).sum()
}
#[must_use]
pub fn tuning(&self) -> Tuning {
self.tuning
}
pub fn retune(&mut self, tuning: Tuning) {
self.tuning = tuning;
}
#[must_use]
pub fn quantizer(&self) -> &Quantizer {
&self.quant
}
#[must_use]
pub fn code_bytes(&self) -> usize {
self.postings.iter().map(|p| p.codes.len()).sum()
}
pub fn insert(&mut self, id: u64, v: &[f32]) {
self.insert_tagged(id, v, 0);
}
pub fn insert_tagged(&mut self, id: u64, v: &[f32], tag: u64) {
assert_eq!(
v.len(),
self.dim(),
"this collection holds {} dimensional vectors and was handed {}",
self.dim(),
v.len()
);
self.remove(id);
let x = self.quant.rotate(v);
if self.postings.is_empty() {
let p = self.add_partition(&x);
self.place(p, id, tag, &x);
return;
}
let mut into = core::mem::take(&mut self.spill);
self.spill_into(&x, &mut into);
for &(p, _) in &into {
self.place(p, id, tag, &x);
}
self.spill = into;
}
fn spill_into(&mut self, x: &[f32], into: &mut Vec<(usize, f32)>) {
into.clear();
let dim = self.dim();
let want = self.tuning.spill.max(1);
let mut short = core::mem::take(&mut self.scratch);
if want == 1 || self.tuning.slack <= 0.0 {
let p = self.roughly_nearest(x, &mut short);
self.scratch = short;
into.push((p, 0.0));
return;
}
self.coarse.shortlist(x, dim, &mut short);
let mut near: Vec<(usize, f32)> = if short.is_empty() {
(0..self.postings.len())
.map(|p| (p, sqdist(x, self.centroid(p))))
.collect()
} else {
short
.iter()
.map(|&p| (p as usize, sqdist(x, self.centroid(p as usize))))
.collect()
};
self.scratch = short;
near.sort_unstable_by(|a, b| a.1.total_cmp(&b.1));
let Some(&(first, best)) = near.first() else {
return;
};
into.push((first, best));
let ceiling = best * (1.0 + self.tuning.slack) * (1.0 + self.tuning.slack);
for &(q, d) in near.iter().skip(1) {
if into.len() >= want {
break;
}
if d > ceiling {
break;
}
into.push((q, d));
}
}
#[must_use]
pub fn tag(&self, id: u64) -> Option<u64> {
let at = self.any_place(id)?;
Some(self.postings[at.partition as usize].tags[at.slot as usize])
}
pub fn retag(&mut self, id: u64, tag: u64) -> bool {
let mut walk = self.at.get(&id).copied().unwrap_or(END);
let mut found = false;
while walk != END {
let place = self.places[walk as usize];
self.postings[place.partition as usize].tags[place.slot as usize] = tag;
found = true;
walk = place.next;
}
found
}
pub fn remove(&mut self, id: u64) -> bool {
let mut copies = core::mem::take(&mut self.spare);
self.every_place(id, &mut copies);
if copies.is_empty() {
self.spare = copies;
return false;
}
self.detach_all(id);
copies.sort_unstable_by_key(|c| core::cmp::Reverse(c.slot));
for copy in &copies {
let p = copy.partition as usize;
let s = copy.slot as usize;
if let Some(moved) = self.pull(p, s) {
self.reslot(moved, p, s);
}
}
self.spare = copies;
true
}
#[must_use]
pub fn contains(&self, id: u64) -> bool {
self.at.contains_key(&id)
}
pub(crate) fn all_centroids(&self) -> &[f32] {
&self.centroids
}
pub(crate) fn posting_parts(&self, p: usize) -> (&[u64], &[u64], &[u8], &[Coded], usize) {
let posting = &self.postings[p];
(
&posting.ids,
&posting.tags,
&posting.codes,
&posting.meta,
posting.stuck,
)
}
pub(crate) fn absorb(
&mut self,
centroid: &[f32],
ids: Vec<u64>,
tags: Vec<u64>,
codes: Vec<u8>,
meta: Vec<Coded>,
stuck: usize,
) -> Result<()> {
let width = self.quant.code_bytes();
if centroid.len() != self.dim()
|| tags.len() != ids.len()
|| meta.len() != ids.len()
|| codes.len() != ids.len() * width
{
return Err(Error::new(
Code::Corrupt,
"the parts of a partition do not describe the same members",
)
.with_detail(format!(
"centroid={} ids={} tags={} codes={} meta={}",
centroid.len(),
ids.len(),
tags.len(),
codes.len(),
meta.len()
)));
}
let p = self.postings.len();
for (slot, &id) in ids.iter().enumerate() {
if !self.attach(id, p, slot) {
return Err(Error::new(
Code::Corrupt,
"an id is twice in one partition of an image",
)
.with_detail(format!("id={id} partition={p}")));
}
}
self.centroids.extend_from_slice(centroid);
self.postings.push(Posting {
ids,
tags,
codes,
meta,
stuck,
});
Ok(())
}
pub(crate) fn finish_image(&mut self) {
let dim = self.quant.dim();
let n = self.postings.len();
self.coarse.rebuild(&self.centroids, dim, n);
self.big.clear();
self.small.clear();
for p in 0..n {
if self.over(p) {
self.big.push(p as u32);
}
if self.under(p) {
self.small.push(p as u32);
}
}
}
#[must_use]
pub fn search(&self, q: &[f32], k: usize, vectors: &impl Vectors) -> Vec<Hit> {
self.search_where(q, k, &Any, vectors)
}
#[must_use]
pub fn search_where(
&self,
q: &[f32],
k: usize,
filter: &impl Filter,
vectors: &impl Vectors,
) -> Vec<Hit> {
self.search_costed(q, k, filter, vectors).0
}
#[must_use]
pub fn search_costed(
&self,
q: &[f32],
k: usize,
filter: &impl Filter,
vectors: &impl Vectors,
) -> (Vec<Hit>, Work) {
if k == 0 {
return (Vec::new(), Work::default());
}
let (candidates, work) =
self.candidates_costed(q, (k * self.tuning.rerank).max(FLOOR), filter);
let mut buf = vec![0.0f32; self.dim()];
let mut hits = Vec::with_capacity(candidates.len());
for (id, _) in candidates {
if vectors.get(id, &mut buf) {
hits.push(Hit {
id,
distance: sqdist(q, &buf),
});
}
}
hits.sort_by(|a, b| a.distance.total_cmp(&b.distance));
hits.truncate(k);
(hits, work)
}
#[must_use]
pub fn candidates(&self, q: &[f32], want: usize) -> Vec<(u64, f32)> {
self.candidates_where(q, want, &Any)
}
#[must_use]
pub fn candidates_where(
&self,
q: &[f32],
want: usize,
filter: &impl Filter,
) -> Vec<(u64, f32)> {
self.candidates_costed(q, want, filter).0
}
#[must_use]
pub fn candidates_costed(
&self,
q: &[f32],
want: usize,
filter: &impl Filter,
) -> (Vec<(u64, f32)>, Work) {
assert_eq!(
q.len(),
self.dim(),
"this collection holds {} dimensional vectors and was handed {}",
self.dim(),
q.len()
);
if want == 0 || self.postings.is_empty() {
return (Vec::new(), Work::default());
}
let u = self.quant.rotate(q);
let mut best = Bounded::new(want);
let mut scores: Vec<f32> = Vec::new();
let reach = if filter.narrowing() {
self.tuning.probe.saturating_mul(self.tuning.widen.max(1))
} else {
self.tuning.probe
};
let mut work = Work::default();
let mut quiet = 0;
for (n, p) in self.near_partitions(&u, reach).into_iter().enumerate() {
if best.full()
&& (n >= self.tuning.probe
|| (self.tuning.patience > 0 && quiet >= self.tuning.patience))
{
break;
}
let prepared = self.quant.query_rotated(&u, self.centroid(p));
let posting = &self.postings[p];
let held = posting.ids.len();
if scores.len() < held {
scores.resize(held, 0.0);
}
prepared.scan(&posting.codes, &posting.meta, &mut scores[..held]);
work.probed += 1;
work.scanned += held;
let mut took = 0;
for (i, &at) in scores[..held].iter().enumerate() {
if !best.wants(at) {
continue;
}
if !filter.allows(posting.tags[i]) {
continue;
}
if !filter.exact(posting.ids[i]) {
continue;
}
best.put(posting.ids[i], at);
took += 1;
}
quiet = if took == 0 { quiet + 1 } else { 0 };
}
let mut out = best.sorted();
if self.tuning.spill > 1 {
let mut seen = HashSet::with_capacity(out.len());
out.retain(|&(id, _)| seen.insert(id));
}
(out, work)
}
#[must_use]
pub fn needs_maintenance(&self) -> bool {
self.big.iter().any(|&p| {
let p = p as usize;
self.over(p) && self.postings[p].len() > self.postings[p].stuck
}) || (self.postings.len() > 1 && self.small.iter().any(|&p| self.under(p as usize)))
}
fn over(&self, p: usize) -> bool {
self.postings
.get(p)
.is_some_and(|posting| posting.len() > self.tuning.posting * 2)
}
fn under(&self, p: usize) -> bool {
self.postings
.get(p)
.is_some_and(|posting| posting.len() * 4 < self.tuning.posting)
}
pub fn maintain(&mut self, vectors: &impl Vectors, budget: usize) -> usize {
let mut done = 0;
while done < budget {
let Some(job) = self.job() else { break };
done += match job {
Job::Split(p) => self.split(p, vectors),
Job::Merge(p) => self.merge(p, vectors),
};
}
done
}
fn note(&mut self, p: usize) {
let (over, under) = (self.over(p), self.under(p));
let p = p as u32;
if over && !self.big.contains(&p) {
self.big.push(p);
}
if under && !self.small.contains(&p) {
self.small.push(p);
}
}
fn job(&mut self) -> Option<Job> {
let mut big = std::mem::take(&mut self.big);
big.retain(|&p| self.over(p as usize));
big.sort_unstable();
let split = big
.iter()
.map(|&p| p as usize)
.filter(|&p| self.postings[p].len() > self.postings[p].stuck)
.max_by_key(|&p| self.postings[p].len());
self.big = big;
if let Some(split) = split {
return Some(Job::Split(split));
}
if self.postings.len() > 1 {
let mut small = std::mem::take(&mut self.small);
small.retain(|&p| self.under(p as usize));
small.sort_unstable();
let merge = small
.iter()
.map(|&p| p as usize)
.min_by_key(|&p| self.postings[p].len());
self.small = small;
if let Some(merge) = merge {
return Some(Job::Merge(merge));
}
}
None
}
fn split(&mut self, p: usize, vectors: &impl Vectors) -> usize {
let (members, xs) = self.take(p, vectors);
let dim = self.dim();
if members.len() < 2 {
for (i, m) in members.iter().enumerate() {
self.place(p, m.id, m.tag, &xs[i * dim..(i + 1) * dim]);
}
return members.len();
}
let (a, b) = two_means(&xs, dim);
let sides: Vec<bool> = (0..members.len())
.map(|i| {
sqdist(&xs[i * dim..(i + 1) * dim], &a) <= sqdist(&xs[i * dim..(i + 1) * dim], &b)
})
.collect();
let small = sides.iter().filter(|&&s| s).count();
let small = small.min(members.len() - small);
if small == 0 || small * 4 < self.tuning.posting {
for (i, m) in members.iter().enumerate() {
self.place(p, m.id, m.tag, &xs[i * dim..(i + 1) * dim]);
}
self.postings[p].stuck = members.len() * 2;
return members.len();
}
self.centroids[p * dim..(p + 1) * dim].copy_from_slice(&a);
self.coarse.moved(p, &a, dim);
let q = self.add_partition(&b);
for (i, m) in members.iter().enumerate() {
let to = if sides[i] { p } else { q };
self.place(to, m.id, m.tag, &xs[i * dim..(i + 1) * dim]);
}
members.len() + self.sweep(&[p, q], vectors)
}
fn merge(&mut self, p: usize, vectors: &impl Vectors) -> usize {
let (members, xs) = self.take(p, vectors);
let dim = self.dim();
self.drop_partition(p);
let mut short = core::mem::take(&mut self.scratch);
for (i, m) in members.iter().enumerate() {
let x = &xs[i * dim..(i + 1) * dim];
let to = self.roughly_nearest(x, &mut short);
self.place(to, m.id, m.tag, x);
}
self.scratch = short;
members.len()
}
fn sweep(&mut self, changed: &[usize], vectors: &impl Vectors) -> usize {
let dim = self.dim();
let mut look: Vec<usize> = Vec::new();
for &p in changed {
let centre = self.centroid(p).to_vec();
for q in self.roughly_near_partitions(¢re, self.tuning.sweep) {
if !changed.contains(&q) && !look.contains(&q) {
look.push(q);
}
}
}
let fresh: Vec<(usize, Vec<f32>)> = changed
.iter()
.map(|&p| (p, self.centroid(p).to_vec()))
.collect();
let mut seen = 0;
let mut buf = vec![0.0f32; dim];
for p in look {
let here = self.centroid(p).to_vec();
for i in (0..self.postings[p].len()).rev() {
seen += 1;
let id = self.postings[p].ids[i];
let tag = self.postings[p].tags[i];
if !vectors.get(id, &mut buf) {
self.pull_and_forget(p, i);
continue;
}
let x = self.quant.rotate(&buf);
let mut best = (p, sqdist(&x, &here));
for (q, centre) in &fresh {
let d = sqdist(&x, centre);
if d < best.1 {
best = (*q, d);
}
}
if best.0 != p {
self.pull_and_forget(p, i);
self.place(best.0, id, tag, &x);
}
}
}
seen
}
fn take(&mut self, p: usize, vectors: &impl Vectors) -> (Vec<Member>, Vec<f32>) {
let dim = self.dim();
let ids = std::mem::take(&mut self.postings[p].ids);
let tags = std::mem::take(&mut self.postings[p].tags);
self.postings[p].codes.clear();
self.postings[p].meta.clear();
self.note(p);
let mut kept = Vec::with_capacity(ids.len());
let mut xs = Vec::with_capacity(ids.len() * dim);
let mut buf = vec![0.0f32; dim];
for (id, tag) in ids.into_iter().zip(tags) {
self.detach(id, p);
if vectors.get(id, &mut buf) {
xs.extend_from_slice(&self.quant.rotate(&buf));
kept.push(Member { id, tag });
}
}
(kept, xs)
}
fn roughly_near_partitions(&self, x: &[f32], n: usize) -> Vec<usize> {
if !self.coarse.ready() {
return self.near_partitions(x, n);
}
let mut short = Vec::new();
self.coarse.shortlist(x, self.dim(), &mut short);
let mut by: Vec<(usize, f32)> = short
.iter()
.map(|&p| (p as usize, sqdist(x, self.centroid(p as usize))))
.collect();
let n = n.min(by.len());
by.select_nth_unstable_by(n.saturating_sub(1), |a, b| a.1.total_cmp(&b.1));
by.truncate(n);
by.sort_by(|a, b| a.1.total_cmp(&b.1));
by.into_iter().map(|(p, _)| p).collect()
}
fn near_partitions(&self, x: &[f32], n: usize) -> Vec<usize> {
let mut by: Vec<(usize, f32)> = (0..self.postings.len())
.map(|p| (p, sqdist(x, self.centroid(p))))
.collect();
let n = n.min(by.len());
by.select_nth_unstable_by(n.saturating_sub(1), |a, b| a.1.total_cmp(&b.1));
by.truncate(n);
by.sort_by(|a, b| a.1.total_cmp(&b.1));
by.into_iter().map(|(p, _)| p).collect()
}
#[cfg(test)]
pub(crate) fn probe_order(&self, q: &[f32], into: &mut Vec<usize>) {
let u = self.quant.rotate(q);
*into = self.near_partitions(&u, self.postings.len());
}
#[cfg(test)]
pub(crate) fn holder(&self, id: u64) -> Option<usize> {
self.any_place(id).map(|s| s.partition as usize)
}
fn roughly_nearest(&self, x: &[f32], short: &mut Vec<u32>) -> usize {
if !self.coarse.ready() {
return self.nearest(x);
}
self.coarse.shortlist(x, self.dim(), short);
short
.iter()
.map(|&p| (p as usize, sqdist(x, self.centroid(p as usize))))
.min_by(|a, b| a.1.total_cmp(&b.1))
.map_or(0, |(p, _)| p)
}
fn nearest(&self, x: &[f32]) -> usize {
(0..self.postings.len())
.map(|p| (p, sqdist(x, self.centroid(p))))
.min_by(|a, b| a.1.total_cmp(&b.1))
.map_or(0, |(p, _)| p)
}
fn centroid(&self, p: usize) -> &[f32] {
let dim = self.dim();
&self.centroids[p * dim..(p + 1) * dim]
}
fn add_partition(&mut self, centroid: &[f32]) -> usize {
let dim = self.quant.dim();
self.centroids.extend_from_slice(centroid);
self.postings.push(Posting::default());
let p = self.postings.len() - 1;
self.coarse.added(p, centroid, dim);
self.note(p);
self.refresh_coarse();
p
}
fn refresh_coarse(&mut self) {
let n = self.postings.len();
if self.coarse.stale(n) {
let dim = self.quant.dim();
self.coarse.rebuild(&self.centroids, dim, n);
}
}
fn drop_partition(&mut self, p: usize) {
debug_assert_eq!(self.postings[p].len(), 0, "a partition is emptied first");
let dim = self.dim();
let last = self.postings.len() - 1;
self.coarse.dropped(p);
self.postings.swap_remove(p);
for i in 0..dim {
self.centroids[p * dim + i] = self.centroids[last * dim + i];
}
self.centroids.truncate(last * dim);
if p != last {
for i in 0..self.postings[p].len() {
let id = self.postings[p].ids[i];
if let Some(at) = self.placed_at(id, last) {
self.places[at as usize].partition = p as u32;
}
}
self.note(p);
}
self.refresh_coarse();
}
fn place(&mut self, p: usize, id: u64, tag: u64, x: &[f32]) {
let dim = self.dim();
let width = self.quant.code_bytes();
let slot = self.postings[p].len();
if !self.attach(id, p, slot) {
return;
}
self.postings[p].codes.resize((slot + 1) * width, 0);
let centroid = &self.centroids[p * dim..(p + 1) * dim];
let coded = self.quant.encode_rotated(
x,
centroid,
&mut self.postings[p].codes[slot * width..(slot + 1) * width],
);
self.postings[p].ids.push(id);
self.postings[p].tags.push(tag);
self.postings[p].meta.push(coded);
self.note(p);
}
fn attach(&mut self, id: u64, p: usize, slot: usize) -> bool {
let head = self.at.get(&id).copied().unwrap_or(END);
let mut walk = head;
while walk != END {
if self.places[walk as usize].partition as usize == p {
return false;
}
walk = self.places[walk as usize].next;
}
let place = Place {
partition: p as u32,
slot: slot as u32,
next: head,
};
let at = if self.free == END {
self.places.push(place);
(self.places.len() - 1) as u32
} else {
let at = self.free;
self.free = self.places[at as usize].next;
self.places[at as usize] = place;
at
};
self.at.insert(id, at);
true
}
fn detach(&mut self, id: u64, p: usize) -> bool {
let Some(&head) = self.at.get(&id) else {
return false;
};
let mut prev = END;
let mut walk = head;
while walk != END {
let this = self.places[walk as usize];
if this.partition as usize == p {
if prev == END {
if this.next == END {
self.at.remove(&id);
} else {
self.at.insert(id, this.next);
}
} else {
self.places[prev as usize].next = this.next;
}
self.places[walk as usize].next = self.free;
self.free = walk;
return true;
}
prev = walk;
walk = this.next;
}
false
}
fn detach_all(&mut self, id: u64) -> bool {
let Some(head) = self.at.remove(&id) else {
return false;
};
let mut walk = head;
while walk != END {
let next = self.places[walk as usize].next;
self.places[walk as usize].next = self.free;
self.free = walk;
walk = next;
}
true
}
fn placed_at(&self, id: u64, p: usize) -> Option<u32> {
let mut walk = self.at.get(&id).copied().unwrap_or(END);
while walk != END {
if self.places[walk as usize].partition as usize == p {
return Some(walk);
}
walk = self.places[walk as usize].next;
}
None
}
fn reslot(&mut self, id: u64, p: usize, s: usize) {
if let Some(at) = self.placed_at(id, p) {
self.places[at as usize].slot = s as u32;
} else {
debug_assert!(
false,
"id {id} is in partition {p} and the map does not say so"
);
}
}
#[cfg(test)]
fn placements_of(&self, id: u64) -> usize {
let mut walk = self.at.get(&id).copied().unwrap_or(END);
let mut n = 0;
while walk != END {
n += 1;
walk = self.places[walk as usize].next;
}
n
}
fn any_place(&self, id: u64) -> Option<Place> {
self.at.get(&id).map(|&at| self.places[at as usize])
}
fn every_place(&self, id: u64, into: &mut Vec<Place>) {
into.clear();
let mut walk = self.at.get(&id).copied().unwrap_or(END);
while walk != END {
let place = self.places[walk as usize];
into.push(place);
walk = place.next;
}
}
fn pull(&mut self, p: usize, s: usize) -> Option<u64> {
let width = self.quant.code_bytes();
let posting = &mut self.postings[p];
let last = posting.len() - 1;
posting.ids.swap_remove(s);
posting.tags.swap_remove(s);
posting.meta.swap_remove(s);
if s != last {
let (head, tail) = posting.codes.split_at_mut(last * width);
head[s * width..(s + 1) * width].copy_from_slice(&tail[..width]);
}
posting.codes.truncate(last * width);
let moved = (s != last).then(|| posting.ids[s]);
self.note(p);
moved
}
fn pull_and_forget(&mut self, p: usize, s: usize) {
let id = self.postings[p].ids[s];
self.detach(id, p);
if let Some(moved) = self.pull(p, s) {
self.reslot(moved, p, s);
}
}
}
#[derive(Clone, Copy)]
struct Member {
id: u64,
tag: u64,
}
#[derive(Debug, PartialEq, Eq)]
enum Job {
Split(usize),
Merge(usize),
}
fn two_means(xs: &[f32], dim: usize) -> (Vec<f32>, Vec<f32>) {
let n = xs.len() / dim;
let mut middle = vec![0.0f32; dim];
for i in 0..n {
for (m, c) in middle.iter_mut().zip(&xs[i * dim..(i + 1) * dim]) {
*m += c;
}
}
for m in &mut middle {
*m /= n as f32;
}
let far = |from: &[f32]| {
(0..n)
.max_by(|&i, &j| {
sqdist(from, &xs[i * dim..(i + 1) * dim])
.total_cmp(&sqdist(from, &xs[j * dim..(j + 1) * dim]))
})
.unwrap_or(0)
};
let i = far(&middle);
let mut a = xs[i * dim..(i + 1) * dim].to_vec();
let j = far(&a);
let mut b = xs[j * dim..(j + 1) * dim].to_vec();
for _ in 0..8 {
let mut sums = (vec![0.0f32; dim], vec![0.0f32; dim]);
let mut counts = (0usize, 0usize);
for i in 0..n {
let x = &xs[i * dim..(i + 1) * dim];
if sqdist(x, &a) <= sqdist(x, &b) {
for (s, c) in sums.0.iter_mut().zip(x) {
*s += c;
}
counts.0 += 1;
} else {
for (s, c) in sums.1.iter_mut().zip(x) {
*s += c;
}
counts.1 += 1;
}
}
if counts.0 > 0 {
for (m, s) in a.iter_mut().zip(&sums.0) {
*m = s / counts.0 as f32;
}
}
if counts.1 > 0 {
for (m, s) in b.iter_mut().zip(&sums.1) {
*m = s / counts.1 as f32;
}
}
}
(a, b)
}
#[derive(PartialEq)]
struct Ranked {
at: f32,
id: u64,
}
impl Eq for Ranked {}
impl Ord for Ranked {
fn cmp(&self, other: &Ranked) -> std::cmp::Ordering {
self.at.total_cmp(&other.at).then(self.id.cmp(&other.id))
}
}
impl PartialOrd for Ranked {
fn partial_cmp(&self, other: &Ranked) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
struct Bounded {
want: usize,
heap: std::collections::BinaryHeap<Ranked>,
}
impl Bounded {
fn new(want: usize) -> Bounded {
Bounded {
want,
heap: std::collections::BinaryHeap::with_capacity(want + 1),
}
}
fn full(&self) -> bool {
self.heap.len() >= self.want
}
#[inline]
fn wants(&self, at: f32) -> bool {
match self.heap.peek() {
Some(worst) if self.heap.len() >= self.want => at < worst.at,
_ => true,
}
}
fn put(&mut self, id: u64, at: f32) {
if self.heap.len() >= self.want {
self.heap.pop();
}
self.heap.push(Ranked { at, id });
}
fn sorted(self) -> Vec<(u64, f32)> {
self.heap
.into_sorted_vec()
.into_iter()
.map(|r| (r.id, r.at))
.collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
use yo_common::Rng;
fn slow_job(ix: &Partitions) -> Option<Job> {
let big = (0..ix.postings.len())
.filter(|&p| ix.postings[p].len() > ix.postings[p].stuck)
.max_by_key(|&p| ix.postings[p].len());
if let Some(big) = big
&& ix.postings[big].len() > ix.tuning.posting * 2
{
return Some(Job::Split(big));
}
if ix.postings.len() > 1 {
let small = (0..ix.postings.len()).min_by_key(|&p| ix.postings[p].len())?;
if ix.postings[small].len() * 4 < ix.tuning.posting {
return Some(Job::Merge(small));
}
}
None
}
#[test]
fn the_candidate_lists_answer_what_the_two_passes_answered() {
let (n, dim, posting) = shrunk(3000, 16, Tuning::default().posting);
let store = corpus(dim, n, 12, 0x105E);
let tuning = Tuning {
posting,
..Tuning::default()
};
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
let mut rng = Rng::new(0x105F);
let mut live: Vec<u64> = Vec::new();
for id in 0..n as u64 {
ix.insert(id, &store.0[id as usize]);
live.push(id);
if id % 7 == 3 && !live.is_empty() {
let at = rng.below(live.len());
let gone = live.swap_remove(at);
ix.remove(gone);
}
assert_eq!(
ix.job(),
slow_job(&ix),
"after {id} inserts, before maintaining"
);
ix.maintain(&store, 8);
assert_eq!(
ix.job(),
slow_job(&ix),
"after {id} inserts, after maintaining"
);
assert_eq!(ix.needs_maintenance(), slow_job(&ix).is_some());
}
assert!(ix.postings.len() > 5, "the test never split anything");
}
struct Store(Vec<Vec<f32>>);
impl Vectors for Store {
fn get(&self, id: u64, into: &mut [f32]) -> bool {
match self.0.get(id as usize) {
Some(v) => {
into.copy_from_slice(v);
true
}
None => false,
}
}
}
struct Holey(Vec<Vec<f32>>, u64);
impl Vectors for Holey {
fn get(&self, id: u64, into: &mut [f32]) -> bool {
if id == self.1 {
return false;
}
match self.0.get(id as usize) {
Some(v) => {
into.copy_from_slice(v);
true
}
None => false,
}
}
}
fn corpus(dim: usize, n: usize, clusters: usize, seed: u64) -> Store {
let mut rng = Rng::new(seed);
let centres: Vec<Vec<f32>> = (0..clusters).map(|_| draw(dim, &mut rng)).collect();
Store(
(0..n)
.map(|i| {
let off = draw(dim, &mut rng);
let mut v: Vec<f32> = centres[i % clusters]
.iter()
.zip(&off)
.map(|(c, o)| c + o * 0.7)
.collect();
unit(&mut v);
v
})
.collect(),
)
}
fn shrunk(n: usize, dim: usize, posting: usize) -> (usize, usize, usize) {
if cfg!(miri) {
((n / 20).max(120), dim.min(8), (posting / 20).max(6))
} else {
(n, dim, posting)
}
}
fn draw(dim: usize, rng: &mut Rng) -> Vec<f32> {
let mut v: Vec<f32> = (0..dim)
.map(|i| {
let u = (rng.next_u64() >> 40) as f32 / (1u32 << 24) as f32;
let heavy = if i < (dim / 16).max(1) { 6.0 } else { 1.0 };
(u * 2.0 - 1.0) * heavy
})
.collect();
unit(&mut v);
v
}
fn unit(v: &mut [f32]) {
let len = v.iter().map(|c| c * c).sum::<f32>().sqrt();
for c in v {
*c /= len;
}
}
fn truth(store: &Store, q: &[f32], k: usize) -> Vec<u64> {
let mut all: Vec<(u64, f32)> = store
.0
.iter()
.enumerate()
.map(|(i, v)| (i as u64, sqdist(q, v)))
.collect();
all.sort_by(|a, b| a.1.total_cmp(&b.1));
all.truncate(k);
all.into_iter().map(|(i, _)| i).collect()
}
fn build(store: &Store, dim: usize, tuning: Tuning) -> Partitions {
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
for (i, v) in store.0.iter().enumerate() {
ix.insert(i as u64, v);
if i % 64 == 0 {
ix.maintain(store, 4096);
}
}
ix.maintain(store, 1 << 20);
ix
}
fn recall(ix: &Partitions, store: &Store, k: usize, queries: usize) -> f32 {
let mut hits = 0usize;
for i in 0..queries {
let q = &store.0[i * 7 % store.0.len()];
let want = truth(store, q, k);
let got: Vec<u64> = ix.search(q, k, store).into_iter().map(|h| h.id).collect();
hits += want.iter().filter(|id| got.contains(id)).count();
}
hits as f32 / (queries * k) as f32
}
fn consistent(ix: &Partitions) {
assert_eq!(ix.centroids.len(), ix.postings.len() * ix.dim());
let width = ix.quant.code_bytes();
let mut seen = 0usize;
for (p, posting) in ix.postings.iter().enumerate() {
assert_eq!(posting.codes.len(), posting.len() * width, "partition {p}");
assert_eq!(posting.meta.len(), posting.len(), "partition {p}");
assert_eq!(posting.tags.len(), posting.len(), "partition {p}");
let mut here = HashSet::new();
for (s, id) in posting.ids.iter().enumerate() {
assert!(here.insert(*id), "id {id} is twice in partition {p}");
let at = ix
.placed_at(*id, p)
.expect("every member is in the map, under the partition holding it");
assert_eq!(ix.places[at as usize].slot as usize, s, "id {id}");
seen += 1;
}
}
let mut held = 0usize;
for (&id, &head) in &ix.at {
let mut walk = head;
let mut mine = HashSet::new();
while walk != END {
let place = ix.places[walk as usize];
let p = place.partition as usize;
assert!(mine.insert(p), "id {id} is filed twice under partition {p}");
assert!(
p < ix.postings.len(),
"id {id} is filed under partition {p}"
);
assert_eq!(
ix.postings[p].ids[place.slot as usize], id,
"id {id} is filed at a slot holding something else"
);
held += 1;
walk = place.next;
}
}
assert_eq!(seen, held, "the map and the postings disagree on the count");
let mut spare = 0usize;
let mut walk = ix.free;
while walk != END {
spare += 1;
assert!(spare <= ix.places.len(), "the free list has a cycle in it");
walk = ix.places[walk as usize].next;
}
assert_eq!(held + spare, ix.places.len(), "the arena has leaked");
}
fn build_tagged(
store: &Store,
dim: usize,
tuning: Tuning,
tag: impl Fn(u64) -> u64,
) -> Partitions {
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
for (i, v) in store.0.iter().enumerate() {
ix.insert_tagged(i as u64, v, tag(i as u64));
if i % 64 == 0 {
ix.maintain(store, 4096);
}
}
ix.maintain(store, 1 << 20);
ix
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: one document in fifty of three thousand, and the whole point is what the near misses were, which needs a corpus with near misses in it"
)]
fn a_filter_in_the_scan_finds_what_a_filter_after_it_cannot() {
let dim = 96;
let store = corpus(dim, 3000, 12, 47);
let tuning = Tuning {
posting: 64,
..Tuning::default()
};
let wanted = |id: u64| id.is_multiple_of(50);
let ix = build_tagged(&store, dim, tuning, |id| u64::from(wanted(id)));
let (mut pushed, mut after) = (0usize, 0usize);
let k = 10;
for i in 0..40 {
let q = &store.0[i * 71 % store.0.len()];
let mut all: Vec<(u64, f32)> = store
.0
.iter()
.enumerate()
.filter(|(id, _)| wanted(*id as u64))
.map(|(id, v)| (id as u64, sqdist(q, v)))
.collect();
all.sort_by(|a, b| a.1.total_cmp(&b.1));
let want: Vec<u64> = all[..k].iter().map(|(id, _)| *id).collect();
let got: Vec<u64> = ix
.search_where(q, k, &|tag: u64| tag == 1, &store)
.into_iter()
.map(|h| h.id)
.collect();
pushed += want.iter().filter(|id| got.contains(id)).count();
let late: Vec<u64> = ix
.search(q, k * tuning.rerank, &store)
.into_iter()
.map(|h| h.id)
.filter(|id| wanted(*id))
.take(k)
.collect();
after += want.iter().filter(|id| late.contains(id)).count();
}
let (pushed, after) = (pushed as f32 / 400.0, after as f32 / 400.0);
assert!(pushed >= 0.95, "pushing the filter down gave {pushed}");
assert!(
after < pushed / 2.0,
"filtering afterwards gave {after} against {pushed}, which is not the point being made"
);
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: recall at ten where the tag lets one in ten through and the exact test keeps one in fifty"
)]
fn the_exact_test_decides_and_the_scan_widens_for_it() {
struct Summary;
impl Filter for Summary {
fn allows(&self, tag: u64) -> bool {
tag == 1
}
fn exact(&self, id: u64) -> bool {
id.is_multiple_of(50)
}
}
let dim = 64;
let store = corpus(dim, 3000, 9, 71);
let tuning = Tuning {
posting: 64,
..Tuning::default()
};
let ix = build_tagged(&store, dim, tuning, |id| u64::from(id.is_multiple_of(10)));
let k = 10;
let mut found = 0usize;
for i in 0..20 {
let q = &store.0[i * 131 % store.0.len()];
let mut all: Vec<(u64, f32)> = store
.0
.iter()
.enumerate()
.map(|(id, v)| (id as u64, sqdist(q, v)))
.filter(|(id, _)| id.is_multiple_of(50))
.collect();
all.sort_by(|a, b| a.1.total_cmp(&b.1));
let want: Vec<u64> = all[..k].iter().map(|(id, _)| *id).collect();
let got: Vec<u64> = ix
.search_where(q, k, &Summary, &store)
.into_iter()
.map(|h| h.id)
.collect();
assert!(
got.iter().all(|id| id.is_multiple_of(50)),
"the exact test did not decide: {got:?}"
);
found += want.iter().filter(|id| got.contains(id)).count();
}
let recall = found as f32 / (20.0 * k as f32);
assert!(recall >= 0.9, "two stage filtering gave {recall}");
}
#[test]
fn a_filter_that_matches_nothing_answers_nothing() {
let (n, dim, posting) = shrunk(500, 64, Tuning::default().posting);
let store = corpus(dim, n, 4, 53);
let tuning = Tuning {
posting,
..Tuning::default()
};
let ix = build_tagged(&store, dim, tuning, |_| 1);
assert!(
ix.search_where(&store.0[0], 10, &|tag: u64| tag == 2, &store)
.is_empty()
);
let all = ix.search_where(&store.0[0], 10, &Any, &store);
assert_eq!(all, ix.search(&store.0[0], 10, &store));
}
#[test]
fn a_tag_survives_a_split_and_a_merge() {
let (n, dim, posting) = shrunk(800, 64, 24);
let store = corpus(dim, n, 6, 59);
let tuning = Tuning {
posting,
..Tuning::default()
};
let mut ix = build_tagged(&store, dim, tuning, |id| id * 7 + 1);
assert!(ix.partitions() > 4, "it never split");
for id in 0..n as u64 {
assert_eq!(ix.tag(id), Some(id * 7 + 1), "id {id} after the splits");
}
let left = n as u64 - 40;
for id in 0..left {
ix.remove(id);
}
ix.maintain(&store, 1 << 20);
consistent(&ix);
for id in left..n as u64 {
assert_eq!(ix.tag(id), Some(id * 7 + 1), "id {id} after the merges");
}
assert_eq!(ix.tag(0), None);
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: one member in a hundred of two thousand is twenty answers, and the assertion is that exact number"
)]
fn a_selective_filter_makes_the_search_look_further() {
let dim = 64;
let store = corpus(dim, 2000, 10, 61);
let tuning = Tuning {
posting: 32,
..Tuning::default()
};
let tag = |id: u64| u64::from(id.is_multiple_of(100));
let ix = build_tagged(&store, dim, tuning, tag);
let narrow = build_tagged(&store, dim, Tuning { widen: 1, ..tuning }, tag);
let mut wide_found = 0usize;
let mut narrow_found = 0usize;
for i in 0..20 {
let q = &store.0[i * 91 % store.0.len()];
wide_found += ix.search_where(q, 10, &|t: u64| t == 1, &store).len();
narrow_found += narrow.search_where(q, 10, &|t: u64| t == 1, &store).len();
}
assert_eq!(
wide_found, 200,
"one in a hundred of two thousand is twenty"
);
assert!(
narrow_found < wide_found,
"not widening found {narrow_found} of {wide_found}"
);
}
#[test]
fn a_signature_never_rejects_something_it_should_have_matched() {
let english = Signature::of(&[("lang", b"en")]);
let doc = Signature::of(&[("lang", b"en"), ("topic", b"finance"), ("year", b"2026")]);
assert!(doc.covers(english));
assert!(english.allows(doc.bits()));
assert_eq!(Signature::from_bits(doc.bits()), doc);
for i in 0..500u32 {
let value = i.to_string();
let one = Signature::of(&[("id", value.as_bytes())]);
let with = Signature::of(&[("id", value.as_bytes()), ("kind", b"page")]);
assert!(with.covers(one), "value {value}");
}
}
#[test]
fn an_empty_index_answers_nothing() {
let ix = Partitions::new(32, Bits::One, 1, Tuning::default());
let store = Store(Vec::new());
assert!(ix.is_empty());
assert_eq!(ix.partitions(), 0);
assert!(ix.search(&[0.0; 32], 10, &store).is_empty());
assert!(!ix.needs_maintenance());
}
#[test]
fn the_first_vector_is_the_first_partition() {
let store = corpus(32, 1, 1, 3);
let mut ix = Partitions::new(32, Bits::One, 1, Tuning::default());
ix.insert(0, &store.0[0]);
assert_eq!(ix.partitions(), 1);
assert_eq!(ix.len(), 1);
let hits = ix.search(&store.0[0], 5, &store);
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].id, 0);
assert!(hits[0].distance < 1e-6, "{}", hits[0].distance);
consistent(&ix);
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: recall at ten over two thousand vectors, and a recall figure over a corpus small enough to interpret is a number about nothing"
)]
fn a_search_finds_what_brute_force_finds() {
let dim = 128;
let store = corpus(dim, 2000, 12, 5);
let ix = build(&store, dim, Tuning::default());
assert!(ix.partitions() > 1, "it never split");
consistent(&ix);
let r = recall(&ix, &store, 10, 50);
assert!(r >= 0.95, "recall at 10 was {r}");
}
#[test]
fn a_posting_that_grows_too_big_splits() {
let (n, dim, posting) = shrunk(600, 64, 32);
let tuning = Tuning {
posting,
..Tuning::default()
};
let store = corpus(dim, n, 6, 9);
let ix = build(&store, dim, tuning);
assert!(
ix.partitions() >= n / (posting * 2),
"{n} vectors in {} partitions",
ix.partitions()
);
for held in &ix.postings {
assert!(
held.len() <= posting * 2 || held.len() <= held.stuck,
"a posting is {} long and did not give up splitting",
held.len()
);
}
consistent(&ix);
}
#[test]
fn a_posting_that_shrinks_merges() {
let (n, dim, posting) = shrunk(600, 64, 32);
let tuning = Tuning {
posting,
..Tuning::default()
};
let store = corpus(dim, n, 6, 9);
let mut ix = build(&store, dim, tuning);
let grown = ix.partitions();
assert!(grown > 4);
let left = n as u64 - 30;
for id in 0..left {
assert!(ix.remove(id));
}
ix.maintain(&store, 1 << 20);
consistent(&ix);
assert_eq!(ix.len(), 30);
assert!(
ix.partitions() < grown,
"{} partitions for 30 vectors, was {grown}",
ix.partitions()
);
let last = n as u64 - 1;
let hits = ix.search(&store.0[last as usize], 1, &store);
assert_eq!(hits[0].id, last);
}
#[test]
fn maintenance_runs_out_of_things_to_do() {
for posting in [6, 8, 12, 16, 24] {
let store = corpus(16, 160, 4, 11);
let tuning = Tuning {
posting,
..Tuning::default()
};
let mut ix = build(&store, 16, tuning);
let left = ix.maintain(&store, 1 << 20);
assert_eq!(left, 0, "a posting of {posting} never settles");
consistent(&ix);
assert_eq!(ix.len(), 160, "settling lost something");
}
}
#[test]
fn a_removed_vector_stops_coming_back() {
let (n, dim, posting) = shrunk(400, 64, Tuning::default().posting);
let store = corpus(dim, n, 4, 11);
let mut ix = build(
&store,
dim,
Tuning {
posting,
..Tuning::default()
},
);
let q = store.0[7].clone();
assert_eq!(ix.search(&q, 1, &store)[0].id, 7);
assert!(ix.remove(7));
assert!(!ix.remove(7), "removing it twice should say so");
assert!(!ix.contains(7));
assert_eq!(ix.len(), n - 1);
consistent(&ix);
assert!(ix.search(&q, 5, &store).iter().all(|h| h.id != 7));
}
fn copies(ix: &Partitions) -> f32 {
let held: usize = ix.postings.iter().map(Posting::len).sum();
held as f32 / ix.len() as f32
}
#[test]
fn spilling_puts_boundary_vectors_in_more_than_one_partition() {
let (n, dim, posting) = shrunk(3000, 32, Tuning::default().posting);
let store = corpus(dim, n, 12, 5);
let base = Tuning {
posting,
..Tuning::default()
};
let off = Tuning { spill: 1, ..base };
let none = build(&store, dim, off);
consistent(&none);
assert_eq!(copies(&none), 1.0, "spill of one is one copy of everything");
let on = build(&store, dim, base);
consistent(&on);
let rate = copies(&on);
assert!(rate > 1.0, "spilling should make copies, made {rate}");
assert!(
rate < Tuning::default().spill as f32,
"slack should stop short of copying everything into everything, made {rate}"
);
assert_eq!(on.len(), store.0.len(), "a copy is not a member");
}
#[test]
fn a_copy_is_found_from_the_partition_it_was_copied_into() {
let (n, dim, posting) = shrunk(3000, 32, Tuning::default().posting);
let store = corpus(dim, n, 12, 5);
let t = Tuning {
posting,
slack: 0.25,
..Tuning::default()
};
let mut ix = build(&store, dim, t);
let (id, copies) = (0..n as u64)
.filter_map(|id| {
let mut places = Vec::new();
ix.every_place(id, &mut places);
(places.len() > 1).then_some((id, places))
})
.next()
.expect("some member near a boundary got copied");
let mut narrow = t;
narrow.probe = 1;
narrow.widen = 1;
ix.retune(narrow);
for place in &copies {
let p = place.partition as usize;
let neighbour = ix.postings[p]
.ids
.iter()
.copied()
.find(|&other| other != id)
.expect("the partition holds more than the copy");
let got = ix.candidates(&store.0[neighbour as usize], ix.postings[p].len());
assert!(
got.iter().any(|&(seen, _)| seen == id),
"member {id} has a copy in partition {p} and a search of it did not find it"
);
}
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: a saving measured as partitions read per query over a hundred queries, against a recall that has to survive it"
)]
fn patience_reads_fewer_partitions_and_keeps_the_answers() {
let dim = 32;
let store = corpus(dim, 4000, 16, 77);
let wide = Tuning {
probe: 64,
..Tuning::default()
};
let mut ix = build(&store, dim, wide);
let queries = 100;
let full = recall(&ix, &store, 10, queries);
let cost = |ix: &Partitions| -> f64 {
(0..queries)
.map(|i| {
let q = &store.0[i * 7 % store.0.len()];
ix.search_costed(q, 10, &Any, &store).1.probed
})
.sum::<usize>() as f64
/ queries as f64
};
let spent = cost(&ix);
ix.retune(Tuning {
probe: 64,
patience: 2,
..Tuning::default()
});
let cut = cost(&ix);
assert!(
cut < spent * 0.75,
"patience of two read {cut:.1} partitions a query against {spent:.1}, which is not a saving worth the knob"
);
let after = recall(&ix, &store, 10, queries);
assert!(
after >= full - 0.02,
"recall went from {full} to {after}, which is more than giving up early is allowed to cost"
);
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: ten answers at one member in fifty, spread over enough partitions that the first few cannot hold them"
)]
fn patience_does_not_cut_off_a_filter_that_is_still_short() {
let dim = 32;
let store = corpus(dim, 4000, 16, 91);
let mut ix = build(
&store,
dim,
Tuning {
patience: 1,
..Tuning::default()
},
);
for id in 0..4000u64 {
ix.retag(id, u64::from(id % 50 == 0));
}
struct Rare;
impl Filter for Rare {
fn allows(&self, tag: u64) -> bool {
tag == 1
}
}
let q = &store.0[3];
let got = ix.search_where(q, 10, &Rare, &store);
assert_eq!(got.len(), 10, "the filtered search came back short");
for hit in &got {
assert!(hit.id.is_multiple_of(50), "{} is not a match", hit.id);
}
}
#[test]
fn a_replicated_member_comes_back_once() {
let (n, dim, posting) = shrunk(2000, 32, Tuning::default().posting);
let store = corpus(dim, n, 8, 31);
let t = Tuning {
posting,
probe: 1 << 20,
..Tuning::default()
};
let ix = build(&store, dim, t);
for i in 0..50 {
let q = &store.0[i * 37 % store.0.len()];
let got: Vec<u64> = ix.search(q, 20, &store).into_iter().map(|h| h.id).collect();
let mut once = got.clone();
once.sort_unstable();
once.dedup();
assert_eq!(got.len(), once.len(), "a duplicate answer for query {i}");
}
}
#[test]
fn removing_a_replicated_member_takes_every_copy() {
let (n, dim, posting) = shrunk(1500, 32, Tuning::default().posting);
let store = corpus(dim, n, 6, 41);
let mut ix = build(
&store,
dim,
Tuning {
posting,
..Tuning::default()
},
);
let before: usize = ix.postings.iter().map(Posting::len).sum();
let mut gone = 0usize;
for id in (0..n as u64).step_by(3) {
gone += ix.placements_of(id);
assert!(ix.remove(id));
assert!(!ix.contains(id));
}
consistent(&ix);
let after: usize = ix.postings.iter().map(Posting::len).sum();
assert_eq!(before - after, gone, "a copy was left behind");
assert_eq!(ix.len(), n - n.div_ceil(3));
for id in (0..n as u64).step_by(3) {
let q = &store.0[id as usize];
assert!(ix.search(q, 5, &store).iter().all(|h| h.id != id));
}
}
#[test]
fn retagging_a_replicated_member_reaches_every_copy() {
let (n, dim, posting) = shrunk(1200, 32, Tuning::default().posting);
let store = corpus(dim, n, 6, 47);
let tuning = Tuning {
posting,
..Tuning::default()
};
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
for (i, v) in store.0.iter().enumerate() {
ix.insert_tagged(i as u64, v, 1);
if i % 64 == 0 {
ix.maintain(&store, 4096);
}
}
ix.maintain(&store, 1 << 20);
let mut pair = (0, 1, f32::INFINITY);
for a in 0..ix.partitions() {
for b in a + 1..ix.partitions() {
let d = sqdist(ix.centroid(a), ix.centroid(b));
if d < pair.2 {
pair = (a, b, d);
}
}
}
let (a, b, _) = pair;
let mid: Vec<f32> = ix
.centroid(a)
.iter()
.zip(ix.centroid(b))
.map(|(x, y)| (x + y) / 2.0)
.collect();
let id = n as u64;
ix.insert_tagged(id, &mid, 1);
assert!(
ix.placements_of(id) > 1,
"a member equidistant from the two nearest centroids was not copied"
);
assert!(ix.retag(id, 9));
let mut copies = Vec::new();
ix.every_place(id, &mut copies);
for place in &copies {
assert_eq!(
ix.postings[place.partition as usize].tags[place.slot as usize], 9,
"a copy kept the old tag"
);
}
consistent(&ix);
}
#[test]
fn inserting_the_same_id_twice_replaces_it() {
let (n, dim, posting) = shrunk(200, 64, Tuning::default().posting);
let store = corpus(dim, n, 2, 13);
let mut ix = build(
&store,
dim,
Tuning {
posting,
..Tuning::default()
},
);
let before = ix.len();
ix.insert(3, &store.0[3]);
assert_eq!(ix.len(), before);
consistent(&ix);
assert_eq!(ix.search(&store.0[3], 1, &store)[0].id, 3);
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: a thousand identical vectors is what makes maintenance try the same split over and over, and the failure it looks for is a hang"
)]
fn a_thousand_copies_of_one_vector_do_not_spin() {
let dim = 32;
let one = corpus(dim, 1, 1, 41).0.pop().expect("one vector");
let store = Store(vec![one; 1000]);
let tuning = Tuning {
posting: 16,
..Tuning::default()
};
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
for (i, v) in store.0.iter().enumerate() {
ix.insert(i as u64, v);
ix.maintain(&store, 4096);
}
ix.maintain(&store, 1 << 20);
consistent(&ix);
assert_eq!(ix.len(), 1000);
assert!(!ix.needs_maintenance(), "it still thinks there is work");
let hits = ix.search(&store.0[0], 5, &store);
assert_eq!(hits.len(), 5);
assert!(hits.iter().all(|h| h.distance < 1e-6));
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: recall at the end of three thousand writes with a tenth of them churned, and a short stream is a fresh build, which is the measurement this one exists to avoid"
)]
fn recall_holds_over_a_write_stream_with_no_rebuild() {
let dim = 96;
let store = corpus(dim, 3000, 15, 17);
let tuning = Tuning {
posting: 64,
..Tuning::default()
};
let mut ix = Partitions::new(dim, Bits::One, 7, tuning);
let mut rng = Rng::new(23);
for (i, v) in store.0.iter().enumerate() {
ix.insert(i as u64, v);
if i > 100 && i % 10 == 0 {
let victim = rng.below(i) as u64;
ix.remove(victim);
ix.insert(victim, &store.0[victim as usize]);
}
ix.maintain(&store, 512);
}
ix.maintain(&store, 1 << 20);
consistent(&ix);
assert_eq!(ix.len(), store.0.len());
let r = recall(&ix, &store, 10, 60);
assert!(r >= 0.95, "recall at 10 after the stream was {r}");
}
#[test]
#[cfg_attr(
miri,
ignore = "the count is the claim: drifted members with the sweep against drifted members without it, and the assertion is the ratio between the two"
)]
fn the_sweep_is_what_keeps_members_under_their_nearest_centroid() {
let dim = 96;
let store = corpus(dim, 2000, 10, 29);
let tuning = Tuning {
posting: 48,
..Tuning::default()
};
let with = misfiled(&build(&store, dim, tuning), &store);
let without = misfiled(&build(&store, dim, Tuning { sweep: 0, ..tuning }), &store);
assert!(
with * 4 < without,
"sweeping left {with} members drifted and not sweeping left {without}"
);
}
fn misfiled(ix: &Partitions, store: &Store) -> usize {
let mut buf = vec![0.0f32; ix.dim()];
let mut wrong = 0;
for id in 0..store.0.len() as u64 {
if !ix.contains(id) {
continue;
}
assert!(store.get(id, &mut buf));
let near = ix.nearest(&ix.quant.rotate(&buf));
if ix.placed_at(id, near).is_none() {
wrong += 1;
}
}
wrong
}
#[test]
fn a_vector_the_log_forgot_is_dropped_rather_than_returned() {
let (n, dim, posting) = shrunk(400, 64, 24);
let store = corpus(dim, n, 4, 31);
let tuning = Tuning {
posting,
..Tuning::default()
};
let mut ix = build(&store, dim, tuning);
assert!(ix.contains(11));
let holey = Holey(store.0.clone(), 11);
assert!(
ix.search(&store.0[11], 5, &holey)
.iter()
.all(|h| h.id != 11)
);
for id in 0..n as u64 - 100 {
ix.remove(id);
}
ix.maintain(&holey, 1 << 20);
consistent(&ix);
assert!(!ix.contains(11));
}
#[test]
fn rotating_first_is_the_same_as_rotating_inside() {
let dim = 128;
let q = Quantizer::new(dim, Bits::One, 5);
let store = corpus(dim, 2, 1, 37);
let (v, c) = (&store.0[0], &store.0[1]);
let mut a = vec![0u8; q.code_bytes()];
let one = q.encode(v, c, &mut a);
let mut b = vec![0u8; q.code_bytes()];
let two = q.encode_rotated(&q.rotate(v), &q.rotate(c), &mut b);
assert_eq!(a, b, "the two ways round should write the same code");
assert!((one.norm - two.norm).abs() < 1e-4);
assert!((one.scale - two.scale).abs() < 1e-4);
}
#[test]
fn two_means_splits_two_clouds_apart() {
let dim = 8;
let mut xs = Vec::new();
for i in 0..40 {
let far = if i % 2 == 0 { 0.0 } else { 10.0 };
for d in 0..dim {
xs.push(far + (i as f32 + d as f32) * 0.01);
}
}
let (a, b) = two_means(&xs, dim);
let (near, away) = if a[0] < b[0] { (a, b) } else { (b, a) };
assert!(near[0] < 1.0, "{near:?}");
assert!(away[0] > 9.0, "{away:?}");
}
}