Skip to main content

writerslogic_warm/core/documents/
vector_index.rs

1// Copyright 2024-2026 WritersLogic Contributors
2// SPDX-License-Identifier: Apache-2.0
3
4//! Vector query dispatch for the document store.
5//!
6//! Mechanism: a vector-only, unfiltered query does not need the lexical capture in
7//! `search.rs`, which clones one `Arc<StoredDocument>` per stored chunk and scores rows
8//! through the document pointers on every query. It is answered from a [`VectorSnapshot`]:
9//! the embedded chunks captured at one store generation as a row list over the store's own
10//! `f32` embeddings ([`super::Embedding`], one shared allocation per chunk, so the snapshot
11//! adds no copy of the matrix), with an optional vertex graph (`qgraph::VGraph`) over the
12//! same rows. Every [`VectorIndex`] maps a query to row ids and the snapshot maps rows back
13//! to documents. Readers share the snapshot through one `Arc`, so a query holds the store's
14//! read lock only to read the generation and the text segment (and, when stale, to validate
15//! hits).
16//!
17//! Data layout: `rows[i]` owns the document of row `i` and the chunk index inside it;
18//! `exact.rows[i]` is that chunk's embedding handle, l2-normalised `f32` values rounded from
19//! the `f64` normalisation at ingest. `members` holds the address of every document
20//! allocation the snapshot covers; the addresses stay unique while the snapshot holds the
21//! `Arc`s, so a replaced document (a new allocation under the same id) is never mistaken for
22//! the captured one. The graph build needs one contiguous matrix; it is assembled for the
23//! build and freed with the builder, so it is transient memory, not part of the snapshot.
24//!
25//! Consistency: the store bumps `generation` on every insert, replace and delete
26//! (`index.rs`), and a snapshot records the generation it was captured at.
27//! - current (`store.generation == snapshot.generation`): the rows are the live store, hits
28//!   are returned as found.
29//! - stale: every hit is validated by identity against the live store under its read lock.
30//!   A deleted or replaced document's `Arc` is no longer the one in `store.documents`, so
31//!   its rows drop out (the tombstone), and the search over-fetches up to `4k` rows to cover
32//!   them. Documents the snapshot does not cover (inserted or replaced since the capture)
33//!   are scored in a delta pass over the live documents: one walk over `store.documents`
34//!   with a hash probe per document plus a dot product per uncovered chunk. That pass is
35//!   bounded by the store size and never costs more than the generic exact path; when fewer
36//!   than `k` hits survive, the caller falls back to that path. `rebuild_recommended` asks
37//!   maintenance for a fresh capture after [`REBUILD_AFTER_MUTATIONS`] mutations, so the
38//!   stale window is bounded by the maintenance interval.
39//! - third implementation slot: [`VectorIndexKind::Scan`] is reserved for the graph-free
40//!   scan under research in `benchmarks/research/scan`; nothing implements it here.
41//!
42//! Scores: exact-mode selection uses [`super::Embedding::cosine`], the canonical `f64`
43//! accumulation. Graph candidates use the `f32` dot before canonical re-scoring; graph
44//! membership remains approximate. Ties break on `(document id, chunk)` ascending.
45
46use std::{cmp::Ordering, collections::BinaryHeap, sync::Arc};
47
48use crate::core::hash::FastSet;
49use crate::core::sync::{Mutex, MutexGuard, RwLock};
50use anyhow::{bail, ensure, Context, Result};
51use serde::{Deserialize, Serialize};
52
53use super::index::{text_of, text_scrubbed};
54use super::{DocumentResult, DocumentStore, Embedding, StoredDocument, TextSegment};
55use crate::core::qgraph::{BuildParams, Graph, VGraph, VSearchParams};
56
57/// Fewest embedded chunks worth a snapshot; smaller stores take the generic exact path.
58pub const SNAPSHOT_MIN_ROWS: usize = 1;
59/// Default `graph_min_rows`: a snapshot with fewer rows carries only the exact scan, which
60/// beats a graph traversal at that size.
61pub const DEFAULT_GRAPH_MIN_ROWS: usize = 4096;
62/// Default beam width of a graph query. The accepted qgraph rows reached recall 0.95 at
63/// `ef` 736 on glove-100-angular (1.18M vectors, `benchmarks/results/ann_modal_confirm_v1.json`);
64/// stores of up to a few hundred thousand embedding rows reach it with a far smaller beam, and
65/// `tests/server_ann.rs` holds the default to recall >= 0.95 on its corpus.
66pub const DEFAULT_EF: usize = 128;
67pub const MAX_EF: usize = 4096;
68pub const MAX_K: usize = 1000;
69/// Mutations since the capture after which maintenance builds a fresh snapshot.
70pub const REBUILD_AFTER_MUTATIONS: u64 = 64;
71const MAX_ROWS: usize = 1 << 22;
72/// Upper bound on the graph build matrix: 512 Mi `f32` values (2 GiB).
73const MAX_MATRIX_VALUES: usize = 512 * 1024 * 1024;
74
75/// Which implementation answered, or may answer, a vector query.
76#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
77#[serde(rename_all = "lowercase")]
78pub enum VectorIndexKind {
79    /// Exact cosine over the store's `f32` rows.
80    Exact,
81    /// The vertex graph (`qgraph::VGraph`) with exact re-scoring of the fetched rows.
82    Graph,
83    /// Reserved for the graph-free scan under research in `benchmarks/research/scan`.
84    Scan,
85}
86
87/// Route of one vector query.
88#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(rename_all = "lowercase")]
90pub enum VectorIndexMode {
91    /// The graph when the snapshot has one, the exact scan otherwise.
92    #[default]
93    Auto,
94    /// Always the exact scan (the generic path when no snapshot exists).
95    Exact,
96    /// The graph, or [`GraphUnavailable`] when none is built.
97    Graph,
98}
99
100impl VectorIndexMode {
101    pub fn parse(name: &str) -> Option<Self> {
102        match name {
103            "auto" => Some(Self::Auto),
104            "exact" => Some(Self::Exact),
105            "graph" => Some(Self::Graph),
106            _ => None,
107        }
108    }
109
110    pub fn as_str(self) -> &'static str {
111        match self {
112            Self::Auto => "auto",
113            Self::Exact => "exact",
114            Self::Graph => "graph",
115        }
116    }
117}
118
119/// Per-query options of [`crate::WarmCore::search_vectors`].
120#[derive(Clone, Debug)]
121pub struct VectorQueryOptions {
122    pub mode: VectorIndexMode,
123    /// Beam width of a graph query (`1..=MAX_EF`); ignored by the exact scan.
124    pub ef: usize,
125    /// Lowest cosine returned, in `[-1, 1]`.
126    pub min_score: f64,
127}
128
129impl Default for VectorQueryOptions {
130    fn default() -> Self {
131        Self {
132            mode: VectorIndexMode::Auto,
133            ef: DEFAULT_EF,
134            min_score: -1.0,
135        }
136    }
137}
138
139/// State of the snapshot relative to the live store.
140#[derive(Clone, Debug, Serialize)]
141#[serde(rename_all = "camelCase")]
142pub struct VectorIndexStatus {
143    /// Embedded chunks the snapshot covers (0 without a snapshot).
144    pub rows: usize,
145    /// Store generation the snapshot was captured at.
146    pub generation: u64,
147    /// Whether the snapshot is the live store.
148    pub current: bool,
149    /// Inserts, replaces and deletes since the capture.
150    pub stale_mutations: u64,
151    /// Whether the snapshot carries a graph index.
152    pub graph: bool,
153    /// Whether maintenance should capture a fresh snapshot.
154    pub rebuild_recommended: bool,
155}
156
157/// A `graph` query reached a store without a graph index.
158#[derive(Debug, Clone, Copy)]
159pub struct GraphUnavailable;
160
161impl std::fmt::Display for GraphUnavailable {
162    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163        f.write_str("no graph index is built; run index maintenance")
164    }
165}
166
167impl std::error::Error for GraphUnavailable {}
168
169/// A nearest-neighbour index over the rows of one `VectorSnapshot`.
170pub trait VectorIndex: Send + Sync {
171    fn kind(&self) -> VectorIndexKind;
172    fn rows(&self) -> usize;
173    /// The `k` best rows by cosine, best first, as `(row, score)`; `ef` is the beam of an
174    /// approximate index and ignored by an exact one. `query` is l2-normalised and has the
175    /// snapshot's dimensions.
176    fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>>;
177}
178
179#[derive(Clone, Copy)]
180struct Ranked {
181    score: f32,
182    row: u32,
183}
184
185impl PartialEq for Ranked {
186    fn eq(&self, other: &Self) -> bool {
187        self.cmp(other) == Ordering::Equal
188    }
189}
190impl Eq for Ranked {}
191impl PartialOrd for Ranked {
192    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
193        Some(self.cmp(other))
194    }
195}
196impl Ord for Ranked {
197    /// Greater is worse: a lower score, or the same score at a higher row.
198    fn cmp(&self, other: &Self) -> Ordering {
199        other
200            .score
201            .total_cmp(&self.score)
202            .then(self.row.cmp(&other.row))
203    }
204}
205
206/// Bounded best-`k` set; the heap top is the worst kept candidate.
207struct TopK {
208    heap: BinaryHeap<Ranked>,
209    k: usize,
210}
211
212impl TopK {
213    fn new(k: usize) -> Self {
214        Self {
215            heap: BinaryHeap::with_capacity(k + 1),
216            k,
217        }
218    }
219
220    fn offer(&mut self, row: u32, score: f32) {
221        let candidate = Ranked { score, row };
222        if self.k == 0 {
223            return;
224        }
225        if self.heap.len() < self.k {
226            self.heap.push(candidate);
227        } else if let Some(mut worst) = self.heap.peek_mut() {
228            if candidate < *worst {
229                *worst = candidate;
230            }
231        }
232    }
233
234    fn into_sorted(self) -> Vec<(u32, f32)> {
235        self.heap
236            .into_sorted_vec()
237            .into_iter()
238            .map(|ranked| (ranked.row, ranked.score))
239            .collect()
240    }
241}
242
243/// Dot product with eight independent accumulators so the loop vectorises.
244#[inline]
245fn dot(a: &[f32], b: &[f32]) -> f32 {
246    debug_assert_eq!(a.len(), b.len());
247    let mut lanes = [0.0f32; 8];
248    let (wide_a, tail_a) = a.as_chunks::<8>();
249    let (wide_b, tail_b) = b.as_chunks::<8>();
250    for (x, y) in wide_a.iter().zip(wide_b) {
251        for ((lane, x), y) in lanes.iter_mut().zip(x).zip(y) {
252            *lane += x * y;
253        }
254    }
255    let mut sum = lanes.iter().sum::<f32>();
256    for (x, y) in tail_a.iter().zip(tail_b) {
257        sum += x * y;
258    }
259    sum
260}
261
262/// Exact cosine over the snapshot rows, one query per thread. Each row is the store's own
263/// embedding allocation (one pointer per row, shared with the chunk).
264pub(crate) struct ExactScan {
265    dim: usize,
266    rows: Arc<[Embedding]>,
267}
268
269impl ExactScan {
270    fn row(&self, row: u32) -> &[f32] {
271        &self.rows[row as usize]
272    }
273
274    fn search_canonical(&self, query: &[f64], k: usize) -> Vec<(u32, f64)> {
275        #[derive(Clone, Copy)]
276        struct Candidate {
277            row: u32,
278            score: f64,
279        }
280        impl PartialEq for Candidate {
281            fn eq(&self, other: &Self) -> bool {
282                self.cmp(other) == Ordering::Equal
283            }
284        }
285        impl Eq for Candidate {}
286        impl PartialOrd for Candidate {
287            fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
288                Some(self.cmp(other))
289            }
290        }
291        impl Ord for Candidate {
292            fn cmp(&self, other: &Self) -> Ordering {
293                other
294                    .score
295                    .total_cmp(&self.score)
296                    .then(self.row.cmp(&other.row))
297            }
298        }
299        let k = k.min(self.rows.len());
300        let mut best = BinaryHeap::<Candidate>::with_capacity(k);
301        for (row, values) in self.rows.iter().enumerate() {
302            let candidate = Candidate {
303                row: row as u32,
304                score: values.cosine(query),
305            };
306            if best.len() < k {
307                best.push(candidate);
308            } else if let Some(mut worst) = best.peek_mut() {
309                if candidate < *worst {
310                    *worst = candidate;
311                }
312            }
313        }
314        best.into_sorted_vec()
315            .into_iter()
316            .map(|hit| (hit.row, hit.score))
317            .collect()
318    }
319}
320
321impl VectorIndex for ExactScan {
322    fn kind(&self) -> VectorIndexKind {
323        VectorIndexKind::Exact
324    }
325
326    fn rows(&self) -> usize {
327        self.rows.len()
328    }
329
330    fn search(&self, query: &[f32], k: usize, _ef: usize) -> Result<Vec<(u32, f32)>> {
331        ensure!(
332            query.len() == self.dim,
333            "query has {} dimensions, the index {}",
334            query.len(),
335            self.dim
336        );
337        let mut best = TopK::new(k.min(self.rows()));
338        for (row, values) in self.rows.iter().enumerate() {
339            best.offer(row as u32, dot(query, values));
340        }
341        Ok(best.into_sorted())
342    }
343}
344
345/// The vertex graph over the snapshot rows; fetched ids are re-scored exactly.
346pub(crate) struct GraphIndex {
347    graph: VGraph,
348    exact: ExactScan,
349}
350
351impl VectorIndex for GraphIndex {
352    fn kind(&self) -> VectorIndexKind {
353        VectorIndexKind::Graph
354    }
355
356    fn rows(&self) -> usize {
357        self.exact.rows()
358    }
359
360    fn search(&self, query: &[f32], k: usize, ef: usize) -> Result<Vec<(u32, f32)>> {
361        ensure!(
362            query.len() == self.exact.dim,
363            "query has {} dimensions, the index {}",
364            query.len(),
365            self.exact.dim
366        );
367        let k = k.min(self.rows());
368        if k == 0 {
369            return Ok(Vec::new());
370        }
371        let ef = ef.clamp(k, MAX_EF.max(k));
372        let params = VSearchParams {
373            ef,
374            rerank: ef,
375            ..VSearchParams::default()
376        };
377        let mut ids = Vec::with_capacity(k);
378        self.graph
379            .searcher()
380            .search(query, k, params, &mut ids)
381            .map_err(|e| anyhow::anyhow!("graph search: {e}"))?;
382        let mut hits: Vec<(u32, f32)> = ids
383            .into_iter()
384            .filter(|&id| (id as usize) < self.rows())
385            .map(|id| (id, dot(query, self.exact.row(id))))
386            .collect();
387        hits.sort_by(|a, b| b.1.total_cmp(&a.1).then(a.0.cmp(&b.0)));
388        hits.dedup_by(|a, b| a.0 == b.0);
389        hits.truncate(k);
390        Ok(hits)
391    }
392}
393
394struct RowRef {
395    document: Arc<StoredDocument>,
396    chunk: u32,
397}
398
399/// The embedded chunks of one store generation with the indexes over them.
400pub(crate) struct VectorSnapshot {
401    generation: u64,
402    dim: usize,
403    rows: Vec<RowRef>,
404    members: FastSet<usize>,
405    exact: ExactScan,
406    graph: Option<GraphIndex>,
407}
408
409impl VectorSnapshot {
410    pub(crate) fn rows(&self) -> usize {
411        self.rows.len()
412    }
413
414    pub(crate) fn has_graph(&self) -> bool {
415        self.graph.is_some()
416    }
417
418    fn index(&self, mode: VectorIndexMode) -> Result<&dyn VectorIndex> {
419        match mode {
420            VectorIndexMode::Exact => Ok(&self.exact),
421            VectorIndexMode::Graph => self
422                .graph
423                .as_ref()
424                .map(|graph| graph as &dyn VectorIndex)
425                .ok_or_else(|| anyhow::Error::new(GraphUnavailable)),
426            VectorIndexMode::Auto => Ok(self
427                .graph
428                .as_ref()
429                .map_or(&self.exact as &dyn VectorIndex, |graph| {
430                    graph as &dyn VectorIndex
431                })),
432        }
433    }
434}
435
436/// The rows captured under the store's read lock; [`build_vector_snapshot`] turns it into a
437/// snapshot without any lock. `embeddings[i]` is the handle of `rows[i]`'s chunk embedding,
438/// shared with the store (a reference count, not a copy).
439pub(crate) struct VectorCapture {
440    generation: u64,
441    dim: usize,
442    rows: Vec<RowRef>,
443    embeddings: Vec<Embedding>,
444}
445
446/// Captures every embedded chunk's embedding handle. `None` when the store has fewer than
447/// [`SNAPSHOT_MIN_ROWS`] embedded chunks.
448pub(crate) fn capture_vector_rows(
449    store: &DocumentStore,
450    dim: usize,
451) -> Result<Option<VectorCapture>> {
452    let count = store.embedded_count();
453    if dim == 0 || count < SNAPSHOT_MIN_ROWS {
454        return Ok(None);
455    }
456    ensure!(
457        count <= MAX_ROWS,
458        "vector snapshot exceeds {MAX_ROWS} embedded chunks"
459    );
460    ensure!(
461        count
462            .checked_mul(dim)
463            .is_some_and(|values| values <= MAX_MATRIX_VALUES),
464        "vector snapshot exceeds {MAX_MATRIX_VALUES} embedding values"
465    );
466    let mut rows = Vec::new();
467    rows.try_reserve_exact(count)
468        .context("cannot allocate vector snapshot rows")?;
469    let mut embeddings = Vec::new();
470    embeddings
471        .try_reserve_exact(count)
472        .context("cannot allocate vector snapshot embedding handles")?;
473    for id in &store.embedded_documents {
474        let Some(document) = store.documents.get(id) else {
475            bail!("embedded document {id:?} is missing from the store");
476        };
477        for (chunk, stored) in document.chunks.iter().enumerate() {
478            let Some(embedding) = &stored.embedding else {
479                continue;
480            };
481            ensure!(
482                embedding.len() == dim,
483                "stored embedding has {} dimensions, the embedding space {dim}",
484                embedding.len()
485            );
486            ensure!(
487                rows.len() < count,
488                "store counts {count} embedded chunks but holds more"
489            );
490            embeddings.push(embedding.clone());
491            rows.push(RowRef {
492                document: Arc::clone(document),
493                chunk: u32::try_from(chunk).context("chunk index over u32")?,
494            });
495        }
496    }
497    ensure!(
498        rows.len() == count,
499        "store counts {count} embedded chunks but holds {}",
500        rows.len()
501    );
502    Ok(Some(VectorCapture {
503        generation: store.generation,
504        dim,
505        rows,
506        embeddings,
507    }))
508}
509
510/// Builds the exact scan and, with at least `graph_min_rows` rows, the vertex graph. The
511/// graph build is deterministic (`BuildParams::default()`, fixed seed) and uses the rayon
512/// pool; its contiguous input matrix (`rows * dim * 4` bytes) lives only for the build.
513pub(crate) fn build_vector_snapshot(
514    capture: VectorCapture,
515    graph_min_rows: usize,
516) -> Result<VectorSnapshot> {
517    let VectorCapture {
518        generation,
519        dim,
520        rows,
521        embeddings,
522    } = capture;
523    let members = rows
524        .iter()
525        .map(|row| Arc::as_ptr(&row.document) as usize)
526        .collect();
527    let embeddings: Arc<[Embedding]> = embeddings.into();
528    let exact = ExactScan {
529        dim,
530        rows: Arc::clone(&embeddings),
531    };
532    let graph = if rows.len() >= graph_min_rows.max(SNAPSHOT_MIN_ROWS) {
533        let mut matrix = Vec::new();
534        matrix
535            .try_reserve_exact(rows.len() * dim)
536            .context("cannot allocate the graph build matrix")?;
537        for embedding in embeddings.iter() {
538            matrix.extend_from_slice(embedding);
539        }
540        let built = Graph::build_owned(matrix, dim, &BuildParams::default())
541            .map_err(|e| anyhow::anyhow!("graph build: {e}"))?;
542        let encoded =
543            VGraph::from_graph(&built, true).map_err(|e| anyhow::anyhow!("graph encode: {e}"))?;
544        drop(built);
545        // The builder's matrix (`rows * dim * 4` bytes) and its per-vertex neighbour lists
546        // are freed now; hand the allocator's free pages back so they leave the resident set.
547        crate::core::os_memory::release_free_heap();
548        ensure!(
549            encoded.len() == rows.len(),
550            "graph index has {} vertices for {} rows",
551            encoded.len(),
552            rows.len()
553        );
554        Some(GraphIndex {
555            graph: encoded,
556            exact: ExactScan {
557                dim,
558                rows: Arc::clone(&embeddings),
559            },
560        })
561    } else {
562        None
563    };
564    Ok(VectorSnapshot {
565        generation,
566        dim,
567        rows,
568        members,
569        exact,
570        graph,
571    })
572}
573
574fn staleness(store_generation: u64, snapshot_generation: u64) -> u64 {
575    if store_generation == snapshot_generation {
576        0
577    } else {
578        store_generation.abs_diff(snapshot_generation).max(1)
579    }
580}
581
582pub(crate) fn vector_index_status(
583    snapshot: Option<&VectorSnapshot>,
584    store: &DocumentStore,
585) -> VectorIndexStatus {
586    match snapshot {
587        None => VectorIndexStatus {
588            rows: 0,
589            generation: 0,
590            current: false,
591            stale_mutations: 0,
592            graph: false,
593            rebuild_recommended: store.embedded_count() >= SNAPSHOT_MIN_ROWS,
594        },
595        Some(snapshot) => {
596            let stale = staleness(store.generation, snapshot.generation);
597            VectorIndexStatus {
598                rows: snapshot.rows(),
599                generation: snapshot.generation,
600                current: stale == 0,
601                stale_mutations: stale,
602                graph: snapshot.has_graph(),
603                rebuild_recommended: stale >= REBUILD_AFTER_MUTATIONS,
604            }
605        }
606    }
607}
608
609struct Hit {
610    document: Arc<StoredDocument>,
611    chunk: u32,
612    /// Candidate score from the `f32` scan, replaced by the `f64` re-score before ranking.
613    score: f64,
614}
615
616fn to_result(hit: &Hit, segment: Option<&TextSegment>) -> DocumentResult {
617    let document = &hit.document;
618    let chunk = &document.chunks[hit.chunk as usize];
619    let score = hit.score.clamp(-1.0, 1.0);
620    DocumentResult {
621        id: chunk.id.clone(),
622        document_id: document.id.clone(),
623        source_uri: document.source_uri.clone(),
624        version: document.version.clone(),
625        metadata: document.metadata.clone(),
626        text: text_of(chunk, segment),
627        start_byte: chunk.start_byte,
628        end_byte: chunk.end_byte,
629        score,
630        lexical_score: 0.0,
631        semantic_score: Some(score),
632    }
633}
634
635/// Answers an unfiltered vector query from the snapshot. `query` is l2-normalised. `None`
636/// means the snapshot is too stale to answer this query (fewer than `k` hits survived
637/// validation, or a hit's text was scrubbed before it was read); the caller takes the
638/// generic exact path.
639pub(crate) fn search_vector_snapshot(
640    snapshot: &VectorSnapshot,
641    store: &RwLock<DocumentStore>,
642    query: &[f32],
643    exact_query: &[f64],
644    k: usize,
645    options: &VectorQueryOptions,
646) -> Result<Option<Vec<DocumentResult>>> {
647    let Some(ranked) = rank_vector_snapshot(snapshot, store, query, exact_query, k, options)?
648    else {
649        return Ok(None);
650    };
651    Ok(read_vector_hits(ranked, &store.read()))
652}
653
654/// The top hits of [`search_vector_snapshot`], ranked with no lock held across the scan;
655/// [`read_vector_hits`] reads their text.
656pub(crate) struct RankedVectorHits {
657    hits: Vec<Hit>,
658    segment: Option<Arc<TextSegment>>,
659}
660
661/// The hits with their text, or `None` when the text of one was scrubbed since the ranking
662/// (its document replaced or deleted). `_store` is the store's read guard: a scrub happens
663/// under its write lock, so a read never overlaps one.
664pub(crate) fn read_vector_hits(
665    ranked: RankedVectorHits,
666    _store: &DocumentStore,
667) -> Option<Vec<DocumentResult>> {
668    let segment = ranked.segment.as_deref();
669    ranked
670        .hits
671        .iter()
672        .map(|hit| {
673            let scrubbed = text_scrubbed(&hit.document.chunks[hit.chunk as usize], segment);
674            (!scrubbed).then(|| to_result(hit, segment))
675        })
676        .collect()
677}
678
679pub(crate) fn rank_vector_snapshot(
680    snapshot: &VectorSnapshot,
681    store: &RwLock<DocumentStore>,
682    query: &[f32],
683    exact_query: &[f64],
684    k: usize,
685    options: &VectorQueryOptions,
686) -> Result<Option<RankedVectorHits>> {
687    ensure!(
688        query.len() == snapshot.dim,
689        "query has {} dimensions, the index {}",
690        query.len(),
691        snapshot.dim
692    );
693    ensure!(
694        exact_query.len() == snapshot.dim,
695        "canonical query dimension mismatch"
696    );
697    ensure!((1..=MAX_K).contains(&k), "require 1 <= k <= {MAX_K}");
698    ensure!(
699        (1..=MAX_EF).contains(&options.ef),
700        "require 1 <= ef <= {MAX_EF}"
701    );
702    ensure!(
703        options.min_score.is_finite() && (-1.0..=1.0).contains(&options.min_score),
704        "minScore must be in [-1,1]"
705    );
706    let index = snapshot.index(options.mode)?;
707    let (generation, segment) = {
708        let store = store.read();
709        (store.generation, store.text_segment())
710    };
711    let stale = staleness(generation, snapshot.generation);
712    let extra = usize::try_from(stale).unwrap_or(usize::MAX).min(4 * k);
713    let fetch = (k + extra + 8).min(snapshot.rows());
714    let hits = if index.kind() == VectorIndexKind::Exact {
715        snapshot.exact.search_canonical(exact_query, fetch)
716    } else {
717        index
718            .search(query, fetch, options.ef)?
719            .into_iter()
720            .map(|(row, score)| (row, f64::from(score)))
721            .collect()
722    };
723    let mut found: Vec<Hit> = Vec::with_capacity(fetch);
724    if stale == 0 {
725        for (row, score) in hits {
726            let row = &snapshot.rows[row as usize];
727            found.push(Hit {
728                document: Arc::clone(&row.document),
729                chunk: row.chunk,
730                score,
731            });
732        }
733    } else {
734        let store = store.read();
735        for (row, score) in hits {
736            let row = &snapshot.rows[row as usize];
737            let live = store
738                .get(&row.document.id)
739                .is_some_and(|live| Arc::ptr_eq(live, &row.document));
740            if live {
741                found.push(Hit {
742                    document: Arc::clone(&row.document),
743                    chunk: row.chunk,
744                    score,
745                });
746            }
747        }
748        for document in store.documents.values() {
749            if snapshot.members.contains(&(Arc::as_ptr(document) as usize)) {
750                continue;
751            }
752            for (chunk, stored) in document.chunks.iter().enumerate() {
753                let Some(embedding) = &stored.embedding else {
754                    continue;
755                };
756                if embedding.len() != snapshot.dim {
757                    continue;
758                }
759                found.push(Hit {
760                    document: Arc::clone(document),
761                    chunk: u32::try_from(chunk).context("chunk index over u32")?,
762                    score: f64::from(dot(query, embedding)),
763                });
764            }
765        }
766        drop(store);
767        if found.len() < k.min(snapshot.rows()) {
768            return Ok(None);
769        }
770    }
771    for hit in &mut found {
772        if let Some(embedding) = hit
773            .document
774            .chunks
775            .get(hit.chunk as usize)
776            .and_then(|chunk| chunk.embedding.as_ref())
777        {
778            hit.score = embedding.cosine(exact_query);
779        }
780    }
781    found.sort_by(|a, b| {
782        b.score
783            .total_cmp(&a.score)
784            .then_with(|| a.document.id.cmp(&b.document.id))
785            .then(a.chunk.cmp(&b.chunk))
786    });
787    found.retain(|hit| hit.score >= options.min_score);
788    found.truncate(k);
789    Ok(Some(RankedVectorHits {
790        hits: found,
791        segment,
792    }))
793}
794
795/// The engine's published snapshot and the build lock.
796#[derive(Default)]
797pub(crate) struct VectorIndexSlot {
798    current: RwLock<Option<Arc<VectorSnapshot>>>,
799    building: Mutex<()>,
800}
801
802impl VectorIndexSlot {
803    pub(crate) fn get(&self) -> Option<Arc<VectorSnapshot>> {
804        self.current.read().clone()
805    }
806
807    pub(crate) fn publish(&self, snapshot: Arc<VectorSnapshot>) {
808        *self.current.write() = Some(snapshot);
809    }
810
811    pub(crate) fn clear(&self) {
812        *self.current.write() = None;
813    }
814
815    /// `None` while another build is running.
816    pub(crate) fn try_build(&self) -> Option<MutexGuard<'_, ()>> {
817        self.building.try_lock()
818    }
819
820    /// The build lock, waiting for a running build to finish.
821    pub(crate) fn build(&self) -> MutexGuard<'_, ()> {
822        self.building.lock()
823    }
824}
825
826#[cfg(test)]
827mod tests {
828    use super::*;
829    use crate::core::documents::{prepare, DocumentInput};
830    use crate::core::encoding::TextEncoder;
831    use serde_json::json;
832
833    fn matrix(rows: &[[f32; 4]]) -> ExactScan {
834        let rows: Vec<Embedding> = rows
835            .iter()
836            .map(|row| {
837                let norm = row.iter().map(|v| v * v).sum::<f32>().sqrt();
838                Embedding::from_f32(row.iter().map(|v| v / norm).collect())
839            })
840            .collect();
841        ExactScan {
842            dim: 4,
843            rows: rows.into(),
844        }
845    }
846
847    #[test]
848    fn exact_scan_orders_by_score_then_row_and_bounds_k() {
849        let scan = matrix(&[
850            [0.0, 1.0, 0.0, 0.0],
851            [1.0, 0.0, 0.0, 0.0],
852            [1.0, 0.0, 0.0, 0.0],
853            [0.5, 0.5, 0.0, 0.0],
854        ]);
855        let hits = scan.search(&[1.0, 0.0, 0.0, 0.0], 3, 0).unwrap();
856        assert_eq!(hits.iter().map(|h| h.0).collect::<Vec<_>>(), [1, 2, 3]);
857        assert!(
858            (hits[0].1 - 1.0).abs() < 1e-6
859                && (hits[2].1 - std::f32::consts::FRAC_1_SQRT_2).abs() < 1e-3
860        );
861        assert_eq!(scan.search(&[1.0, 0.0, 0.0, 0.0], 10, 0).unwrap().len(), 4);
862        assert!(scan.search(&[1.0, 0.0, 0.0], 1, 0).is_err());
863    }
864
865    #[test]
866    fn audit_regression_exact_snapshot_retains_all_canonical_rounding_winners() {
867        for distractors in [8, 9, 10, 100] {
868            let mut store = DocumentStore::default();
869            for i in 0..=distractors {
870                let input: DocumentInput = serde_json::from_value(json!({
871                    "id": if i == distractors { "z".to_owned() } else { format!("a{i:03}") },
872                    "text": "rounding regression",
873                    "embeddings": [[1.0, if i == distractors { 0.0001 } else { 0.0 }]],
874                }))
875                .unwrap();
876                let document = prepare(input, 1024, Some(2), TextEncoder::default()).unwrap();
877                store.insert(document.id.clone(), document);
878            }
879            let snapshot =
880                build_vector_snapshot(capture_vector_rows(&store, 2).unwrap().unwrap(), usize::MAX)
881                    .unwrap();
882            let query = super::super::normalize_embedding(&[1.0, 0.0001], 2).unwrap();
883            let rounded: Vec<f32> = query.iter().map(|&x| x as f32).collect();
884            let store = RwLock::new(store);
885            for minimum in [-1.0, 0.999_999_999] {
886                let hits = search_vector_snapshot(
887                    &snapshot,
888                    &store,
889                    &rounded,
890                    &query,
891                    1,
892                    &VectorQueryOptions {
893                        mode: VectorIndexMode::Exact,
894                        min_score: minimum,
895                        ..Default::default()
896                    },
897                )
898                .unwrap()
899                .unwrap();
900                assert_eq!(
901                    hits.len(),
902                    1,
903                    "{distractors} distractors, threshold {minimum}"
904                );
905                assert_eq!(hits[0].document_id, "z", "{distractors} distractors");
906                assert_eq!(hits[0].semantic_score, Some(1.0));
907            }
908        }
909    }
910
911    #[test]
912    fn dot_matches_the_scalar_sum_with_a_remainder() {
913        let a: Vec<f32> = (0..19).map(|i| i as f32 * 0.25).collect();
914        let b: Vec<f32> = (0..19).map(|i| 1.0 - i as f32 * 0.1).collect();
915        let expected: f32 = a.iter().zip(&b).map(|(x, y)| x * y).sum();
916        assert!((dot(&a, &b) - expected).abs() < 1e-4);
917    }
918
919    /// The snapshot holds the store's embedding allocations, not copies, and the graph-free
920    /// and graph snapshots both answer from them.
921    #[test]
922    fn snapshot_rows_share_the_store_embeddings() {
923        const DIM: usize = 16;
924        let axis = |lead: usize, second: usize, value: f64| -> Vec<f64> {
925            let mut values = vec![0.0; DIM];
926            values[lead] = 1.0;
927            values[second] = value;
928            values
929        };
930        let mut store = DocumentStore::default();
931        for i in 0..40 {
932            let input: DocumentInput = serde_json::from_value(json!({
933                "id": format!("d{i:03}"),
934                "text": format!("document {i} with a few words of text for two chunks here"),
935                "chunkWords": 6, "overlapWords": 0,
936                "embeddings": [axis(0, 1, i as f64 / 50.0), axis(2, 3, i as f64 / 40.0)]
937            }))
938            .unwrap();
939            let document = prepare(input, 1024, Some(DIM), TextEncoder::default()).unwrap();
940            store.insert(document.id.clone(), document);
941        }
942        let capture = capture_vector_rows(&store, DIM).unwrap().unwrap();
943        assert_eq!(capture.rows.len(), 80);
944        for (row, embedding) in capture.rows.iter().zip(&capture.embeddings) {
945            let stored = store.documents[&row.document.id].chunks[row.chunk as usize]
946                .embedding
947                .as_ref()
948                .unwrap();
949            assert!(embedding.shares_allocation(stored));
950        }
951        for graph_min_rows in [usize::MAX, 1] {
952            let capture = capture_vector_rows(&store, DIM).unwrap().unwrap();
953            let snapshot = build_vector_snapshot(capture, graph_min_rows).unwrap();
954            assert_eq!(snapshot.has_graph(), graph_min_rows == 1);
955            for (row, embedding) in snapshot.rows.iter().zip(snapshot.exact.rows.iter()) {
956                let stored = store.documents[&row.document.id].chunks[row.chunk as usize]
957                    .embedding
958                    .as_ref()
959                    .unwrap();
960                assert!(embedding.shares_allocation(stored));
961            }
962            let query = Embedding::from_f64(
963                &super::super::normalize_embedding(&axis(0, 1, 39.0 / 50.0), DIM).unwrap(),
964            );
965            let hits = snapshot
966                .index(VectorIndexMode::Auto)
967                .unwrap()
968                .search(&query, 1, DEFAULT_EF)
969                .unwrap();
970            let row = &snapshot.rows[hits[0].0 as usize];
971            assert_eq!((row.document.id.as_str(), row.chunk), ("d039", 0));
972            assert!((hits[0].1 - 1.0).abs() < 1e-6);
973        }
974    }
975}