uqa_storage/sqlite/catalog/
graph_access.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Storage-side graph identity, membership, and adjacency lookups.
8
9use rusqlite::{params_from_iter, types::Value as SQLValue};
10use std::fmt::Write as _;
11
12use super::{
13    decode_catalog_id, encode_catalog_id, Catalog, EdgeRow, OptionalExtension, Result, SQLiteError,
14};
15use crate::{GraphEntityFilter, GraphEntityKind, GraphVertexRow};
16
17fn entity_table(kind: GraphEntityKind) -> (&'static str, &'static str) {
18    match kind {
19        GraphEntityKind::Vertex => ("_graph_vertices", "vertex_id"),
20        GraphEntityKind::Edge => ("_graph_edges", "edge_id"),
21    }
22}
23
24struct Selection {
25    from: String,
26    id: String,
27    stored_id: String,
28    values: Vec<SQLValue>,
29}
30
31fn selection(filter: GraphEntityFilter<'_>) -> Result<Selection> {
32    filter
33        .validate()
34        .map_err(|error| SQLiteError::StorageBackend(error.to_string()))?;
35    let (table, key) = entity_table(filter.kind);
36    let mut values = Vec::new();
37    let index = match (filter.kind, filter.source, filter.target, filter.label) {
38        (GraphEntityKind::Edge, Some(_), _, _) => Some("_graph_edges_out"),
39        (GraphEntityKind::Edge, _, Some(_), _) => Some("_graph_edges_in"),
40        (GraphEntityKind::Edge, _, _, Some(_)) => Some("_graph_edges_label"),
41        (GraphEntityKind::Vertex, _, _, Some(_)) => Some("_graph_vertices_label"),
42        _ => None,
43    };
44    let (mut from, id) = if let Some(index) = index {
45        // Adjacency and label reads start at the selective physical index,
46        // never at the potentially much larger graph membership partition.
47        (
48            format!("FROM {table} AS e INDEXED BY {index} WHERE 1 = 1"),
49            format!("e.{key}"),
50        )
51    } else if let Some(graph) = filter.graph {
52        values.push(SQLValue::Text(graph.to_owned()));
53        values.push(SQLValue::Text(filter.kind.as_str().to_owned()));
54        (format!("FROM _graph_membership AS m LEFT JOIN {table} AS e ON e.{key} = m.entity_id WHERE m.graph_name = ? AND m.entity_type = ?"), "m.entity_id".to_owned())
55    } else {
56        (format!("FROM {table} AS e WHERE 1 = 1"), format!("e.{key}"))
57    };
58    if let Some(label) = filter.label {
59        from.push_str(" AND e.label = ?");
60        values.push(SQLValue::Text(label.to_owned()));
61    }
62    if let Some(source) = filter.source {
63        from.push_str(" AND e.source_id = ?");
64        values.push(SQLValue::Integer(encode_catalog_id("edge source", source)?));
65    }
66    if let Some(target) = filter.target {
67        from.push_str(" AND e.target_id = ?");
68        values.push(SQLValue::Integer(encode_catalog_id("edge target", target)?));
69    }
70    if index.is_some() {
71        if let Some(graph) = filter.graph {
72            write!(from, " AND EXISTS (SELECT 1 FROM _graph_membership AS m WHERE m.entity_type = ? AND m.entity_id = e.{key} AND m.graph_name = ?)").expect("write graph membership predicate");
73            values.push(SQLValue::Text(filter.kind.as_str().to_owned()));
74            values.push(SQLValue::Text(graph.to_owned()));
75        }
76    }
77    Ok(Selection {
78        from,
79        id,
80        stored_id: format!("e.{key}"),
81        values,
82    })
83}
84
85impl Catalog {
86    pub fn named_graph_exists(&self, name: &str) -> Result<bool> {
87        self.conn.with(|conn| {
88            Ok(conn.query_row(
89                "SELECT EXISTS(SELECT 1 FROM _named_graphs WHERE name = ?1)",
90                [name],
91                |row| row.get(0),
92            )?)
93        })
94    }
95
96    pub fn graph_vertex(&self, id: u64) -> Result<Option<GraphVertexRow>> {
97        let encoded = encode_catalog_id("vertex", id)?;
98        self.conn.with(|conn| {
99            Ok(conn
100                .query_row(
101                    "SELECT label, properties_json FROM _graph_vertices WHERE vertex_id = ?1",
102                    [encoded],
103                    |row| {
104                        Ok(GraphVertexRow {
105                            vertex_id: id,
106                            label: row.get(0)?,
107                            properties_json: row.get(1)?,
108                        })
109                    },
110                )
111                .optional()?)
112        })
113    }
114
115    pub fn graph_edge(&self, id: u64) -> Result<Option<EdgeRow>> {
116        let encoded = encode_catalog_id("edge", id)?;
117        self.conn.with(|conn| {
118            let row = conn.query_row(
119                "SELECT source_id, target_id, label, properties_json FROM _graph_edges WHERE edge_id = ?1",
120                [encoded],
121                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?)),
122            ).optional()?;
123            row.map(|(source, target, label, properties_json)| Ok(EdgeRow {
124                edge_id: id,
125                source_id: decode_catalog_id("edge source", source)?,
126                target_id: decode_catalog_id("edge target", target)?,
127                label, properties_json,
128            })).transpose()
129        })
130    }
131
132    pub fn graph_entity_ids(
133        &self,
134        filter: GraphEntityFilter<'_>,
135        after: Option<u64>,
136        limit: usize,
137    ) -> Result<Vec<u64>> {
138        crate::catalog::validate_graph_page(limit)
139            .map_err(|error| SQLiteError::StorageBackend(error.to_string()))?;
140        let mut query = selection(filter)?;
141        if let Some(after) = after {
142            write!(query.from, " AND {} > ?", query.id).expect("write graph cursor predicate");
143            query.values.push(SQLValue::Integer(encode_catalog_id(
144                "graph scan cursor",
145                after,
146            )?));
147        }
148        query.values.push(SQLValue::Integer(
149            i64::try_from(limit).map_err(|error| SQLiteError::StorageBackend(error.to_string()))?,
150        ));
151        let sql = format!(
152            "SELECT {}, {} {} ORDER BY {} LIMIT ?",
153            query.id, query.stored_id, query.from, query.id
154        );
155        self.conn.with(|conn| {
156            let mut statement = conn.prepare_cached(&sql)?;
157            let mut rows = statement.query(params_from_iter(query.values))?;
158            let mut ids = Vec::new();
159            while let Some(row) = rows.next()? {
160                let id = decode_catalog_id("graph entity", row.get(0)?)?;
161                if row.get::<_, Option<i64>>(1)?.is_none() {
162                    return Err(SQLiteError::StorageBackend(format!(
163                        "graph {:?} references missing {} {id}",
164                        filter.graph,
165                        filter.kind.as_str()
166                    )));
167                }
168                ids.push(id);
169            }
170            Ok(ids)
171        })
172    }
173
174    pub fn graph_entity_count(&self, filter: GraphEntityFilter<'_>) -> Result<u64> {
175        let query = selection(filter)?;
176        let sql = format!("SELECT count(*), count({}) {}", query.stored_id, query.from);
177        self.conn.with(|conn| {
178            let (count, stored): (i64, i64) =
179                conn.query_row(&sql, params_from_iter(query.values), |row| {
180                    Ok((row.get(0)?, row.get(1)?))
181                })?;
182            if count != stored {
183                return Err(SQLiteError::StorageBackend(format!(
184                    "graph {:?} references missing {} records",
185                    filter.graph,
186                    filter.kind.as_str()
187                )));
188            }
189            u64::try_from(count).map_err(|error| SQLiteError::StorageBackend(error.to_string()))
190        })
191    }
192
193    pub fn graph_entity_max_id(&self, kind: GraphEntityKind) -> Result<Option<u64>> {
194        let (table, key) = entity_table(kind);
195        self.conn.with(|conn| {
196            let id: Option<i64> =
197                conn.query_row(&format!("SELECT max({key}) FROM {table}"), [], |row| {
198                    row.get(0)
199                })?;
200            id.map(|id| decode_catalog_id("graph entity", id))
201                .transpose()
202        })
203    }
204
205    pub fn graph_entity_memberships(&self, kind: GraphEntityKind, id: u64) -> Result<Vec<String>> {
206        let encoded = encode_catalog_id("graph entity", id)?;
207        self.conn.with(|conn| {
208            let mut statement = conn.prepare_cached("SELECT graph_name FROM _graph_membership WHERE entity_type = ?1 AND entity_id = ?2 ORDER BY graph_name")?;
209            let rows = statement.query_map(rusqlite::params![kind.as_str(), encoded], |row| row.get(0))?;
210            Ok(rows.collect::<std::result::Result<_, _>>()?)
211        })
212    }
213
214    pub fn graph_has_membership(
215        &self,
216        kind: GraphEntityKind,
217        id: u64,
218        graph: &str,
219    ) -> Result<bool> {
220        let encoded = encode_catalog_id("graph entity", id)?;
221        self.conn.with(|conn| Ok(conn.query_row(
222            "SELECT EXISTS(SELECT 1 FROM _graph_membership WHERE entity_type = ?1 AND entity_id = ?2 AND graph_name = ?3)",
223            rusqlite::params![kind.as_str(), encoded, graph], |row| row.get(0),
224        )?))
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231
232    #[test]
233    fn adjacency_and_label_pages_start_at_the_selective_entity_index() {
234        let connection = crate::ManagedConnection::open_in_memory().unwrap();
235        Catalog::open(connection.clone()).unwrap();
236        for (kind, source, target, label, expected) in [
237            (
238                GraphEntityKind::Edge,
239                Some(17),
240                None,
241                Some("LINK"),
242                "_graph_edges_out",
243            ),
244            (
245                GraphEntityKind::Edge,
246                Some(17),
247                None,
248                None,
249                "_graph_edges_out",
250            ),
251            (
252                GraphEntityKind::Edge,
253                None,
254                Some(17),
255                Some("LINK"),
256                "_graph_edges_in",
257            ),
258            (
259                GraphEntityKind::Edge,
260                None,
261                None,
262                Some("LINK"),
263                "_graph_edges_label",
264            ),
265            (
266                GraphEntityKind::Vertex,
267                None,
268                None,
269                Some("Item"),
270                "_graph_vertices_label",
271            ),
272        ] {
273            let query = selection(GraphEntityFilter {
274                kind,
275                graph: Some("test_graph"),
276                source,
277                target,
278                label,
279            })
280            .unwrap();
281            let sql = format!(
282                "EXPLAIN QUERY PLAN SELECT {} {} ORDER BY {} LIMIT 256",
283                query.id, query.from, query.id
284            );
285            let plan: Vec<String> = connection
286                .with(|conn| {
287                    let mut statement = conn.prepare(&sql)?;
288                    let rows =
289                        statement.query_map(params_from_iter(query.values), |row| row.get(3))?;
290                    Ok(rows.collect::<std::result::Result<_, _>>()?)
291                })
292                .unwrap();
293            assert!(
294                plan.iter()
295                    .any(|step| step.contains("SEARCH e USING") && step.contains(expected)),
296                "{plan:?}"
297            );
298            assert!(!plan.iter().any(|step| step.contains("SCAN m")), "{plan:?}");
299        }
300    }
301}
uqa_storage/sqlite/catalog/graph_access.rs

uqa_storage/sqlite/catalog/
graph_access.rs
