uqa-storage 0.4.0

Document store, inverted index, IVF/HNSW vectors, B-tree, R*Tree, catalog
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Checked field-count deltas retain the analyzer revision selected during evaluation.

use super::{
    resolve::{bytes, revision, Resolver},
    OccurrenceRecordValue as Value,
};
use crate::inverted_index::IndexedFieldRevision;
use crate::mvcc::{PreparedRecordWrite, VersionError, VersionResult};

#[derive(Clone, Copy)]
struct Statistics {
    revision: IndexedFieldRevision,
    doc_count: u64,
    total_length: u64,
}

impl Resolver<'_> {
    fn statistics(&self, key: &[u8], value: Option<&[u8]>) -> VersionResult<Option<Statistics>> {
        value
            .map(
                |value| match self.layout.decode(key, value, self.control)? {
                    Value::Statistics {
                        revision,
                        doc_count,
                        total_length,
                    } => Ok(Statistics {
                        revision,
                        doc_count,
                        total_length,
                    }),
                    _ => Err(VersionError::InvalidEncoding(
                        "invalid occurrence statistics projection",
                    )),
                },
            )
            .transpose()
    }

    pub(super) fn merge_statistics(
        &self,
        mutation: usize,
        write: &PreparedRecordWrite,
    ) -> VersionResult<()> {
        let base = self.base.get(write.key(), self.control)?;
        let current = self.current.get(write.key(), self.control)?;
        let before = self.statistics(write.key(), bytes(base.as_ref()))?;
        let after = self.statistics(write.key(), write.value())?;
        let latest = self.statistics(write.key(), bytes(current.as_ref()))?;
        let mut revisions = [before, after, latest]
            .into_iter()
            .flatten()
            .map(|stats| stats.revision);
        let Some(revision_value) = revisions.next() else {
            return self.replace(write.key(), None);
        };
        if revisions.any(|candidate| candidate != revision_value) {
            return Err(VersionError::WriteConflict {
                mutation,
                expected: write.expected(),
                actual: revision(self.current, write.key(), self.control)?,
            });
        }
        let delta = |select: fn(Statistics) -> u64| -> VersionResult<u64> {
            let total = i128::from(latest.map_or(0, select)) + i128::from(after.map_or(0, select))
                - i128::from(before.map_or(0, select));
            u64::try_from(total).map_err(|_| {
                VersionError::InvalidEncoding("occurrence field total overflow or underflow")
            })
        };
        let doc_count = delta(|stats| stats.doc_count)?;
        let total_length = delta(|stats| stats.total_length)?;
        if doc_count == 0 {
            if total_length != 0 {
                return Err(VersionError::InvalidEncoding(
                    "empty occurrence field retains length",
                ));
            }
            self.replace(write.key(), None)
        } else {
            let template = write
                .value()
                .or_else(|| bytes(current.as_ref()))
                .or_else(|| bytes(base.as_ref()))
                .expect("nonempty field statistics");
            self.encode_replace(
                write.key(),
                template,
                Value::Statistics {
                    revision: revision_value,
                    doc_count,
                    total_length,
                },
            )
        }
    }
}