uqa-storage 0.5.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

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

//! Commit preconditions share definition revisions without manufacturing writes to those definitions.

use std::sync::Arc;

use sha2::{Digest, Sha256};
use uqa_core::memory::BudgetedVec;
use uqa_core::CancellationToken;

use crate::mvcc::key::RecordKey;
use crate::mvcc::{CommitSequence, VersionError, VersionResult};
use crate::read_control::StorageReadControl;

use super::{hash_bytes, PreparedRecordCommit};

#[derive(Clone)]
pub(in crate::mvcc) struct RecordRequirement {
    pub(in crate::mvcc) key: RecordKey,
    pub(in crate::mvcc) expected: Option<CommitSequence>,
}

impl PreparedRecordCommit {
    pub(in crate::mvcc) fn with_requirements(
        mut self,
        requirements: &[RecordRequirement],
        control: &StorageReadControl,
    ) -> VersionResult<Self> {
        if requirements.is_empty() {
            return Ok(self);
        }
        let mut owned = BudgetedVec::new(control.memory());
        owned.reserve(requirements.len())?;
        let mut digest = Sha256::new();
        digest.update(b"UQA prepared record requirements 1");
        digest.update(self.fingerprint);
        digest.update((requirements.len() as u64).to_be_bytes());
        for requirement in requirements {
            control.cancellation().check()?;
            digest.update((requirement.key.bytes().len() as u64).to_be_bytes());
            hash_bytes(&mut digest, requirement.key.bytes(), control)?;
            digest.update([u8::from(requirement.expected.is_some())]);
            digest.update(
                requirement
                    .expected
                    .map_or(0, CommitSequence::as_u64)
                    .to_be_bytes(),
            );
            owned.push(requirement.clone())?;
        }
        self.fingerprint = digest.finalize().into();
        self.requirements = Some(Arc::new(owned));
        Ok(self)
    }

    pub(crate) fn has_requirements(&self) -> bool {
        self.requirements.is_some()
    }

    /// Keys validated without being written. Providers must charge any temporary driver bindings for these keys as well as replacement records.
    pub fn required_keys(&self) -> impl Iterator<Item = &[u8]> {
        self.requirements.iter().flat_map(|requirements| {
            requirements
                .iter()
                .map(|requirement| requirement.key.bytes())
        })
    }

    pub(crate) fn validate_requirements(
        &self,
        cancellation: &CancellationToken,
        mut head_revision: impl FnMut(&[u8]) -> VersionResult<Option<CommitSequence>>,
    ) -> VersionResult<()> {
        cancellation.check()?;
        if let Some(requirements) = &self.requirements {
            for (index, requirement) in requirements.iter().enumerate() {
                cancellation.check()?;
                let actual = head_revision(requirement.key.bytes())?;
                if requirement.expected != actual {
                    return Err(VersionError::ReadConflict {
                        dependency: index,
                        expected: requirement.expected,
                        actual,
                    });
                }
            }
        }
        Ok(())
    }
}