uqa_storage/sqlite/catalog/
graph.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Named graph entities, membership, and snapshot replacement.
8
9use super::{
10    decode_catalog_id, encode_catalog_id, params, Catalog, EdgeRow, GraphSnapshot,
11    OptionalExtension, Result, SQLiteError,
12};
13
14impl Catalog {
15    /// Indexed, graph-scoped hydration. LEFT JOIN retains invalid memberships
16    /// so missing entities cannot silently disappear from a restored graph.
17    pub fn load_named_graph_snapshot(&self, name: &str) -> Result<Option<GraphSnapshot>> {
18        self.conn.with(|conn| {
19            let exists: bool = conn.query_row(
20                "SELECT EXISTS(SELECT 1 FROM _named_graphs WHERE name = ?1)",
21                [name], |row| row.get(0),
22            )?;
23            let mut stmt = conn.prepare_cached(
24                "SELECT m.entity_type, m.entity_id, v.vertex_id, v.label, v.properties_json, \
25                        e.edge_id, e.source_id, e.target_id, e.label, e.properties_json \
26                 FROM _graph_membership AS m \
27                 LEFT JOIN _graph_vertices AS v ON m.entity_type = 'vertex' AND v.vertex_id = m.entity_id \
28                 LEFT JOIN _graph_edges AS e ON m.entity_type = 'edge' AND e.edge_id = m.entity_id \
29                 WHERE m.graph_name = ?1 ORDER BY m.entity_type, m.entity_id",
30            )?;
31            let mut rows = stmt.query([name])?;
32            let mut snapshot = GraphSnapshot {
33                vertices: Vec::new(), edges: Vec::new(),
34                label_registry_json: conn.query_row(
35                    "SELECT value FROM _metadata WHERE key = ?1",
36                    [format!("graph_label_registry::{name}")], |row| row.get(0),
37                ).optional()?.unwrap_or_default(),
38            };
39            while let Some(row) = rows.next()? {
40                if !exists {
41                    return Err(SQLiteError::StorageBackend(format!("graph membership references unregistered graph `{name}`")));
42                }
43                let kind: String = row.get(0)?;
44                let id = decode_catalog_id("graph membership entity", row.get(1)?)?;
45                match kind.as_str() {
46                    "vertex" => {
47                        if row.get::<_, Option<i64>>(2)?.is_none() {
48                            return Err(SQLiteError::StorageBackend(format!("graph `{name}` references missing vertex {id}")));
49                        }
50                        snapshot.vertices.push(crate::GraphVertexRow {
51                            vertex_id: id, label: row.get(3)?, properties_json: row.get(4)?,
52                        });
53                    }
54                    "edge" => {
55                        if row.get::<_, Option<i64>>(5)?.is_none() {
56                            return Err(SQLiteError::StorageBackend(format!("graph `{name}` references missing edge {id}")));
57                        }
58                        snapshot.edges.push(EdgeRow {
59                            edge_id: id,
60                            source_id: decode_catalog_id("edge source vertex", row.get(6)?)?,
61                            target_id: decode_catalog_id("edge target vertex", row.get(7)?)?,
62                            label: row.get(8)?, properties_json: row.get(9)?,
63                        });
64                    }
65                    _ => return Err(SQLiteError::StorageBackend(format!("graph `{name}` has invalid membership type `{kind}`"))),
66                }
67            }
68            Ok(exists.then_some(snapshot))
69        })
70    }
71
72    /// Register the existence of a named graph in the catalog.
73    pub fn save_named_graph(&self, name: &str) -> Result<()> {
74        self.conn.with(|c| {
75            c.execute(
76                "INSERT OR IGNORE INTO _named_graphs (name) VALUES (?1)",
77                params![name],
78            )?;
79            Ok(())
80        })
81    }
82
83    /// Drop the named-graph registry row plus every membership entry
84    /// that scopes a vertex or edge to this graph. Vertex / edge rows
85    /// stay in `_graph_vertices` / `_graph_edges` until they go
86    /// orphan; call [`Catalog::purge_orphan_graph_entities`] afterwards to
87    /// collect them. The engine performs that sweep on the catalog's behalf.
88    pub fn drop_named_graph(&self, name: &str) -> Result<()> {
89        self.conn.with(|c| {
90            c.execute("DELETE FROM _named_graphs WHERE name = ?1", params![name])?;
91            c.execute(
92                "DELETE FROM _graph_membership WHERE graph_name = ?1",
93                params![name],
94            )?;
95            Ok(())
96        })
97    }
98
99    /// Return every persisted named graph in sorted order.
100    pub fn load_named_graphs(&self) -> Result<Vec<String>> {
101        self.conn.with(|c| {
102            let mut stmt = c.prepare("SELECT name FROM _named_graphs ORDER BY name")?;
103            let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
104            let mut out = Vec::new();
105            for row in rows {
106                out.push(row?);
107            }
108            Ok(out)
109        })
110    }
111
112    /// Persist a vertex by global id, label, and JSON-encoded property map.
113    pub fn save_vertex(&self, vertex_id: u64, label: &str, properties_json: &str) -> Result<()> {
114        let vertex_id = encode_catalog_id("vertex", vertex_id)?;
115        self.conn.with(|c| {
116            c.execute(
117                "INSERT OR REPLACE INTO _graph_vertices (vertex_id, label, properties_json) \
118                 VALUES (?1, ?2, ?3)",
119                params![vertex_id, label, properties_json],
120            )?;
121            Ok(())
122        })
123    }
124
125    /// Delete a vertex by global id.
126    pub fn delete_vertex(&self, vertex_id: u64) -> Result<()> {
127        let vertex_id = encode_catalog_id("vertex", vertex_id)?;
128        self.conn.with(|c| {
129            c.execute(
130                "DELETE FROM _graph_vertices WHERE vertex_id = ?1",
131                params![vertex_id],
132            )?;
133            Ok(())
134        })
135    }
136
137    /// Every vertex row sorted by id, returned as
138    /// `(vertex_id, label, properties_json)` so the caller can rebuild each
139    /// `Vertex` from typed columns and the JSON-encoded property map.
140    pub fn load_vertices(&self) -> Result<Vec<(u64, String, String)>> {
141        self.conn.with(|c| {
142            let mut stmt = c.prepare(
143                "SELECT vertex_id, label, properties_json FROM _graph_vertices ORDER BY vertex_id",
144            )?;
145            let rows = stmt.query_map([], |r| {
146                Ok((
147                    r.get::<_, i64>(0)?,
148                    r.get::<_, String>(1)?,
149                    r.get::<_, String>(2)?,
150                ))
151            })?;
152            let mut out = Vec::new();
153            for row in rows {
154                let (id, label, props) = row?;
155                out.push((decode_catalog_id("vertex", id)?, label, props));
156            }
157            Ok(out)
158        })
159    }
160
161    /// Persist an edge by global id with its source and target vertices,
162    /// label, and JSON-encoded property map.
163    pub fn save_edge(
164        &self,
165        edge_id: u64,
166        source_id: u64,
167        target_id: u64,
168        label: &str,
169        properties_json: &str,
170    ) -> Result<()> {
171        let edge_id = encode_catalog_id("edge", edge_id)?;
172        let source_id = encode_catalog_id("edge source vertex", source_id)?;
173        let target_id = encode_catalog_id("edge target vertex", target_id)?;
174        self.conn.with(|c| {
175            c.execute(
176                "INSERT OR REPLACE INTO _graph_edges \
177                    (edge_id, source_id, target_id, label, properties_json) \
178                 VALUES (?1, ?2, ?3, ?4, ?5)",
179                params![edge_id, source_id, target_id, label, properties_json],
180            )?;
181            Ok(())
182        })
183    }
184
185    /// Delete an edge by global id.
186    pub fn delete_edge(&self, edge_id: u64) -> Result<()> {
187        let edge_id = encode_catalog_id("edge", edge_id)?;
188        self.conn.with(|c| {
189            c.execute(
190                "DELETE FROM _graph_edges WHERE edge_id = ?1",
191                params![edge_id],
192            )?;
193            Ok(())
194        })
195    }
196
197    /// Return every edge row in identifier order.
198    pub fn load_edges(&self) -> Result<Vec<EdgeRow>> {
199        self.conn.with(|c| {
200            let mut stmt = c.prepare(
201                "SELECT edge_id, source_id, target_id, label, properties_json \
202                   FROM _graph_edges ORDER BY edge_id",
203            )?;
204            let rows = stmt.query_map([], |r| {
205                Ok((
206                    r.get::<_, i64>(0)?,
207                    r.get::<_, i64>(1)?,
208                    r.get::<_, i64>(2)?,
209                    r.get::<_, String>(3)?,
210                    r.get::<_, String>(4)?,
211                ))
212            })?;
213            let mut out = Vec::new();
214            for row in rows {
215                let (id, src, tgt, label, props) = row?;
216                out.push(EdgeRow {
217                    edge_id: decode_catalog_id("edge", id)?,
218                    source_id: decode_catalog_id("edge source vertex", src)?,
219                    target_id: decode_catalog_id("edge target vertex", tgt)?,
220                    label,
221                    properties_json: props,
222                });
223            }
224            Ok(out)
225        })
226    }
227
228    /// Attach `entity_id` (a vertex when `entity_type == "vertex"`, an
229    /// edge when `"edge"`) to `graph_name`. The same entity can sit in
230    /// many graphs; the row is keyed by the full triple so duplicate
231    /// attaches no-op.
232    pub fn save_graph_membership(
233        &self,
234        entity_type: &str,
235        entity_id: u64,
236        graph_name: &str,
237    ) -> Result<()> {
238        let entity_id = encode_catalog_id("graph membership entity", entity_id)?;
239        self.conn.with(|c| {
240            c.execute(
241                "INSERT OR IGNORE INTO _graph_membership \
242                    (entity_type, entity_id, graph_name) \
243                 VALUES (?1, ?2, ?3)",
244                params![entity_type, entity_id, graph_name],
245            )?;
246            Ok(())
247        })
248    }
249
250    /// Detach `entity_id` from `graph_name`.
251    pub fn delete_graph_membership(
252        &self,
253        entity_type: &str,
254        entity_id: u64,
255        graph_name: &str,
256    ) -> Result<()> {
257        let entity_id = encode_catalog_id("graph membership entity", entity_id)?;
258        self.conn.with(|c| {
259            c.execute(
260                "DELETE FROM _graph_membership \
261                  WHERE entity_type = ?1 AND entity_id = ?2 AND graph_name = ?3",
262                params![entity_type, entity_id, graph_name],
263            )?;
264            Ok(())
265        })
266    }
267
268    /// Detach every entity from `graph_name`. Used as the prelude to a
269    /// full graph drop / Cypher resync.
270    pub fn delete_graph_membership_for_graph(&self, graph_name: &str) -> Result<()> {
271        self.conn.with(|c| {
272            c.execute(
273                "DELETE FROM _graph_membership WHERE graph_name = ?1",
274                params![graph_name],
275            )?;
276            Ok(())
277        })
278    }
279
280    /// Every membership row, returned as `(entity_type, entity_id, graph_name)`.
281    pub fn load_graph_memberships(&self) -> Result<Vec<(String, u64, String)>> {
282        self.conn.with(|c| {
283            let mut stmt = c.prepare(
284                "SELECT entity_type, entity_id, graph_name FROM _graph_membership \
285                  ORDER BY graph_name, entity_type, entity_id",
286            )?;
287            let rows = stmt.query_map([], |r| {
288                Ok((
289                    r.get::<_, String>(0)?,
290                    r.get::<_, i64>(1)?,
291                    r.get::<_, String>(2)?,
292                ))
293            })?;
294            let mut out = Vec::new();
295            for row in rows {
296                let (ty, id, graph) = row?;
297                out.push((ty, decode_catalog_id("graph membership entity", id)?, graph));
298            }
299            Ok(out)
300        })
301    }
302
303    /// Drop vertex / edge rows that no membership row still references.
304    /// Run after a detach / drop to garbage-collect orphaned entities.
305    pub fn purge_orphan_graph_entities(&self) -> Result<()> {
306        self.conn.with(|c| {
307            c.execute(
308                "DELETE FROM _graph_vertices \
309                  WHERE vertex_id NOT IN ( \
310                    SELECT entity_id FROM _graph_membership WHERE entity_type = 'vertex' \
311                  )",
312                [],
313            )?;
314            c.execute(
315                "DELETE FROM _graph_edges \
316                  WHERE edge_id NOT IN ( \
317                    SELECT entity_id FROM _graph_membership WHERE entity_type = 'edge' \
318                  )",
319                [],
320            )?;
321            Ok(())
322        })
323    }
324
325    pub fn replace_named_graph(&self, graph_name: &str, snapshot: &GraphSnapshot) -> Result<()> {
326        self.conn.with_mut(|c| {
327            let tx = c.savepoint()?;
328            tx.execute(
329                "INSERT OR IGNORE INTO _named_graphs (name) VALUES (?1)",
330                params![graph_name],
331            )?;
332            tx.execute(
333                "DELETE FROM _graph_membership WHERE graph_name = ?1",
334                params![graph_name],
335            )?;
336            tx.execute(
337                "DELETE FROM _path_indexes
338                  WHERE substr(graph_name, 1, length(?1) + 2) = ?1 || '::'",
339                params![graph_name],
340            )?;
341            for vertex in &snapshot.vertices {
342                let vertex_id = encode_catalog_id("vertex", vertex.vertex_id)?;
343                tx.execute(
344                    "INSERT OR REPLACE INTO _graph_vertices
345                        (vertex_id, label, properties_json) VALUES (?1, ?2, ?3)",
346                    params![vertex_id, vertex.label, vertex.properties_json],
347                )?;
348                tx.execute(
349                    "INSERT OR IGNORE INTO _graph_membership
350                        (entity_type, entity_id, graph_name) VALUES ('vertex', ?1, ?2)",
351                    params![vertex_id, graph_name],
352                )?;
353            }
354            for edge in &snapshot.edges {
355                let edge_id = encode_catalog_id("edge", edge.edge_id)?;
356                let source_id = encode_catalog_id("edge source vertex", edge.source_id)?;
357                let target_id = encode_catalog_id("edge target vertex", edge.target_id)?;
358                tx.execute(
359                    "INSERT OR REPLACE INTO _graph_edges
360                        (edge_id, source_id, target_id, label, properties_json)
361                     VALUES (?1, ?2, ?3, ?4, ?5)",
362                    params![
363                        edge_id,
364                        source_id,
365                        target_id,
366                        edge.label,
367                        edge.properties_json
368                    ],
369                )?;
370                tx.execute(
371                    "INSERT OR IGNORE INTO _graph_membership
372                        (entity_type, entity_id, graph_name) VALUES ('edge', ?1, ?2)",
373                    params![edge_id, graph_name],
374                )?;
375            }
376            tx.execute(
377                "INSERT OR REPLACE INTO _metadata (key, value) VALUES (?1, ?2)",
378                params![
379                    format!("graph_label_registry::{graph_name}"),
380                    snapshot.label_registry_json
381                ],
382            )?;
383            Self::purge_orphan_graph_entities_on(&tx)?;
384            tx.commit()?;
385            Ok(())
386        })
387    }
388
389    pub fn drop_named_graph_data(&self, graph_name: &str) -> Result<()> {
390        self.conn.with_mut(|c| {
391            let tx = c.savepoint()?;
392            tx.execute(
393                "DELETE FROM _named_graphs WHERE name = ?1",
394                params![graph_name],
395            )?;
396            tx.execute(
397                "DELETE FROM _graph_membership WHERE graph_name = ?1",
398                params![graph_name],
399            )?;
400            tx.execute(
401                "DELETE FROM _metadata WHERE key = ?1",
402                params![format!("graph_label_registry::{graph_name}")],
403            )?;
404            tx.execute(
405                "DELETE FROM _path_indexes
406                  WHERE substr(graph_name, 1, length(?1) + 2) = ?1 || '::'",
407                params![graph_name],
408            )?;
409            Self::purge_orphan_graph_entities_on(&tx)?;
410            tx.commit()?;
411            Ok(())
412        })
413    }
414
415    pub(super) fn purge_orphan_graph_entities_on(c: &rusqlite::Connection) -> Result<()> {
416        c.execute(
417            "DELETE FROM _graph_vertices
418              WHERE vertex_id NOT IN (
419                SELECT entity_id FROM _graph_membership WHERE entity_type = 'vertex'
420              )",
421            [],
422        )?;
423        c.execute(
424            "DELETE FROM _graph_edges
425              WHERE edge_id NOT IN (
426                SELECT entity_id FROM _graph_membership WHERE entity_type = 'edge'
427              )",
428            [],
429        )?;
430        Ok(())
431    }
432}
uqa_storage/sqlite/catalog/graph.rs

uqa_storage/sqlite/catalog/
graph.rs
