uqa_storage/key_value/
graph_commit.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Existing Key/Value graph addresses and path validity payloads for common commit resolution.
8
9use uqa_core::memory::BudgetedVec;
10
11use crate::mvcc::{
12    DatabaseId, GraphRecordKey, GraphRecordLayout, MergedRecordSnapshot, VersionError,
13    VersionResult,
14};
15use crate::{read_control::StorageReadControl, GraphEntityKind};
16
17use super::codec::{key_segment_length, read_segment, read_u64};
18use super::{
19    TAG_EDGE, TAG_GRAPH_LOOKUP, TAG_GRAPH_MEMBERSHIP, TAG_METADATA, TAG_NAMED_GRAPH,
20    TAG_PATH_INDEX, TAG_PATH_INDEX_DATA, TAG_VERTEX,
21};
22
23pub struct KeyValueGraphRecords;
24
25fn owns_path(
26    view: &MergedRecordSnapshot,
27    validity_key: &[u8],
28    graph: &str,
29    control: &StorageReadControl,
30) -> VersionResult<bool> {
31    if !KeyValueGraphRecords.is_validity_key(validity_key)? {
32        return Err(VersionError::InvalidEncoding("invalid path validity key"));
33    }
34    let mut key = BudgetedVec::new(control.memory());
35    key.extend_from_slice(validity_key)?;
36    key[1] = b's';
37    Ok(view
38        .get(&key, control)?
39        .as_ref()
40        .and_then(|record| record.value())
41        == Some(graph.as_bytes()))
42}
43
44fn segment(key: &mut BudgetedVec<u8>, parts: &[&[u8]]) -> VersionResult<()> {
45    let size = parts.iter().try_fold(0_usize, |length, part| {
46        length
47            .checked_add(part.len())
48            .ok_or(uqa_core::memory::MemoryError::SizeOverflow)
49    })?;
50    key.extend_from_slice(&key_segment_length(size)?.to_be_bytes())?;
51    for part in parts {
52        key.extend_from_slice(part)?;
53    }
54    Ok(())
55}
56
57fn text<'a>(key: &'a [u8], offset: &mut usize) -> VersionResult<&'a str> {
58    std::str::from_utf8(read_segment(key, offset)?)
59        .map_err(|_| VersionError::InvalidEncoding("graph key text is not UTF-8"))
60}
61
62fn complete(key: &[u8], offset: usize) -> VersionResult<()> {
63    if key.len() != offset {
64        return Err(VersionError::InvalidEncoding(
65            "graph key has trailing bytes",
66        ));
67    }
68    Ok(())
69}
70
71fn kind(name: &str) -> VersionResult<GraphEntityKind> {
72    match name {
73        "vertex" => Ok(GraphEntityKind::Vertex),
74        "edge" => Ok(GraphEntityKind::Edge),
75        _ => Err(VersionError::InvalidEncoding(
76            "unsupported graph entity kind",
77        )),
78    }
79}
80
81impl GraphRecordLayout for KeyValueGraphRecords {
82    fn key(
83        &self,
84        _: DatabaseId,
85        address: GraphRecordKey<'_>,
86        control: &StorageReadControl,
87    ) -> VersionResult<BudgetedVec<u8>> {
88        control.check()?;
89        let mut key = BudgetedVec::new(control.memory());
90        match address {
91            GraphRecordKey::Entity(kind, id) => {
92                key.push(if kind == GraphEntityKind::Vertex {
93                    TAG_VERTEX
94                } else {
95                    TAG_EDGE
96                })?;
97                key.extend_from_slice(&id.to_be_bytes())?;
98            }
99            GraphRecordKey::EntityMemberships(kind, id) => {
100                key.extend_from_slice(&[TAG_GRAPH_LOOKUP, b'm'])?;
101                segment(&mut key, &[kind.as_str().as_bytes()])?;
102                key.extend_from_slice(&id.to_be_bytes())?;
103            }
104            GraphRecordKey::GraphMemberships(graph) => {
105                key.push(TAG_GRAPH_MEMBERSHIP)?;
106                segment(&mut key, &[graph.as_bytes()])?;
107            }
108            GraphRecordKey::GraphPaths(graph) => {
109                key.extend_from_slice(&[TAG_PATH_INDEX_DATA, b'g'])?;
110                segment(&mut key, &[graph.as_bytes()])?;
111            }
112            GraphRecordKey::GraphName(graph) => {
113                key.push(TAG_NAMED_GRAPH)?;
114                segment(&mut key, &[graph.as_bytes()])?;
115            }
116            GraphRecordKey::LabelRegistry(graph) => {
117                key.push(TAG_METADATA)?;
118                segment(&mut key, &[b"graph_label_registry::", graph.as_bytes()])?;
119            }
120            GraphRecordKey::PathDefinition(index) => {
121                key.push(TAG_PATH_INDEX)?;
122                segment(&mut key, &[index.as_bytes()])?;
123            }
124            GraphRecordKey::PathValidity(index) => {
125                key.extend_from_slice(&[TAG_PATH_INDEX_DATA, b'v'])?;
126                segment(&mut key, &[index.as_bytes()])?;
127            }
128        }
129        Ok(key)
130    }
131
132    fn is_validity_key(&self, key: &[u8]) -> VersionResult<bool> {
133        if !key.starts_with(&[TAG_PATH_INDEX_DATA, b'v']) {
134            return Ok(false);
135        }
136        let mut offset = 2;
137        text(key, &mut offset)?;
138        complete(key, offset)?;
139        Ok(true)
140    }
141
142    fn membership_graph(
143        &self,
144        key: &[u8],
145        control: &StorageReadControl,
146    ) -> VersionResult<BudgetedVec<u8>> {
147        if !key.starts_with(&[TAG_GRAPH_LOOKUP, b'm']) {
148            return Err(VersionError::InvalidEncoding(
149                "invalid reverse graph membership key",
150            ));
151        }
152        let mut offset = 2;
153        kind(text(key, &mut offset)?)?;
154        read_u64(key, &mut offset)?;
155        let graph = text(key, &mut offset)?;
156        complete(key, offset)?;
157        let mut result = BudgetedVec::new(control.memory());
158        result.extend_from_slice(graph.as_bytes())?;
159        Ok(result)
160    }
161
162    fn membership_entity(
163        &self,
164        key: &[u8],
165        _: &StorageReadControl,
166    ) -> VersionResult<(GraphEntityKind, u64)> {
167        if key.first() != Some(&TAG_GRAPH_MEMBERSHIP) {
168            return Err(VersionError::InvalidEncoding(
169                "invalid graph membership key",
170            ));
171        }
172        let mut offset = 1;
173        text(key, &mut offset)?;
174        let kind = kind(text(key, &mut offset)?)?;
175        let id = read_u64(key, &mut offset)?;
176        complete(key, offset)?;
177        Ok((kind, id))
178    }
179
180    fn path_validity_key(
181        &self,
182        view: &MergedRecordSnapshot,
183        database: DatabaseId,
184        graph: &str,
185        key: &[u8],
186        value: &[u8],
187        control: &StorageReadControl,
188    ) -> VersionResult<Option<BudgetedVec<u8>>> {
189        if !key.starts_with(&[TAG_PATH_INDEX_DATA, b'g']) || !value.is_empty() {
190            return Err(VersionError::InvalidEncoding(
191                "invalid graph path directory entry",
192            ));
193        }
194        let mut offset = 2;
195        let owner = text(key, &mut offset)?;
196        let index = text(key, &mut offset)?;
197        complete(key, offset)?;
198        if owner != graph {
199            return Ok(None);
200        }
201        let key = self.key(database, GraphRecordKey::PathValidity(index), control)?;
202        Ok(owns_path(view, &key, graph, control)?.then_some(key))
203    }
204
205    fn invalidate(
206        &self,
207        _: &[u8],
208        _: &[u8],
209        _: &StorageReadControl,
210    ) -> VersionResult<Option<BudgetedVec<u8>>> {
211        Ok(None)
212    }
213
214    fn is_published(
215        &self,
216        view: &MergedRecordSnapshot,
217        key: &[u8],
218        value: &[u8],
219        graph: &str,
220        definition: &str,
221        control: &StorageReadControl,
222    ) -> VersionResult<bool> {
223        Ok(value == definition.as_bytes() && owns_path(view, key, graph, control)?)
224    }
225
226    fn definition_matches(
227        &self,
228        _: &[u8],
229        value: &[u8],
230        definition: &str,
231        _: &StorageReadControl,
232    ) -> VersionResult<bool> {
233        Ok(value == definition.as_bytes())
234    }
235}
uqa_storage/key_value/graph_commit.rs

uqa_storage/key_value/
graph_commit.rs
