uqa_storage/diskann_index/
changes.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Bounded journal reclamation compares immutable origins, never writer allocation order.
8
9use super::format::{DiskANNCanonicalOrigin, DiskANNChangeIdentity, DiskANNGeneration};
10use super::pages::DiskANNOriginReader;
11use crate::{read_control::StorageReadControl, StorageBackendError, StorageBackendResult};
12use uqa_core::DocId;
13
14/// Continuation within one selected generation. A page with deletions advances only after commit; a zero-deletion page may advance after confirmed rollback. Restart a new pass after reaching the end to discover later insertions before this key.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub struct DiskANNPruneCursor {
17    generation: DiskANNGeneration,
18    after: DiskANNChangeIdentity,
19}
20
21/// One bounded page. At most 64 journal entries are examined even if a larger maximum is supplied; zero is invalid.
22#[derive(Debug, Clone, Copy)]
23pub struct DiskANNPruneRequest {
24    pub after: Option<DiskANNPruneCursor>,
25    pub max_records: usize,
26}
27
28/// Evaluated deletions, not a commit receipt. `next == None` means the end of this pass was observed, not that future writes cannot add earlier keys.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub struct DiskANNPruneResult {
31    pub examined: usize,
32    pub removed: usize,
33    pub next: Option<DiskANNPruneCursor>,
34}
35
36/// One finite journal-discovery snapshot bound to the original catalog, selected generation and backend session. Each page evaluates current committed origins in the caller's active transaction. Deletions require confirmed commit before cursor advancement; a zero-deletion page permits confirmed rollback. Later journal keys wait for the next pass, and already-deleted keys still count toward the page limit.
37pub trait DiskANNJournalPruner: Send + Sync {
38    fn prune(
39        &self,
40        request: DiskANNPruneRequest,
41        control: &StorageReadControl,
42    ) -> StorageBackendResult<DiskANNPruneResult>;
43}
44
45/// Actual provider journal keys and complete canonical origins. Pruning may retain older discovery with current deletion evidence; exact statistics use one fixed view for every read. Callbacks must not advance either view.
46pub trait DiskANNChangeRead {
47    fn next_after(
48        &self,
49        after: Option<DiskANNChangeIdentity>,
50        control: &StorageReadControl,
51    ) -> StorageBackendResult<Option<DiskANNChangeIdentity>>;
52    /// Read the value on the supplied evidence view. Pruning uses its current transaction, where an older discovery key may already be absent; statistics use the same retained view for discovery and evidence.
53    fn change(
54        &self,
55        identity: DiskANNChangeIdentity,
56        control: &StorageReadControl,
57    ) -> StorageBackendResult<Option<DiskANNCanonicalOrigin>>;
58    fn current_origin(
59        &self,
60        document: DocId,
61        control: &StorageReadControl,
62    ) -> StorageBackendResult<Option<DiskANNCanonicalOrigin>>;
63}
64
65/// Exact-key deletion shares the reader's publishing transaction and its catalog/head guards.
66pub trait DiskANNChangeJournal {
67    fn read(&self) -> &dyn DiskANNChangeRead;
68    fn remove(
69        &mut self,
70        identity: DiskANNChangeIdentity,
71        control: &StorageReadControl,
72    ) -> StorageBackendResult<()>;
73}
74
75pub(crate) fn prune(
76    origins: &DiskANNOriginReader,
77    journal: &mut dyn DiskANNChangeJournal,
78    request: DiskANNPruneRequest,
79    control: &StorageReadControl,
80) -> StorageBackendResult<DiskANNPruneResult> {
81    control.check()?;
82    let generation = origins.manifest().input().generation;
83    if request.max_records == 0
84        || request
85            .after
86            .is_some_and(|cursor| cursor.generation != generation)
87    {
88        return Err(invalid("invalid pruning limit or generation cursor"));
89    }
90    let mut after = request.after.map(|cursor| cursor.after);
91    let mut result = DiskANNPruneResult {
92        examined: 0,
93        removed: 0,
94        next: None,
95    };
96    for _ in 0..request.max_records.min(64) {
97        control.check()?;
98        let Some(identity) = journal.read().next_after(after, control)? else {
99            result.next = None;
100            control.check()?;
101            return Ok(result);
102        };
103        if after.is_some_and(|last| last.encode() >= identity.encode()) {
104            return Err(invalid("journal pruning cursor did not advance"));
105        }
106        result.examined += 1;
107        after = Some(identity);
108        result.next = Some(DiskANNPruneCursor {
109            generation,
110            after: identity,
111        });
112        // A committed mutation never regains a superseded origin: undo cannot reuse revisions and receipt recovery never reevaluates writes. Older readers retain historical journal rows through MVCC.
113        if matches!(
114            classify(origins, journal.read(), identity, control)?,
115            Change::Reclaimable
116        ) {
117            journal.remove(identity, control)?;
118            result.removed += 1;
119        }
120    }
121    control.check()?;
122    Ok(result)
123}
124
125pub(super) enum Change {
126    Missing,
127    Reclaimable,
128    Outstanding(DiskANNCanonicalOrigin),
129}
130
131pub(super) fn classify<J: DiskANNChangeRead + ?Sized>(
132    origins: &DiskANNOriginReader,
133    journal: &J,
134    identity: DiskANNChangeIdentity,
135    control: &StorageReadControl,
136) -> StorageBackendResult<Change> {
137    let Some(change) = journal.change(identity, control)? else {
138        return Ok(Change::Missing);
139    };
140    if change.version() != identity.version() {
141        return Err(invalid(
142            "journal payload differs from its mutation identity",
143        ));
144    }
145    if change.dimensions() != origins.manifest().input().dimensions {
146        return Err(invalid(
147            "journal dimensions differ from the published generation",
148        ));
149    }
150    let current = journal.current_origin(identity.document(), control)?;
151    if current.is_some_and(|origin| origin.version() == identity.version() && origin != change) {
152        return Err(invalid("journal payload differs from the current origin"));
153    }
154    let covered = origins.origin(identity.document(), control)?;
155    if covered.is_some_and(|origin| origin.version() == identity.version() && origin != change) {
156        return Err(invalid("journal payload differs from the published origin"));
157    }
158    Ok(
159        if covered == Some(change)
160            || current.map(DiskANNCanonicalOrigin::version) != Some(identity.version())
161        {
162            Change::Reclaimable
163        } else {
164            Change::Outstanding(change)
165        },
166    )
167}
168
169fn invalid(message: &'static str) -> StorageBackendError {
170    crate::mvcc::VersionError::InvalidEncoding(message).into_storage_error()
171}
uqa_storage/diskann_index/changes.rs

uqa_storage/diskann_index/
changes.rs
