1use super::{
10 columns_json_references, delete_table_rows_if_exists, drop_fts_aux_tables_for_field,
11 drop_fts_aux_tables_for_table, migration_relation, params,
12 rename_btree_field_rows_or_keep_existing, rename_field_rows_or_keep_existing,
13 rename_fts_aux_tables_for_field, renamed_columns_json, table_exists,
14 update_btree_table_name_rows_if_exists, update_table_name_rows_if_exists, Catalog,
15 OptionalExtension, RelationIdentity, RelationKind, Result, SQLiteError, SchemaRow, TableSchema,
16 VectorFieldSchema,
17};
18
19impl Catalog {
20 pub fn set_metadata(&self, key: &str, value: &str) -> Result<()> {
22 self.conn.with(|c| {
23 c.execute(
24 "INSERT OR REPLACE INTO _metadata (key, value) VALUES (?1, ?2)",
25 params![key, value],
26 )?;
27 Ok(())
28 })
29 }
30
31 pub fn get_metadata(&self, key: &str) -> Result<Option<String>> {
33 self.conn.with(|c| {
34 let v: Option<String> = c
35 .query_row(
36 "SELECT value FROM _metadata WHERE key = ?1",
37 params![key],
38 |r| r.get(0),
39 )
40 .optional()?;
41 Ok(v)
42 })
43 }
44
45 pub fn save_schema(&self, name: &str) -> Result<()> {
46 self.save_schema_row(&SchemaRow::legacy(name))
47 }
48
49 pub fn save_schema_row(&self, schema: &SchemaRow) -> Result<()> {
50 let acl_json = schema.acl.as_ref().map(serde_json::to_string).transpose()?;
51 self.conn.with(|c| {
52 c.execute(
53 "INSERT INTO _schemas (name, role_owner, acl_json) VALUES (?1, ?2, ?3)
54 ON CONFLICT(name) DO UPDATE SET role_owner = excluded.role_owner, acl_json = excluded.acl_json",
55 params![schema.name, schema.role_owner, acl_json],
56 )?;
57 Ok(())
58 })
59 }
60
61 pub fn drop_schema(&self, name: &str) -> Result<()> {
62 self.conn.with(|c| {
63 let relation_count: i64 = c.query_row(
64 "SELECT COUNT(*) FROM _relations WHERE schema_name = ?1",
65 params![name],
66 |row| row.get(0),
67 )?;
68 if relation_count != 0 {
69 return Err(SQLiteError::StorageBackend(format!(
70 "schema `{name}` still owns catalog relations"
71 )));
72 }
73 c.execute("DELETE FROM _schemas WHERE name = ?1", params![name])?;
74 Ok(())
75 })
76 }
77
78 pub fn load_schemas(&self) -> Result<Vec<String>> {
79 Ok(self
80 .load_schema_rows()?
81 .into_iter()
82 .map(|schema| schema.name)
83 .collect())
84 }
85
86 pub fn load_schema_rows(&self) -> Result<Vec<SchemaRow>> {
87 self.conn.with(|c| {
88 let mut stmt =
89 c.prepare("SELECT name, role_owner, acl_json FROM _schemas ORDER BY name")?;
90 let rows = stmt.query_map([], |row| {
91 Ok((
92 row.get::<_, String>(0)?,
93 row.get::<_, String>(1)?,
94 row.get::<_, Option<String>>(2)?,
95 ))
96 })?;
97 let mut out = Vec::new();
98 for row in rows {
99 let (name, role_owner, acl_json) = row?;
100 let acl = acl_json
101 .map(|json| serde_json::from_str(&json))
102 .transpose()?;
103 out.push(SchemaRow {
104 name,
105 role_owner,
106 acl,
107 });
108 }
109 Ok(out)
110 })
111 }
112
113 pub fn save_table(&self, schema: &TableSchema) -> Result<()> {
114 let analyzer = schema.analyzer_json.clone();
115 let fts = serde_json::to_string(&schema.fts_fields)?;
116 let vectors = serde_json::to_string(&schema.vector_fields)?;
117 let columns = schema.columns_json.clone();
118 let constraints = schema.constraints_json.clone();
119 let role_owner = schema.role_owner.clone();
120 let acl_json = schema.acl.as_ref().map(serde_json::to_string).transpose()?;
121 let column_acls_json = serde_json::to_string(&schema.column_acls)?;
122 let object_id = schema.object_id;
123 let storage_generation = schema.storage_generation;
124 self.conn.with_mut(|c| {
125 let tx = c.savepoint()?;
126 Self::claim_relation(&tx, &schema.relation, RelationKind::Table)?;
127 tx.execute(
128 "INSERT INTO _tables
129 (schema_name, relation_name, kind, analyzer, fts_fields,
130 vector_fields, columns, constraints, storage_generation, object_id,
131 role_owner, acl_json, column_acls_json)
132 VALUES (?1, ?2, 'table', ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
133 ON CONFLICT(schema_name, relation_name) DO UPDATE SET
134 analyzer = excluded.analyzer,
135 fts_fields = excluded.fts_fields,
136 vector_fields = excluded.vector_fields,
137 columns = excluded.columns,
138 constraints = excluded.constraints,
139 storage_generation = excluded.storage_generation,
140 object_id = excluded.object_id,
141 role_owner = excluded.role_owner,
142 acl_json = excluded.acl_json,
143 column_acls_json = excluded.column_acls_json",
144 params![
145 schema.relation.schema,
146 schema.relation.name,
147 analyzer,
148 fts,
149 vectors,
150 columns,
151 constraints,
152 storage_generation.as_slice(),
153 object_id.as_slice(),
154 role_owner,
155 acl_json,
156 column_acls_json
157 ],
158 )?;
159 tx.commit()?;
160 Ok(())
161 })
162 }
163
164 pub fn load_tables(&self) -> Result<Vec<TableSchema>> {
165 self.conn.with(|c| {
166 let mut stmt = c.prepare(
167 "SELECT schema_name, relation_name, analyzer, fts_fields,
168 vector_fields, columns, constraints, storage_generation, object_id,
169 role_owner, acl_json, column_acls_json
170 FROM _tables ORDER BY schema_name, relation_name",
171 )?;
172 let rows = stmt.query_map([], |r| {
173 Ok((
174 r.get::<_, String>(0)?,
175 r.get::<_, String>(1)?,
176 r.get::<_, String>(2)?,
177 r.get::<_, String>(3)?,
178 r.get::<_, String>(4)?,
179 r.get::<_, Option<String>>(5)?,
180 r.get::<_, String>(6)?,
181 r.get::<_, Vec<u8>>(7)?,
182 r.get::<_, Vec<u8>>(8)?,
183 r.get::<_, String>(9)?,
184 r.get::<_, Option<String>>(10)?,
185 r.get::<_, Option<String>>(11)?,
186 ))
187 })?;
188 let mut out = Vec::new();
189 for row in rows {
190 let (
191 schema_name,
192 relation_name,
193 analyzer_json,
194 fts_str,
195 vec_str,
196 cols_opt,
197 constraints_json,
198 storage_generation,
199 object_id,
200 role_owner,
201 acl_json,
202 column_acls_json,
203 ) = row?;
204 let fts_fields: Vec<String> = serde_json::from_str(&fts_str)?;
205 let vector_fields: Vec<VectorFieldSchema> = serde_json::from_str(&vec_str)?;
206 let storage_generation: [u8; 16] = storage_generation.try_into().map_err(|value: Vec<u8>| {
207 SQLiteError::StorageBackend(format!(
208 "table `{schema_name}.{relation_name}` has a {}-byte storage generation instead of 16 bytes",
209 value.len()
210 ))
211 })?;
212 let object_id: [u8; 16] = object_id.try_into().map_err(|value: Vec<u8>| {
213 SQLiteError::StorageBackend(format!(
214 "table `{schema_name}.{relation_name}` has a {}-byte object identity instead of 16 bytes",
215 value.len()
216 ))
217 })?;
218 let acl = acl_json
219 .map(|json| serde_json::from_str(&json))
220 .transpose()?;
221 let column_acls = column_acls_json
222 .map(|json| serde_json::from_str(&json))
223 .transpose()?
224 .unwrap_or_default();
225 out.push(TableSchema {
226 relation: RelationIdentity::new(schema_name, relation_name),
227 role_owner,
228 acl,
229 column_acls,
230 object_id,
231 storage_generation,
232 analyzer_json,
233 fts_fields,
234 vector_fields,
235 columns_json: cols_opt.unwrap_or_default(),
236 constraints_json,
237 });
238 }
239 Ok(out)
240 })
241 }
242
243 pub fn drop_table(&self, name: &str) -> Result<()> {
244 let relation = migration_relation(name)?;
245 self.conn.with_mut(|c| {
246 let tx = c.savepoint()?;
247 Self::drop_catalog_index_rows_for_table(&tx, &relation)?;
248 tx.execute(
249 "DELETE FROM _tables WHERE schema_name = ?1 AND relation_name = ?2",
250 params![relation.schema, relation.name],
251 )?;
252 Self::release_relation(&tx, &relation, RelationKind::Table)?;
253 tx.commit()?;
254 Ok(())
255 })
256 }
257
258 pub fn purge_table_data(&self, name: &str) -> Result<()> {
264 let relation = migration_relation(name)?;
265 let storage_names = relation.canonical_and_legacy_public_names();
266 self.conn.with_mut(|c| {
267 let tx = c.savepoint()?;
268 for storage_name in &storage_names {
269 for table in [
270 "_documents",
271 "_document_blobs",
272 "_postings",
273 "_posting_clusters",
274 "_posting_documents",
275 "_doc_lengths",
276 "_field_stats",
277 "_occurrence_clusters",
278 "_occurrence_documents",
279 "_occurrence_lengths",
280 "_occurrence_fields",
281 "_occurrence_formats",
282 "_vectors",
283 "_ivf_indexes",
284 "_ivf_centroids",
285 "_ivf_assignments",
286 "_hnsw_indexes",
287 "_hnsw_nodes",
288 "_hnsw_edges",
289 "_column_stats",
290 "_btree_index_entries",
291 "_btree_indexes",
292 ] {
293 delete_table_rows_if_exists(&tx, table, storage_name)?;
294 }
295 drop_fts_aux_tables_for_table(&tx, storage_name)?;
296 }
297 tx.commit()?;
298 Ok(())
299 })
300 }
301
302 pub fn drop_table_and_data(&self, name: &str) -> Result<()> {
303 let relation = migration_relation(name)?;
304 let storage_names = relation.canonical_and_legacy_public_names();
305 self.conn.with_mut(|c| {
306 let tx = c.savepoint()?;
307 Self::drop_catalog_index_rows_for_table(&tx, &relation)?;
308 tx.execute(
309 "DELETE FROM _tables WHERE schema_name = ?1 AND relation_name = ?2",
310 params![relation.schema, relation.name],
311 )?;
312 for storage_name in &storage_names {
313 for table in [
314 "_documents",
315 "_document_blobs",
316 "_postings",
317 "_posting_clusters",
318 "_posting_documents",
319 "_doc_lengths",
320 "_field_stats",
321 "_occurrence_clusters",
322 "_occurrence_documents",
323 "_occurrence_lengths",
324 "_occurrence_fields",
325 "_occurrence_formats",
326 "_vectors",
327 "_ivf_indexes",
328 "_ivf_centroids",
329 "_ivf_assignments",
330 "_hnsw_indexes",
331 "_hnsw_nodes",
332 "_hnsw_edges",
333 "_column_stats",
334 "_btree_index_entries",
335 "_btree_indexes",
336 ] {
337 delete_table_rows_if_exists(&tx, table, storage_name)?;
338 }
339 tx.execute(
340 "DELETE FROM _table_field_analyzers WHERE table_name = ?1",
341 params![storage_name],
342 )?;
343 drop_fts_aux_tables_for_table(&tx, storage_name)?;
344 }
345 Self::release_relation(&tx, &relation, RelationKind::Table)?;
346 tx.commit()?;
347 Ok(())
348 })
349 }
350
351 pub fn rename_table_data(&self, from: &str, to: &str) -> Result<()> {
352 let from_relation = migration_relation(from)?;
353 let to_relation = migration_relation(to)?;
354 if from_relation == to_relation {
355 return Ok(());
356 }
357 if from_relation.schema != to_relation.schema {
358 return Err(SQLiteError::StorageBackend(
359 "moving a table between schemas is not supported by the catalog".into(),
360 ));
361 }
362 self.conn.with_mut(|c| {
363 let tx = c.savepoint()?;
364 Self::claim_relation(&tx, &to_relation, RelationKind::Table)?;
365 let updated = tx.execute(
366 "UPDATE _tables
367 SET schema_name = ?3, relation_name = ?4
368 WHERE schema_name = ?1 AND relation_name = ?2",
369 params![
370 from_relation.schema,
371 from_relation.name,
372 to_relation.schema,
373 to_relation.name
374 ],
375 )?;
376 if updated == 0 {
377 return Err(SQLiteError::StorageBackend(format!(
378 "table `{from}` does not exist"
379 )));
380 }
381 for table in [
382 "_documents",
383 "_document_blobs",
384 "_postings",
385 "_posting_clusters",
386 "_posting_documents",
387 "_doc_lengths",
388 "_field_stats",
389 "_occurrence_clusters",
390 "_occurrence_documents",
391 "_occurrence_lengths",
392 "_occurrence_fields",
393 "_occurrence_formats",
394 "_vectors",
395 "_ivf_indexes",
396 "_ivf_centroids",
397 "_ivf_assignments",
398 "_hnsw_indexes",
399 "_hnsw_nodes",
400 "_hnsw_edges",
401 "_column_stats",
402 "_table_field_analyzers",
403 ] {
404 update_table_name_rows_if_exists(&tx, table, from, to)?;
405 }
406 update_btree_table_name_rows_if_exists(&tx, from, to)?;
407 drop_fts_aux_tables_for_table(&tx, from)?;
408 Self::release_relation(&tx, &from_relation, RelationKind::Table)?;
409 tx.commit()?;
410 Ok(())
411 })
412 }
413
414 pub fn drop_column_data(&self, table_name: &str, column_name: &str) -> Result<()> {
415 let indexes = self.catalog_indexes_referencing_column(table_name, column_name)?;
416 self.conn.with_mut(|c| {
417 let tx = c.savepoint()?;
418 if table_exists(&tx, "_document_blobs")? {
419 tx.execute(
420 "DELETE FROM _document_blobs WHERE table_name = ?1 AND field_name = ?2",
421 params![table_name, column_name],
422 )?;
423 }
424 for table in [
425 "_postings",
426 "_posting_clusters",
427 "_posting_documents",
428 "_doc_lengths",
429 "_field_stats",
430 "_occurrence_clusters",
431 "_occurrence_documents",
432 "_occurrence_lengths",
433 "_occurrence_fields",
434 "_vectors",
435 "_ivf_indexes",
436 "_ivf_centroids",
437 "_ivf_assignments",
438 "_hnsw_indexes",
439 "_hnsw_nodes",
440 "_hnsw_edges",
441 "_btree_index_entries",
442 "_btree_indexes",
443 ] {
444 if matches!(
445 table,
446 "_postings"
447 | "_posting_clusters"
448 | "_posting_documents"
449 | "_doc_lengths"
450 | "_field_stats"
451 ) && !Self::table_columns(&tx, table)?
452 .is_some_and(|columns| columns.contains_key("field"))
453 {
454 continue;
455 }
456 tx.execute(
457 &format!("DELETE FROM {table} WHERE table_name = ?1 AND field = ?2"),
458 params![table_name, column_name],
459 )?;
460 }
461 tx.execute(
462 "DELETE FROM _column_stats WHERE table_name = ?1 AND column_name = ?2",
463 params![table_name, column_name],
464 )?;
465 tx.execute(
466 "DELETE FROM _table_field_analyzers WHERE table_name = ?1 AND field = ?2",
467 params![table_name, column_name],
468 )?;
469 for index in indexes {
470 tx.execute(
471 "DELETE FROM _catalog_indexes
472 WHERE schema_name = ?1 AND relation_name = ?2",
473 params![index.schema, index.name],
474 )?;
475 Self::release_relation(&tx, &index, RelationKind::Index)?;
476 }
477 drop_fts_aux_tables_for_field(&tx, table_name, column_name)?;
478 tx.commit()?;
479 Ok(())
480 })
481 }
482
483 pub fn rename_column_data(&self, table_name: &str, from: &str, to: &str) -> Result<()> {
484 let index_updates = self.catalog_index_column_renames(table_name, from, to)?;
485 self.conn.with_mut(|c| {
486 let tx = c.savepoint()?;
487 rename_field_rows_or_keep_existing(
488 &tx,
489 "_document_blobs",
490 "field_name",
491 table_name,
492 from,
493 to,
494 )?;
495 for table in [
496 "_postings",
497 "_posting_clusters",
498 "_posting_documents",
499 "_doc_lengths",
500 "_field_stats",
501 "_occurrence_clusters",
502 "_occurrence_documents",
503 "_occurrence_lengths",
504 "_occurrence_fields",
505 "_vectors",
506 "_ivf_indexes",
507 "_ivf_centroids",
508 "_ivf_assignments",
509 "_hnsw_indexes",
510 "_hnsw_nodes",
511 "_hnsw_edges",
512 ] {
513 if matches!(
514 table,
515 "_postings"
516 | "_posting_clusters"
517 | "_posting_documents"
518 | "_doc_lengths"
519 | "_field_stats"
520 ) && !Self::table_columns(&tx, table)?
521 .is_some_and(|columns| columns.contains_key("field"))
522 {
523 continue;
524 }
525 rename_field_rows_or_keep_existing(&tx, table, "field", table_name, from, to)?;
526 }
527 rename_btree_field_rows_or_keep_existing(&tx, table_name, from, to)?;
528 rename_field_rows_or_keep_existing(
529 &tx,
530 "_column_stats",
531 "column_name",
532 table_name,
533 from,
534 to,
535 )?;
536 rename_field_rows_or_keep_existing(
537 &tx,
538 "_table_field_analyzers",
539 "field",
540 table_name,
541 from,
542 to,
543 )?;
544 for (index, columns_json) in index_updates {
545 tx.execute(
546 "UPDATE _catalog_indexes
547 SET columns = ?2
548 WHERE schema_name = ?1 AND relation_name = ?3",
549 params![index.schema, columns_json, index.name],
550 )?;
551 }
552 rename_fts_aux_tables_for_field(&tx, table_name, from, to)?;
553 tx.commit()?;
554 Ok(())
555 })
556 }
557
558 pub(super) fn catalog_indexes_referencing_column(
559 &self,
560 table_name: &str,
561 column_name: &str,
562 ) -> Result<Vec<RelationIdentity>> {
563 let mut out = Vec::new();
564 for row in self.load_catalog_indexes()? {
565 if row.table_name == table_name
566 && columns_json_references(&row.columns_json, column_name)?
567 {
568 out.push(row.relation);
569 }
570 }
571 Ok(out)
572 }
573
574 pub(super) fn catalog_index_column_renames(
575 &self,
576 table_name: &str,
577 from: &str,
578 to: &str,
579 ) -> Result<Vec<(RelationIdentity, String)>> {
580 let mut out = Vec::new();
581 for row in self.load_catalog_indexes()? {
582 if row.table_name != table_name {
583 continue;
584 }
585 if let Some(columns_json) = renamed_columns_json(&row.columns_json, from, to)? {
586 out.push((row.relation, columns_json));
587 }
588 }
589 Ok(out)
590 }
591}