uqa_storage/sqlite/catalog/migration/
registry.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Single ordered catalog-migration dispatcher.
8
9use super::super::{
10    params, Catalog, ManagedConnection, OptionalExtension, Result, SQLiteError,
11    CURRENT_SCHEMA_VERSION,
12};
13use super::steps::{MigrationAction, MIGRATIONS};
14
15impl Catalog {
16    /// Open (or create) the catalog and run any pending migrations.
17    pub fn open(conn: ManagedConnection) -> Result<Self> {
18        let mut cat = Self {
19            conn,
20            fts_storage_was_reset: false,
21        };
22        cat.fts_storage_was_reset = cat.run_migrations()?;
23        Ok(cat)
24    }
25
26    pub fn connection(&self) -> ManagedConnection {
27        self.conn.clone()
28    }
29
30    pub(super) fn run_migrations(&self) -> Result<bool> {
31        self.conn.with_mut(|conn| {
32            // Older catalogs (pre-v7) used the table name `_meta`. v7
33            // renames it to `_metadata`; promote the legacy table before
34            // any migration query touches it.
35            let legacy_meta_only: bool = conn
36                .query_row(
37                    "SELECT \
38                        (SELECT COUNT(*) FROM sqlite_master \
39                          WHERE type='table' AND name='_meta') > 0 \
40                     AND (SELECT COUNT(*) FROM sqlite_master \
41                            WHERE type='table' AND name='_metadata') = 0",
42                    [],
43                    |r| r.get::<_, i64>(0),
44                )
45                .optional()?
46                .is_some_and(|n| n != 0);
47            if legacy_meta_only {
48                conn.execute("ALTER TABLE _meta RENAME TO _metadata", [])?;
49            }
50            conn.execute(
51                "CREATE TABLE IF NOT EXISTS _metadata (
52                    key   TEXT PRIMARY KEY,
53                    value TEXT NOT NULL
54                )",
55                [],
56            )?;
57            let current = conn
58                .query_row(
59                    "SELECT value FROM _metadata WHERE key = 'schema_version'",
60                    [],
61                    |r| r.get::<_, String>(0),
62                )
63                .optional()?;
64            let current = match current {
65                Some(version) => version
66                    .parse::<u32>()
67                    .map_err(|_| SQLiteError::InvalidSchemaVersion(version))?,
68                None => 0,
69            };
70            if current > CURRENT_SCHEMA_VERSION {
71                return Err(SQLiteError::UnsupportedSchemaVersion {
72                    found: current,
73                    supported: CURRENT_SCHEMA_VERSION,
74                });
75            }
76
77            debug_assert_eq!(
78                MIGRATIONS.last().map(|migration| migration.version),
79                Some(CURRENT_SCHEMA_VERSION)
80            );
81            for migration in &MIGRATIONS {
82                if migration.version > current {
83                    let tx = conn.transaction()?;
84                    match migration.action {
85                        MigrationAction::Sql(sql) => tx.execute_batch(sql)?,
86                        MigrationAction::Custom(migrate) => migrate(&tx)?,
87                    }
88                    tx.execute(
89                        "INSERT OR REPLACE INTO _metadata (key, value) \
90                         VALUES ('schema_version', ?1)",
91                        params![migration.version.to_string()],
92                    )?;
93                    tx.commit()?;
94                }
95            }
96            let repair = conn.savepoint()?;
97            let schema_before_repair: i64 =
98                repair.pragma_query_value(None, "schema_version", |row| row.get(0))?;
99            Self::ensure_column_stats_shape(&repair)?;
100            let fts_storage_was_reset = Self::ensure_fts_storage_shape(&repair)?;
101            let schema_after_repair: i64 =
102                repair.pragma_query_value(None, "schema_version", |row| row.get(0))?;
103            if schema_before_repair != schema_after_repair {
104                Self::install_cache_revision_tracking(&repair)?;
105            }
106            repair.commit()?;
107            Ok(fts_storage_was_reset)
108        })
109    }
110}
uqa_storage/sqlite/catalog/migration/registry.rs

uqa_storage/sqlite/catalog/migration/
registry.rs
