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 cat = Self::for_initial_restore(conn);
19        cat.initialize_storage()?;
20        Ok(cat)
21    }
22
23    pub(crate) fn for_initial_restore(conn: ManagedConnection) -> Self {
24        Self { conn }
25    }
26
27    pub(in crate::catalog) fn initialize_storage(&self) -> Result<()> {
28        self.run_migrations()
29    }
30
31    pub fn connection(&self) -> ManagedConnection {
32        self.conn.clone()
33    }
34
35    pub(super) fn run_migrations(&self) -> Result<()> {
36        self.conn.with_mut(|conn| {
37            // A savepoint joins Engine's initial restore or owns the entire standalone catalog open. Later migration failures must also restore the original schema and postings.
38            let mut conn = conn.savepoint()?;
39            // Older catalogs (pre-v7) used the table name `_meta`. v7
40            // renames it to `_metadata`; promote the legacy table before
41            // any migration query touches it.
42            let legacy_meta_only: bool = conn
43                .query_row(
44                    "SELECT \
45                        (SELECT COUNT(*) FROM sqlite_master \
46                          WHERE type='table' AND name='_meta') > 0 \
47                     AND (SELECT COUNT(*) FROM sqlite_master \
48                            WHERE type='table' AND name='_metadata') = 0",
49                    [],
50                    |r| r.get::<_, i64>(0),
51                )
52                .optional()?
53                .is_some_and(|n| n != 0);
54            if legacy_meta_only {
55                conn.execute("ALTER TABLE _meta RENAME TO _metadata", [])?;
56            }
57            conn.execute(
58                "CREATE TABLE IF NOT EXISTS _metadata (
59                    key   TEXT PRIMARY KEY,
60                    value TEXT NOT NULL
61                )",
62                [],
63            )?;
64            let current = conn
65                .query_row(
66                    "SELECT value FROM _metadata WHERE key = 'schema_version'",
67                    [],
68                    |r| r.get::<_, String>(0),
69                )
70                .optional()?;
71            let current = match current {
72                Some(version) => version
73                    .parse::<u32>()
74                    .map_err(|_| SQLiteError::InvalidSchemaVersion(version))?,
75                None => 0,
76            };
77            if current > CURRENT_SCHEMA_VERSION {
78                return Err(SQLiteError::UnsupportedSchemaVersion {
79                    found: current,
80                    supported: CURRENT_SCHEMA_VERSION,
81                });
82            }
83
84            debug_assert_eq!(
85                MIGRATIONS.last().map(|migration| migration.version),
86                Some(CURRENT_SCHEMA_VERSION)
87            );
88            for migration in &MIGRATIONS {
89                if migration.version > current {
90                    let tx = conn.savepoint()?;
91                    match migration.action {
92                        MigrationAction::Sql(sql) => tx.execute_batch(sql)?,
93                        MigrationAction::Custom(migrate) => migrate(&tx)?,
94                    }
95                    tx.execute(
96                        "INSERT OR REPLACE INTO _metadata (key, value) \
97                         VALUES ('schema_version', ?1)",
98                        params![migration.version.to_string()],
99                    )?;
100                    tx.commit()?;
101                }
102            }
103            let repair = conn.savepoint()?;
104            let schema_before_repair: i64 =
105                repair.pragma_query_value(None, "schema_version", |row| row.get(0))?;
106            Self::ensure_column_stats_shape(&repair)?;
107            let schema_after_repair: i64 =
108                repair.pragma_query_value(None, "schema_version", |row| row.get(0))?;
109            if schema_before_repair != schema_after_repair {
110                Self::install_cache_revision_tracking(&repair)?;
111            }
112            repair.commit()?;
113            conn.commit()?;
114            Ok(())
115        })
116    }
117}
uqa_storage_sqlite/catalog/migration/registry.rs

uqa_storage_sqlite/catalog/migration/
registry.rs
