uqa_storage/sqlite/catalog/
foreign_indexes.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Foreign servers, foreign tables, catalog indexes, and path indexes.
8
9use super::{
10    params, Catalog, CatalogIndexRow, ForeignTableRow, RelationIdentity, RelationKind, Result,
11    SQLiteError, TableAclEntry,
12};
13
14impl Catalog {
15    // -- Foreign servers ---------------------------------------------------
16
17    pub fn save_foreign_server(
18        &self,
19        name: &str,
20        fdw_type: &str,
21        options_json: &str,
22    ) -> Result<()> {
23        self.conn.with(|c| {
24            c.execute(
25                "INSERT OR REPLACE INTO _foreign_servers (name, fdw_type, options) \
26                 VALUES (?1, ?2, ?3)",
27                params![name, fdw_type, options_json],
28            )?;
29            Ok(())
30        })
31    }
32
33    pub fn drop_foreign_server(&self, name: &str) -> Result<()> {
34        self.conn.with(|c| {
35            c.execute(
36                "DELETE FROM _foreign_servers WHERE name = ?1",
37                params![name],
38            )?;
39            Ok(())
40        })
41    }
42
43    pub fn load_foreign_servers(&self) -> Result<Vec<(String, String, String)>> {
44        self.conn.with(|c| {
45            let mut stmt =
46                c.prepare("SELECT name, fdw_type, options FROM _foreign_servers ORDER BY name")?;
47            let rows = stmt.query_map([], |r| {
48                Ok((
49                    r.get::<_, String>(0)?,
50                    r.get::<_, String>(1)?,
51                    r.get::<_, String>(2)?,
52                ))
53            })?;
54            let mut out = Vec::new();
55            for row in rows {
56                out.push(row?);
57            }
58            Ok(out)
59        })
60    }
61
62    // -- Foreign tables ----------------------------------------------------
63
64    pub fn save_foreign_table(&self, row: &ForeignTableRow) -> Result<()> {
65        self.conn.with_mut(|c| {
66            let tx = c.savepoint()?;
67            Self::claim_relation(&tx, &row.relation, RelationKind::ForeignTable)?;
68            let acl_json = row.acl.as_deref().map(serde_json::to_string).transpose()?;
69            let column_acls_json = serde_json::to_string(&row.column_acls)?;
70            tx.execute(
71                "INSERT OR REPLACE INTO _foreign_tables \
72                    (schema_name, relation_name, kind, role_owner, acl_json, column_acls_json, server_name, columns_json, options) \
73                 VALUES (?1, ?2, 'foreign_table', ?3, ?4, ?5, ?6, ?7, ?8)",
74                params![
75                    row.relation.schema,
76                    row.relation.name,
77                    row.role_owner,
78                    acl_json,
79                    column_acls_json,
80                    row.server_name,
81                    row.columns_json,
82                    row.options_json
83                ],
84            )?;
85            tx.commit()?;
86            Ok(())
87        })
88    }
89
90    pub fn update_foreign_table_security(
91        &self,
92        relation: &RelationIdentity,
93        role_owner: &str,
94        acl: Option<&[TableAclEntry]>,
95        column_acls: &std::collections::BTreeMap<String, Vec<TableAclEntry>>,
96    ) -> Result<bool> {
97        self.conn.with_mut(|connection| {
98            let acl_json = acl.map(serde_json::to_string).transpose()?;
99            let column_acls_json = serde_json::to_string(column_acls)?;
100            Ok(connection.execute(
101                "UPDATE _foreign_tables
102                    SET role_owner = ?3, acl_json = ?4, column_acls_json = ?5
103                  WHERE schema_name = ?1 AND relation_name = ?2",
104                params![
105                    relation.schema,
106                    relation.name,
107                    role_owner,
108                    acl_json,
109                    column_acls_json
110                ],
111            )? != 0)
112        })
113    }
114
115    pub fn rename_foreign_table(
116        &self,
117        from: &RelationIdentity,
118        to: &RelationIdentity,
119    ) -> Result<bool> {
120        if from.schema != to.schema {
121            return Err(SQLiteError::StorageBackend(
122                "moving a foreign table between schemas is not supported by the catalog".into(),
123            ));
124        }
125        self.conn.with_mut(|connection| {
126            let source_exists = connection.query_row(
127                "SELECT EXISTS(SELECT 1 FROM _foreign_tables WHERE schema_name = ?1 AND relation_name = ?2)",
128                params![from.schema, from.name],
129                |row| row.get::<_, bool>(0),
130            )?;
131            if from == to || !source_exists {
132                return Ok(source_exists);
133            }
134            let target_exists = connection.query_row(
135                "SELECT EXISTS(SELECT 1 FROM _relations WHERE schema_name = ?1 AND relation_name = ?2)",
136                params![to.schema, to.name],
137                |row| row.get::<_, bool>(0),
138            )?;
139            if target_exists {
140                return Err(SQLiteError::StorageBackend(format!(
141                    "relation `{}` already exists",
142                    to.qualified_name()
143                )));
144            }
145            let tx = connection.savepoint()?;
146            Self::claim_relation(&tx, to, RelationKind::ForeignTable)?;
147            let updated = tx.execute(
148                "UPDATE _foreign_tables SET schema_name = ?3, relation_name = ?4 WHERE schema_name = ?1 AND relation_name = ?2",
149                params![from.schema, from.name, to.schema, to.name],
150            )?;
151            if updated != 1 {
152                return Err(SQLiteError::StorageBackend(format!(
153                    "foreign table `{}` disappeared during rename",
154                    from.qualified_name()
155                )));
156            }
157            Self::release_relation(&tx, from, RelationKind::ForeignTable)?;
158            tx.commit()?;
159            Ok(true)
160        })
161    }
162
163    pub fn drop_foreign_table(&self, relation: &RelationIdentity) -> Result<()> {
164        self.conn.with_mut(|c| {
165            let tx = c.savepoint()?;
166            let removed = tx.execute(
167                "DELETE FROM _foreign_tables
168                  WHERE schema_name = ?1 AND relation_name = ?2",
169                params![relation.schema, relation.name],
170            )? != 0;
171            if removed {
172                Self::release_relation(&tx, relation, RelationKind::ForeignTable)?;
173            }
174            tx.commit()?;
175            Ok(())
176        })
177    }
178
179    pub fn load_foreign_tables(&self) -> Result<Vec<ForeignTableRow>> {
180        self.conn.with(|c| {
181            let mut stmt = c.prepare(
182                "SELECT schema_name, relation_name, role_owner, acl_json, column_acls_json, server_name, columns_json, options
183                   FROM _foreign_tables ORDER BY schema_name, relation_name",
184            )?;
185            let rows = stmt.query_map([], |r| {
186                Ok((
187                    r.get::<_, String>(0)?,
188                    r.get::<_, String>(1)?,
189                    r.get::<_, String>(2)?,
190                    r.get::<_, Option<String>>(3)?,
191                    r.get::<_, String>(4)?,
192                    r.get::<_, String>(5)?,
193                    r.get::<_, String>(6)?,
194                    r.get::<_, String>(7)?,
195                ))
196            })?;
197            let mut out = Vec::new();
198            for row in rows {
199                let (schema, name, owner, acl_json, column_acls_json, server, cols, opts) = row?;
200                out.push(ForeignTableRow {
201                    relation: RelationIdentity::new(schema, name),
202                    role_owner: owner,
203                    acl: acl_json
204                        .as_deref()
205                        .map(serde_json::from_str)
206                        .transpose()?,
207                    column_acls: serde_json::from_str(&column_acls_json)?,
208                    server_name: server,
209                    columns_json: cols,
210                    options_json: opts,
211                });
212            }
213            Ok(out)
214        })
215    }
216
217    // -- Catalog indexes (CREATE INDEX state) ------------------------------
218
219    pub fn save_catalog_index(
220        &self,
221        relation: &RelationIdentity,
222        index_type: &str,
223        table_name: &str,
224        columns_json: &str,
225        parameters_json: &str,
226    ) -> Result<()> {
227        self.save_catalog_index_row(&CatalogIndexRow {
228            relation: relation.clone(),
229            index_type: index_type.to_string(),
230            table_name: table_name.to_string(),
231            columns_json: columns_json.to_string(),
232            parameters_json: parameters_json.to_string(),
233            definition_json: None,
234        })
235    }
236
237    pub fn save_catalog_index_row(&self, index: &CatalogIndexRow) -> Result<()> {
238        let CatalogIndexRow {
239            relation,
240            index_type,
241            table_name,
242            columns_json,
243            parameters_json,
244            definition_json,
245        } = index;
246        let table =
247            RelationIdentity::from_legacy_name(table_name).map_err(SQLiteError::StorageBackend)?;
248        if relation.schema != table.schema {
249            return Err(SQLiteError::StorageBackend(format!(
250                "catalog index `{}` cannot belong to a different schema than table `{}`",
251                relation.qualified_name(),
252                table.qualified_name()
253            )));
254        }
255        self.conn.with_mut(|c| {
256            let tx = c.savepoint()?;
257            Self::claim_relation(&tx, relation, RelationKind::Index)?;
258            tx.execute(
259                "INSERT INTO _catalog_indexes
260                    (schema_name, relation_name, kind, index_type, table_schema_name,
261                     table_relation_name, columns, parameters, definition)
262                 VALUES (?1, ?2, 'index', ?3, ?4, ?5, ?6, ?7, ?8)
263                 ON CONFLICT(schema_name, relation_name) DO UPDATE SET
264                     index_type = excluded.index_type,
265                     table_schema_name = excluded.table_schema_name,
266                     table_relation_name = excluded.table_relation_name,
267                     columns = excluded.columns,
268                     parameters = excluded.parameters,
269                     definition = excluded.definition",
270                params![
271                    relation.schema,
272                    relation.name,
273                    index_type,
274                    table.schema,
275                    table.name,
276                    columns_json,
277                    parameters_json,
278                    definition_json
279                ],
280            )?;
281            tx.commit()?;
282            Ok(())
283        })
284    }
285
286    pub fn drop_catalog_index(&self, relation: &RelationIdentity) -> Result<()> {
287        self.conn.with_mut(|c| {
288            let tx = c.savepoint()?;
289            tx.execute(
290                "DELETE FROM _catalog_indexes
291                  WHERE schema_name = ?1 AND relation_name = ?2",
292                params![relation.schema, relation.name],
293            )?;
294            Self::release_relation(&tx, relation, RelationKind::Index)?;
295            tx.commit()?;
296            Ok(())
297        })
298    }
299
300    pub fn drop_catalog_indexes_for_table(&self, table_name: &str) -> Result<()> {
301        let table =
302            RelationIdentity::from_legacy_name(table_name).map_err(SQLiteError::StorageBackend)?;
303        self.conn.with_mut(|c| {
304            let tx = c.savepoint()?;
305            Self::drop_catalog_index_rows_for_table(&tx, &table)?;
306            tx.commit()?;
307            Ok(())
308        })
309    }
310
311    pub(in crate::sqlite::catalog) fn drop_catalog_index_rows_for_table(
312        conn: &rusqlite::Connection,
313        table: &RelationIdentity,
314    ) -> Result<()> {
315        let indexes = {
316            let mut statement = conn.prepare(
317                "SELECT schema_name, relation_name
318                   FROM _catalog_indexes
319                  WHERE table_schema_name = ?1 AND table_relation_name = ?2",
320            )?;
321            let indexes = statement
322                .query_map(params![table.schema, table.name], |row| {
323                    Ok(RelationIdentity::new(
324                        row.get::<_, String>(0)?,
325                        row.get::<_, String>(1)?,
326                    ))
327                })?
328                .collect::<rusqlite::Result<Vec<_>>>()?;
329            indexes
330        };
331        for index in indexes {
332            conn.execute(
333                "DELETE FROM _catalog_indexes
334                  WHERE schema_name = ?1 AND relation_name = ?2",
335                params![index.schema, index.name],
336            )?;
337            Self::release_relation(conn, &index, RelationKind::Index)?;
338        }
339        Ok(())
340    }
341
342    pub fn load_catalog_indexes(&self) -> Result<Vec<CatalogIndexRow>> {
343        self.conn.with(|c| {
344            let mut stmt = c.prepare(
345                "SELECT schema_name, relation_name, index_type,
346                        table_schema_name, table_relation_name, columns, parameters, definition
347                   FROM _catalog_indexes ORDER BY schema_name, relation_name",
348            )?;
349            let rows = stmt.query_map([], |r| {
350                Ok((
351                    r.get::<_, String>(0)?,
352                    r.get::<_, String>(1)?,
353                    r.get::<_, String>(2)?,
354                    r.get::<_, String>(3)?,
355                    r.get::<_, String>(4)?,
356                    r.get::<_, String>(5)?,
357                    r.get::<_, String>(6)?,
358                    r.get::<_, Option<String>>(7)?,
359                ))
360            })?;
361            let mut out = Vec::new();
362            for row in rows {
363                let (schema, name, ty, table_schema, table_name, cols, params_json, definition) =
364                    row?;
365                out.push(CatalogIndexRow {
366                    relation: RelationIdentity::new(schema, name),
367                    index_type: ty,
368                    table_name: RelationIdentity::new(table_schema, table_name).qualified_name(),
369                    columns_json: cols,
370                    parameters_json: params_json,
371                    definition_json: definition,
372                });
373            }
374            Ok(out)
375        })
376    }
377
378    // -- Path indexes ------------------------------------------------------
379
380    pub fn save_path_index(&self, graph_name: &str, label_sequences_json: &str) -> Result<()> {
381        self.conn.with(|c| {
382            c.execute(
383                "INSERT OR REPLACE INTO _path_indexes (graph_name, label_sequences) \
384                 VALUES (?1, ?2)",
385                params![graph_name, label_sequences_json],
386            )?;
387            Ok(())
388        })
389    }
390
391    pub fn drop_path_index(&self, graph_name: &str) -> Result<()> {
392        self.conn.with(|c| {
393            c.execute(
394                "DELETE FROM _path_indexes WHERE graph_name = ?1",
395                params![graph_name],
396            )?;
397            Ok(())
398        })
399    }
400
401    /// `(graph_name, label_sequences_json)` for every persisted path index.
402    pub fn load_path_indexes(&self) -> Result<Vec<(String, String)>> {
403        self.conn.with(|c| {
404            let mut stmt = c.prepare(
405                "SELECT graph_name, label_sequences FROM _path_indexes ORDER BY graph_name",
406            )?;
407            let rows =
408                stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
409            let mut out = Vec::new();
410            for row in rows {
411                out.push(row?);
412            }
413            Ok(out)
414        })
415    }
416}
uqa_storage/sqlite/catalog/foreign_indexes.rs

uqa_storage/sqlite/catalog/
foreign_indexes.rs
