use std::{cmp::Ordering, collections::BinaryHeap, sync::Arc};
use crate::core::hash::FastSet;
use crate::core::sync::{Mutex, MutexGuard, RwLock};
use anyhow::{bail, ensure, Context, Result};
use serde::{Deserialize, Serialize};
use super::index::{text_of, text_scrubbed};
use super::{DocumentResult, DocumentStore, Embedding, StoredDocument, TextSegment};
use crate::core::qgraph::{BuildParams, Graph, VGraph, VSearchParams};
pub const SNAPSHOT_MIN_ROWS: usize = 1;
pub const DEFAULT_GRAPH_MIN_ROWS: usize = 4096;
pub const DEFAULT_EF: usize = 128;
pub const MAX_EF: usize = 4096;
pub const MAX_K: usize = 1000;
pub const REBUILD_AFTER_MUTATIONS: u64 = 64;
const MAX_ROWS: usize = 1 << 22;
const MAX_MATRIX_VALUES: usize = 512 * 1024 * 1024;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum VectorIndexKind {
Exact,
Graph,
Scan,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum VectorIndexMode {
#[default]
Auto,
Exact,
Graph,
}
impl VectorIndexMode {
pub fn parse(name: &str) -> Option<Self> {
match name {
"auto" => Some(Self::Auto),
"exact" => Some(Self::Exact),
"graph" => Some(Self::Graph),
_ => None,
}
}
pub fn as_str(self) -> &'static str {
match self {
Self::Auto => "auto",
Self::Exact => "exact",
Self::Graph => "graph",
}
}
}
#[derive(Clone, Debug)]
pub struct VectorQueryOptions {
pub mode: VectorIndexMode,
pub ef: usize,
pub min_score: f64,
}
impl Default for VectorQueryOptions {
fn default() -> Self {
Self {
mode: VectorIndexMode::Auto,
ef: DEFAULT_EF,
min_score: -1.0,
}
}
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct VectorIndexStatus {
pub rows: usize,
pub generation: u64,
pub current: bool,
pub stale_mutations: u64,
pub graph: bool,
pub rebuild_recommended: bool,
}
#[derive(Debug, Clone, Copy)]
pub struct GraphUnavailable;
impl std::fmt::Display for GraphUnavailable {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("no graph index is built; run index maintenance")
}
}
impl std::error::Error for GraphUnavailable {}
pub trait VectorIndex: Send + Sync {
fn kind(&self) -> VectorIndexKind;
fn rows(&self) -> usize;
fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>>;
}
#[derive(Clone, Copy)]
struct Ranked {
score: f32,
row: u32,
}
impl PartialEq for Ranked {
fn eq(&self, other: &Self) -> bool {
self.cmp(other) == Ordering::Equal
}
}
impl Eq for Ranked {}
impl PartialOrd for Ranked {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Ord for Ranked {
fn cmp(&self, other: &Self) -> Ordering {
other
.score
.total_cmp(&self.score)
.then(self.row.cmp(&other.row))
}
}
struct TopK {
heap: BinaryHeap<Ranked>,
k: usize,
}
impl TopK {
fn new(k: usize) -> Self {
Self {
heap: BinaryHeap::with_capacity(k + 1),
k,
}
}
fn offer(&mut self, row: u32, score: f32) {
let candidate = Ranked { score, row };
if self.k == 0 {
return;
}
if self.heap.len() < self.k {
self.heap.push(candidate);
} else if let Some(mut worst) = self.heap.peek_mut() {
if candidate < *worst {
*worst = candidate;
}
}
}
fn into_sorted(self) -> Vec<(u32, f32)> {
self.heap
.into_sorted_vec()
.into_iter()
.map(|ranked| (ranked.row, ranked.score))
.collect()
}
}
#[inline]
fn dot(a: &[f32], b: &[f32]) -> f32 {
debug_assert_eq!(a.len(), b.len());
let mut lanes = [0.0f32; 8];
let (wide_a, tail_a) = a.as_chunks::<8>();
let (wide_b, tail_b) = b.as_chunks::<8>();
for (x, y) in wide_a.iter().zip(wide_b) {
for ((lane, x), y) in lanes.iter_mut().zip(x).zip(y) {
*lane += x * y;
}
}
let mut sum = lanes.iter().sum::<f32>();
for (x, y) in tail_a.iter().zip(tail_b) {
sum += x * y;
}
sum
}
pub(crate) struct ExactScan {
dim: usize,
rows: Arc<[Embedding]>,
}
impl ExactScan {
fn row(&self, row: u32) -> &[f32] {
&self.rows[row as usize]
}
}
impl VectorIndex for ExactScan {
fn kind(&self) -> VectorIndexKind {
VectorIndexKind::Exact
}
fn rows(&self) -> usize {
self.rows.len()
}
fn search(&self, query: &[f32], k: usize, _ef: usize) -> Result<Vec<(u32, f32)>> {
ensure!(
query.len() == self.dim,
"query has {} dimensions, the index {}",
query.len(),
self.dim
);
let mut best = TopK::new(k.min(self.rows()));
for (row, values) in self.rows.iter().enumerate() {
best.offer(row as u32, dot(query, values));
}
Ok(best.into_sorted())
}
}
pub(crate) struct GraphIndex {
graph: VGraph,
exact: ExactScan,
}
impl VectorIndex for GraphIndex {
fn kind(&self) -> VectorIndexKind {
VectorIndexKind::Graph
}
fn rows(&self) -> usize {
self.exact.rows()
}
fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>> {
ensure!(
query.len() == self.exact.dim,
"query has {} dimensions, the index {}",
query.len(),
self.exact.dim
);
let k = k.min(self.rows());
if k == 0 {
return Ok(Vec::new());
}
let ef = ef.clamp(k, MAX_EF.max(k));
let params = VSearchParams {
ef,
rerank: ef,
..VSearchParams::default()
};
let mut ids = Vec::with_capacity(k);
self.graph
.searcher()
.search(query, k, params, &mut ids)
.map_err(|e| anyhow::anyhow!("graph search: {e}"))?;
let mut hits: Vec<(u32, f32)> = ids
.into_iter()
.filter(|&id| (id as usize) < self.rows())
.map(|id| (id, dot(query, self.exact.row(id))))
.collect();
hits.sort_by(|a, b| b.1.total_cmp(&a.1).then(a.0.cmp(&b.0)));
hits.dedup_by(|a, b| a.0 == b.0);
hits.truncate(k);
Ok(hits)
}
}
struct RowRef {
document: Arc<StoredDocument>,
chunk: u32,
}
pub(crate) struct VectorSnapshot {
generation: u64,
dim: usize,
rows: Vec<RowRef>,
members: FastSet<usize>,
exact: ExactScan,
graph: Option<GraphIndex>,
}
impl VectorSnapshot {
pub(crate) fn rows(&self) -> usize {
self.rows.len()
}
pub(crate) fn has_graph(&self) -> bool {
self.graph.is_some()
}
fn index(&self, mode: VectorIndexMode) -> Result<&dyn VectorIndex> {
match mode {
VectorIndexMode::Exact => Ok(&self.exact),
VectorIndexMode::Graph => self
.graph
.as_ref()
.map(|graph| graph as &dyn VectorIndex)
.ok_or_else(|| anyhow::Error::new(GraphUnavailable)),
VectorIndexMode::Auto => Ok(self
.graph
.as_ref()
.map_or(&self.exact as &dyn VectorIndex, |graph| {
graph as &dyn VectorIndex
})),
}
}
}
pub(crate) struct VectorCapture {
generation: u64,
dim: usize,
rows: Vec<RowRef>,
embeddings: Vec<Embedding>,
}
pub(crate) fn capture_vector_rows(
store: &DocumentStore,
dim: usize,
) -> Result<Option<VectorCapture>> {
let count = store.embedded_count();
if dim == 0 || count < SNAPSHOT_MIN_ROWS {
return Ok(None);
}
ensure!(
count <= MAX_ROWS,
"vector snapshot exceeds {MAX_ROWS} embedded chunks"
);
ensure!(
count
.checked_mul(dim)
.is_some_and(|values| values <= MAX_MATRIX_VALUES),
"vector snapshot exceeds {MAX_MATRIX_VALUES} embedding values"
);
let mut rows = Vec::new();
rows.try_reserve_exact(count)
.context("cannot allocate vector snapshot rows")?;
let mut embeddings = Vec::new();
embeddings
.try_reserve_exact(count)
.context("cannot allocate vector snapshot embedding handles")?;
for id in &store.embedded_documents {
let Some(document) = store.documents.get(id) else {
bail!("embedded document {id:?} is missing from the store");
};
for (chunk, stored) in document.chunks.iter().enumerate() {
let Some(embedding) = &stored.embedding else {
continue;
};
ensure!(
embedding.len() == dim,
"stored embedding has {} dimensions, the embedding space {dim}",
embedding.len()
);
ensure!(
rows.len() < count,
"store counts {count} embedded chunks but holds more"
);
embeddings.push(embedding.clone());
rows.push(RowRef {
document: Arc::clone(document),
chunk: u32::try_from(chunk).context("chunk index over u32")?,
});
}
}
ensure!(
rows.len() == count,
"store counts {count} embedded chunks but holds {}",
rows.len()
);
Ok(Some(VectorCapture {
generation: store.generation,
dim,
rows,
embeddings,
}))
}
pub(crate) fn build_vector_snapshot(
capture: VectorCapture,
graph_min_rows: usize,
) -> Result<VectorSnapshot> {
let VectorCapture {
generation,
dim,
rows,
embeddings,
} = capture;
let members = rows
.iter()
.map(|row| Arc::as_ptr(&row.document) as usize)
.collect();
let embeddings: Arc<[Embedding]> = embeddings.into();
let exact = ExactScan {
dim,
rows: Arc::clone(&embeddings),
};
let graph = if rows.len() >= graph_min_rows.max(SNAPSHOT_MIN_ROWS) {
let mut matrix = Vec::new();
matrix
.try_reserve_exact(rows.len() * dim)
.context("cannot allocate the graph build matrix")?;
for embedding in embeddings.iter() {
matrix.extend_from_slice(embedding);
}
let built = Graph::build_owned(matrix, dim, &BuildParams::default())
.map_err(|e| anyhow::anyhow!("graph build: {e}"))?;
let encoded =
VGraph::from_graph(&built, true).map_err(|e| anyhow::anyhow!("graph encode: {e}"))?;
drop(built);
crate::core::os_memory::release_free_heap();
ensure!(
encoded.len() == rows.len(),
"graph index has {} vertices for {} rows",
encoded.len(),
rows.len()
);
Some(GraphIndex {
graph: encoded,
exact: ExactScan {
dim,
rows: Arc::clone(&embeddings),
},
})
} else {
None
};
Ok(VectorSnapshot {
generation,
dim,
rows,
members,
exact,
graph,
})
}
fn staleness(store_generation: u64, snapshot_generation: u64) -> u64 {
if store_generation == snapshot_generation {
0
} else {
store_generation.abs_diff(snapshot_generation).max(1)
}
}
pub(crate) fn vector_index_status(
snapshot: Option<&VectorSnapshot>,
store: &DocumentStore,
) -> VectorIndexStatus {
match snapshot {
None => VectorIndexStatus {
rows: 0,
generation: 0,
current: false,
stale_mutations: 0,
graph: false,
rebuild_recommended: store.embedded_count() >= SNAPSHOT_MIN_ROWS,
},
Some(snapshot) => {
let stale = staleness(store.generation, snapshot.generation);
VectorIndexStatus {
rows: snapshot.rows(),
generation: snapshot.generation,
current: stale == 0,
stale_mutations: stale,
graph: snapshot.has_graph(),
rebuild_recommended: stale >= REBUILD_AFTER_MUTATIONS,
}
}
}
}
struct Hit {
document: Arc<StoredDocument>,
chunk: u32,
score: f64,
}
fn to_result(hit: &Hit, segment: Option<&TextSegment>) -> DocumentResult {
let document = &hit.document;
let chunk = &document.chunks[hit.chunk as usize];
let score = hit.score.clamp(-1.0, 1.0);
DocumentResult {
id: chunk.id.clone(),
document_id: document.id.clone(),
source_uri: document.source_uri.clone(),
version: document.version.clone(),
metadata: document.metadata.clone(),
text: text_of(chunk, segment),
start_byte: chunk.start_byte,
end_byte: chunk.end_byte,
score,
lexical_score: 0.0,
semantic_score: Some(score),
}
}
pub(crate) fn search_vector_snapshot(
snapshot: &VectorSnapshot,
store: &RwLock<DocumentStore>,
query: &[f32],
exact_query: &[f64],
k: usize,
options: &VectorQueryOptions,
) -> Result<Option<Vec<DocumentResult>>> {
let Some(ranked) = rank_vector_snapshot(snapshot, store, query, exact_query, k, options)?
else {
return Ok(None);
};
Ok(read_vector_hits(ranked, &store.read()))
}
pub(crate) struct RankedVectorHits {
hits: Vec<Hit>,
segment: Option<Arc<TextSegment>>,
}
pub(crate) fn read_vector_hits(
ranked: RankedVectorHits,
_store: &DocumentStore,
) -> Option<Vec<DocumentResult>> {
let segment = ranked.segment.as_deref();
ranked
.hits
.iter()
.map(|hit| {
let scrubbed = text_scrubbed(&hit.document.chunks[hit.chunk as usize], segment);
(!scrubbed).then(|| to_result(hit, segment))
})
.collect()
}
pub(crate) fn rank_vector_snapshot(
snapshot: &VectorSnapshot,
store: &RwLock<DocumentStore>,
query: &[f32],
exact_query: &[f64],
k: usize,
options: &VectorQueryOptions,
) -> Result<Option<RankedVectorHits>> {
ensure!(
query.len() == snapshot.dim,
"query has {} dimensions, the index {}",
query.len(),
snapshot.dim
);
ensure!((1..=MAX_K).contains(&k), "require 1 <= k <= {MAX_K}");
ensure!(
(1..=MAX_EF).contains(&options.ef),
"require 1 <= ef <= {MAX_EF}"
);
ensure!(
options.min_score.is_finite() && (-1.0..=1.0).contains(&options.min_score),
"minScore must be in [-1,1]"
);
let index = snapshot.index(options.mode)?;
let (generation, segment) = {
let store = store.read();
(store.generation, store.text_segment())
};
let stale = staleness(generation, snapshot.generation);
let extra = usize::try_from(stale).unwrap_or(usize::MAX).min(4 * k);
let fetch = (k + extra + 8).min(snapshot.rows());
let hits = index.search(query, fetch, options.ef)?;
let mut found: Vec<Hit> = Vec::with_capacity(fetch);
if stale == 0 {
for (row, score) in hits {
let row = &snapshot.rows[row as usize];
found.push(Hit {
document: Arc::clone(&row.document),
chunk: row.chunk,
score: f64::from(score),
});
}
} else {
let store = store.read();
for (row, score) in hits {
let row = &snapshot.rows[row as usize];
let live = store
.get(&row.document.id)
.is_some_and(|live| Arc::ptr_eq(live, &row.document));
if live {
found.push(Hit {
document: Arc::clone(&row.document),
chunk: row.chunk,
score: f64::from(score),
});
}
}
for document in store.documents.values() {
if snapshot.members.contains(&(Arc::as_ptr(document) as usize)) {
continue;
}
for (chunk, stored) in document.chunks.iter().enumerate() {
let Some(embedding) = &stored.embedding else {
continue;
};
if embedding.len() != snapshot.dim {
continue;
}
found.push(Hit {
document: Arc::clone(document),
chunk: u32::try_from(chunk).context("chunk index over u32")?,
score: f64::from(dot(query, embedding)),
});
}
}
drop(store);
if found.len() < k.min(snapshot.rows()) {
return Ok(None);
}
}
for hit in &mut found {
if let Some(embedding) = hit
.document
.chunks
.get(hit.chunk as usize)
.and_then(|chunk| chunk.embedding.as_ref())
{
hit.score = embedding.cosine(exact_query);
}
}
found.sort_by(|a, b| {
b.score
.total_cmp(&a.score)
.then_with(|| a.document.id.cmp(&b.document.id))
.then(a.chunk.cmp(&b.chunk))
});
found.retain(|hit| hit.score >= options.min_score);
found.truncate(k);
Ok(Some(RankedVectorHits {
hits: found,
segment,
}))
}
#[derive(Default)]
pub(crate) struct VectorIndexSlot {
current: RwLock<Option<Arc<VectorSnapshot>>>,
building: Mutex<()>,
}
impl VectorIndexSlot {
pub(crate) fn get(&self) -> Option<Arc<VectorSnapshot>> {
self.current.read().clone()
}
pub(crate) fn publish(&self, snapshot: Arc<VectorSnapshot>) {
*self.current.write() = Some(snapshot);
}
pub(crate) fn clear(&self) {
*self.current.write() = None;
}
pub(crate) fn try_build(&self) -> Option<MutexGuard<'_, ()>> {
self.building.try_lock()
}
pub(crate) fn build(&self) -> MutexGuard<'_, ()> {
self.building.lock()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::documents::{prepare, DocumentInput};
use crate::core::encoding::TextEncoder;
use serde_json::json;
fn matrix(rows: &[[f32; 4]]) -> ExactScan {
let rows: Vec<Embedding> = rows
.iter()
.map(|row| {
let norm = row.iter().map(|v| v * v).sum::<f32>().sqrt();
Embedding::from_f32(row.iter().map(|v| v / norm).collect())
})
.collect();
ExactScan {
dim: 4,
rows: rows.into(),
}
}
#[test]
fn exact_scan_orders_by_score_then_row_and_bounds_k() {
let scan = matrix(&[
[0.0, 1.0, 0.0, 0.0],
[1.0, 0.0, 0.0, 0.0],
[1.0, 0.0, 0.0, 0.0],
[0.5, 0.5, 0.0, 0.0],
]);
let hits = scan.search(&[1.0, 0.0, 0.0, 0.0], 3, 0).unwrap();
assert_eq!(hits.iter().map(|h| h.0).collect::<Vec<_>>(), [1, 2, 3]);
assert!(
(hits[0].1 - 1.0).abs() < 1e-6
&& (hits[2].1 - std::f32::consts::FRAC_1_SQRT_2).abs() < 1e-3
);
assert_eq!(scan.search(&[1.0, 0.0, 0.0, 0.0], 10, 0).unwrap().len(), 4);
assert!(scan.search(&[1.0, 0.0, 0.0], 1, 0).is_err());
}
#[test]
fn dot_matches_the_scalar_sum_with_a_remainder() {
let a: Vec<f32> = (0..19).map(|i| i as f32 * 0.25).collect();
let b: Vec<f32> = (0..19).map(|i| 1.0 - i as f32 * 0.1).collect();
let expected: f32 = a.iter().zip(&b).map(|(x, y)| x * y).sum();
assert!((dot(&a, &b) - expected).abs() < 1e-4);
}
#[test]
fn snapshot_rows_share_the_store_embeddings() {
const DIM: usize = 16;
let axis = |lead: usize, second: usize, value: f64| -> Vec<f64> {
let mut values = vec![0.0; DIM];
values[lead] = 1.0;
values[second] = value;
values
};
let mut store = DocumentStore::default();
for i in 0..40 {
let input: DocumentInput = serde_json::from_value(json!({
"id": format!("d{i:03}"),
"text": format!("document {i} with a few words of text for two chunks here"),
"chunkWords": 6, "overlapWords": 0,
"embeddings": [axis(0, 1, i as f64 / 50.0), axis(2, 3, i as f64 / 40.0)]
}))
.unwrap();
let document = prepare(input, 1024, Some(DIM), TextEncoder::default()).unwrap();
store.insert(document.id.clone(), document);
}
let capture = capture_vector_rows(&store, DIM).unwrap().unwrap();
assert_eq!(capture.rows.len(), 80);
for (row, embedding) in capture.rows.iter().zip(&capture.embeddings) {
let stored = store.documents[&row.document.id].chunks[row.chunk as usize]
.embedding
.as_ref()
.unwrap();
assert!(embedding.shares_allocation(stored));
}
for graph_min_rows in [usize::MAX, 1] {
let capture = capture_vector_rows(&store, DIM).unwrap().unwrap();
let snapshot = build_vector_snapshot(capture, graph_min_rows).unwrap();
assert_eq!(snapshot.has_graph(), graph_min_rows == 1);
for (row, embedding) in snapshot.rows.iter().zip(snapshot.exact.rows.iter()) {
let stored = store.documents[&row.document.id].chunks[row.chunk as usize]
.embedding
.as_ref()
.unwrap();
assert!(embedding.shares_allocation(stored));
}
let query = Embedding::from_f64(
&super::super::normalize_embedding(&axis(0, 1, 39.0 / 50.0), DIM).unwrap(),
);
let hits = snapshot
.index(VectorIndexMode::Auto)
.unwrap()
.search(&query, 1, DEFAULT_EF)
.unwrap();
let row = &snapshot.rows[hits[0].0 as usize];
assert_eq!((row.document.id.as_str(), row.chunk), ("d039", 0));
assert!((hits[0].1 - 1.0).abs() < 1e-6);
}
}
}