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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

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

//! Materialize one auxiliary publication slot against fresh committed state without replaying evaluated user writes.

use sha2::{Digest, Sha256};
use std::sync::Arc;
use uqa_core::memory::BudgetedVec;

use super::{
    key::RecordKey, CommittedRecordSnapshot, PreparedRecordCommit, PreparedRecordWrite,
    SharedRecordValue, VersionError, VersionResult,
};
use crate::{
    notifications::{NotificationPublication, NotificationPublicationView},
    read_control::StorageReadControl,
};

pub const NOTIFICATION_PUBLICATION_KEY: &str = "_uqa_notification_publication";

/// Provider-owned metadata addressing and lossless row encoding. Native adapters must use their stable data namespace, independent of restored transaction history.
pub trait NotificationRecordLayout: Send + Sync {
    fn key(&self, control: &StorageReadControl) -> VersionResult<BudgetedVec<u8>>;
    fn encode(
        &self,
        publication: &NotificationPublication,
        control: &StorageReadControl,
    ) -> VersionResult<SharedRecordValue>;
    fn decode<'a>(
        &self,
        key: &[u8],
        record: &'a [u8],
        control: &StorageReadControl,
    ) -> VersionResult<NotificationPublicationView<'a>>;
}

pub(super) struct NotificationEffect {
    key: RecordKey,
    record: Option<SharedRecordValue>,
    publication: [u8; 32],
}

impl NotificationEffect {
    pub(super) fn publish(
        publication: &NotificationPublication,
        layout: &dyn NotificationRecordLayout,
        control: &StorageReadControl,
    ) -> VersionResult<Arc<Self>> {
        control.check()?;
        Ok(Arc::new(Self {
            key: RecordKey::new(&layout.key(control)?, control.memory())?,
            record: Some(layout.encode(publication, control)?),
            publication: publication.fingerprint(),
        }))
    }

    pub(super) fn acknowledge(
        publication: [u8; 32],
        layout: &dyn NotificationRecordLayout,
        control: &StorageReadControl,
    ) -> VersionResult<Arc<Self>> {
        control.check()?;
        Ok(Arc::new(Self {
            key: RecordKey::new(&layout.key(control)?, control.memory())?,
            record: None,
            publication,
        }))
    }

    pub(super) fn same_publication(&self, other: &Self) -> bool {
        self.key.bytes() == other.key.bytes()
            && self.record.is_some() == other.record.is_some()
            && self.publication == other.publication
    }

    pub(super) fn seal(&self, original: super::CommitFingerprint) -> super::CommitFingerprint {
        let mut digest = Sha256::new();
        digest.update(b"UQA prepared notification effect 1");
        digest.update(original);
        digest.update((self.key.bytes().len() as u64).to_be_bytes());
        digest.update(self.key.bytes());
        digest.update([u8::from(self.record.is_some())]);
        digest.update(self.publication);
        digest.finalize().into()
    }
}

pub(super) fn resolve(
    original: &PreparedRecordCommit,
    effect: &NotificationEffect,
    current: &dyn CommittedRecordSnapshot,
    layout: &dyn NotificationRecordLayout,
    control: &StorageReadControl,
) -> VersionResult<PreparedRecordCommit> {
    control.check()?;
    let key = effect.key.bytes();
    if original.records().iter().any(|record| record.key() == key) {
        return Err(VersionError::InvalidEncoding(
            "notification publication conflicts with an ordinary record write",
        ));
    }
    let latest = current.get(key, control)?;
    let pending = latest
        .as_ref()
        .and_then(super::RecordVersion::value)
        .map(|value| layout.decode(key, value, control))
        .transpose()?;
    let replace = if effect.record.is_some() {
        if pending.is_some() {
            return Err(VersionError::InvalidEncoding(
                "committed notification publication must be acknowledged before replacement",
            ));
        }
        true
    } else {
        pending.is_some_and(|publication| publication.fingerprint() == effect.publication)
    };
    let mut writes = BudgetedVec::new(control.memory());
    writes.reserve(
        original
            .records()
            .len()
            .checked_add(usize::from(replace))
            .ok_or(uqa_core::memory::MemoryError::SizeOverflow)?,
    )?;
    for write in original.records() {
        control.check()?;
        writes.push(write.clone())?;
    }
    if replace {
        writes.push(PreparedRecordWrite::from_shared(
            effect.key.clone(),
            latest.as_ref().map(super::RecordVersion::sequence),
            effect.record.clone(),
        ))?;
    }
    Ok(PreparedRecordCommit::from_unique_owned(writes, control)?
        .resolved(original, current.sequence()))
}