uqa_storage/sqlite/catalog/
path_index_data.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Bounded physical path-index records; no reachability map is loaded at open.
8
9use super::{decode_catalog_id, encode_catalog_id, Catalog, Result, SQLiteError};
10use crate::MAX_GRAPH_ID_PAGE;
11use rusqlite::{params, params_from_iter, types::Value as SQLValue};
12
13impl Catalog {
14    pub fn clear_path_index_data(&self, index: &str) -> Result<()> {
15        self.conn.with_mut(|conn| {
16            let checkpoint = conn.savepoint()?;
17            checkpoint.execute(
18                "DELETE FROM _graph_path_pairs WHERE index_key = ?1",
19                [index],
20            )?;
21            checkpoint.execute(
22                "DELETE FROM _graph_path_index_state WHERE index_key = ?1",
23                [index],
24            )?;
25            checkpoint.commit()?;
26            Ok(())
27        })
28    }
29    pub fn save_path_index_pairs(
30        &self,
31        index: &str,
32        sequence: &str,
33        pairs: &[(u64, u64)],
34    ) -> Result<()> {
35        if pairs.len() > MAX_GRAPH_ID_PAGE {
36            return Err(SQLiteError::StorageBackend(
37                "path index batch exceeds the bounded page size".into(),
38            ));
39        }
40        let pairs = pairs
41            .iter()
42            .map(|&(source, target)| {
43                Ok((
44                    encode_catalog_id("path source", source)?,
45                    encode_catalog_id("path target", target)?,
46                ))
47            })
48            .collect::<Result<Vec<_>>>()?;
49        self.conn.with_mut(|conn| {
50            let checkpoint = conn.savepoint()?;
51            {
52                let mut statement = checkpoint.prepare_cached("INSERT INTO _graph_path_pairs(index_key, sequence_key, source_id, target_id) VALUES (?1, ?2, ?3, ?4) ON CONFLICT DO NOTHING")?;
53                for (source, target) in pairs { statement.execute(params![index, sequence, source, target])?; }
54            }
55            checkpoint.commit()?;
56            Ok(())
57        })
58    }
59    pub fn finish_path_index_data(&self, index: &str, graph: &str, definition: &str) -> Result<()> {
60        self.conn.with(|conn| {
61            let written = conn.execute("INSERT INTO _graph_path_index_state(index_key, graph_name, definition_json, valid) SELECT ?1, ?2, ?3, 1 WHERE EXISTS(SELECT 1 FROM _path_indexes WHERE graph_name = ?1 AND label_sequences = ?3) ON CONFLICT(index_key) DO UPDATE SET graph_name = excluded.graph_name, definition_json = excluded.definition_json, valid = 1", params![index, graph, definition])?;
62            if written == 0 { return Err(SQLiteError::StorageBackend(format!("path index {index:?} definition changed during build"))); }
63            Ok(())
64        })
65    }
66    pub fn path_index_data_is_current(&self, index: &str, definition: &str) -> Result<bool> {
67        self.conn.with(|conn| Ok(conn.query_row("SELECT EXISTS(SELECT 1 FROM _graph_path_index_state WHERE index_key = ?1 AND definition_json = ?2 AND valid = 1)", params![index, definition], |row| row.get(0))?))
68    }
69    pub fn path_index_pairs(
70        &self,
71        index: &str,
72        sequence: &str,
73        after: Option<(u64, u64)>,
74        limit: usize,
75    ) -> Result<Vec<(u64, u64)>> {
76        crate::catalog::validate_graph_page(limit)
77            .map_err(|error| SQLiteError::StorageBackend(error.to_string()))?;
78        let mut values = vec![
79            SQLValue::Text(index.to_owned()),
80            SQLValue::Text(sequence.to_owned()),
81        ];
82        let mut sql = "SELECT source_id, target_id FROM _graph_path_pairs WHERE index_key = ? AND sequence_key = ?".to_owned();
83        if let Some((source, target)) = after {
84            sql.push_str(" AND (source_id, target_id) > (?, ?)");
85            values.push(SQLValue::Integer(encode_catalog_id("path source", source)?));
86            values.push(SQLValue::Integer(encode_catalog_id("path target", target)?));
87        }
88        sql.push_str(" ORDER BY source_id, target_id LIMIT ?");
89        values.push(SQLValue::Integer(
90            i64::try_from(limit).map_err(|error| SQLiteError::StorageBackend(error.to_string()))?,
91        ));
92        self.conn.with(|conn| {
93            let mut statement = conn.prepare_cached(&sql)?;
94            let mut rows = statement.query(params_from_iter(values))?;
95            let mut pairs = Vec::new();
96            while let Some(row) = rows.next()? {
97                pairs.push((
98                    decode_catalog_id("path source", row.get(0)?)?,
99                    decode_catalog_id("path target", row.get(1)?)?,
100                ));
101            }
102            Ok(pairs)
103        })
104    }
105}
uqa_storage/sqlite/catalog/path_index_data.rs

uqa_storage/sqlite/catalog/
path_index_data.rs
