distributed/table/
sql.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use crate::table::{
4    ColumnType, TableMigrationArtifact, TableSchema, TableSchemaAdapter,
5    TableSchemaAdapterCapabilities, TableSchemaBootstrap, TableSchemaRegistry, TableStoreError,
6};
7
8/// SQL dialect used to render table-schema migration artifacts.
9#[derive(Clone, Copy, Debug, PartialEq, Eq)]
10pub enum TableSqlDialect {
11    Sqlite,
12    Postgres,
13}
14
15/// Stateless adapter for generating SQL artifacts from table schemas.
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub struct TableSqlSchemaAdapter {
18    dialect: TableSqlDialect,
19}
20
21impl TableSqlSchemaAdapter {
22    pub fn sqlite() -> Self {
23        Self {
24            dialect: TableSqlDialect::Sqlite,
25        }
26    }
27
28    pub fn postgres() -> Self {
29        Self {
30            dialect: TableSqlDialect::Postgres,
31        }
32    }
33
34    pub fn dialect(&self) -> TableSqlDialect {
35        self.dialect
36    }
37}
38
39impl TableSchemaAdapter for TableSqlSchemaAdapter {
40    fn schema_capabilities(&self) -> TableSchemaAdapterCapabilities {
41        TableSchemaAdapterCapabilities {
42            migration_artifacts: true,
43            schema_verification: false,
44            dev_bootstrap: false,
45        }
46    }
47
48    fn generate_migration_artifacts(
49        &self,
50        registry: &TableSchemaRegistry,
51    ) -> Result<Vec<TableMigrationArtifact>, TableStoreError> {
52        generate_table_migration_artifacts(registry, self.dialect)
53    }
54}
55
56pub fn generate_table_migration_artifacts(
57    registry: &TableSchemaRegistry,
58    dialect: TableSqlDialect,
59) -> Result<Vec<TableMigrationArtifact>, TableStoreError> {
60    Ok(vec![TableMigrationArtifact::new(
61        artifact_name(dialect),
62        table_schema_statements(registry, dialect)?,
63    )])
64}
65
66pub fn table_schema_statements(
67    registry: &TableSchemaRegistry,
68    dialect: TableSqlDialect,
69) -> Result<Vec<String>, TableStoreError> {
70    registry.validate()?;
71
72    let schemas = table_schemas_in_dependency_order(registry)?;
73    let mut statements = Vec::new();
74    for schema in &schemas {
75        statements.push(create_table_statement(schema, dialect)?);
76    }
77    for schema in schemas {
78        statements.extend(index_statements(schema));
79    }
80    Ok(statements)
81}
82
83fn table_schemas_in_dependency_order(
84    registry: &TableSchemaRegistry,
85) -> Result<Vec<&TableSchema>, TableStoreError> {
86    let schemas_by_table = registry
87        .schemas()
88        .map(|schema| (schema.table_name.as_str(), schema))
89        .collect::<BTreeMap<_, _>>();
90    let mut remaining = schemas_by_table.keys().copied().collect::<BTreeSet<_>>();
91    let mut ordered = Vec::with_capacity(remaining.len());
92
93    while !remaining.is_empty() {
94        let ready = remaining
95            .iter()
96            .copied()
97            .filter(|table_name| {
98                let schema = schemas_by_table
99                    .get(table_name)
100                    .expect("remaining table should have schema");
101                schema_dependency_tables(schema)
102                    .all(|dependency| dependency == *table_name || !remaining.contains(dependency))
103            })
104            .collect::<Vec<_>>();
105
106        if ready.is_empty() {
107            let cycle = remaining.into_iter().collect::<Vec<_>>().join(", ");
108            return Err(TableStoreError::Metadata(format!(
109                "table schema foreign-key cycle cannot be bootstrapped inline: {cycle}"
110            )));
111        }
112
113        for table_name in ready {
114            remaining.remove(table_name);
115            ordered.push(
116                *schemas_by_table
117                    .get(table_name)
118                    .expect("ready table should have schema"),
119            );
120        }
121    }
122
123    Ok(ordered)
124}
125
126fn schema_dependency_tables(schema: &TableSchema) -> impl Iterator<Item = &str> {
127    let column_foreign_keys = schema
128        .columns
129        .iter()
130        .filter_map(|column| column.foreign_key.as_ref())
131        .map(|foreign_key| foreign_key.table.as_str());
132    let schema_foreign_keys = schema
133        .foreign_keys
134        .iter()
135        .map(|foreign_key| foreign_key.table.as_str());
136    column_foreign_keys.chain(schema_foreign_keys)
137}
138
139fn create_table_statement(
140    schema: &TableSchema,
141    dialect: TableSqlDialect,
142) -> Result<String, TableStoreError> {
143    let mut definitions = schema
144        .columns
145        .iter()
146        .map(|column| {
147            let mut definition = format!(
148                "{} {}",
149                quote_identifier(&column.column_name),
150                sql_type(&column.column_type, column.jsonb, dialect)?
151            );
152            if !column.nullable || column.primary_key {
153                definition.push_str(" NOT NULL");
154            }
155            if column.has_default {
156                if let Some(default) = column.default.as_deref() {
157                    definition.push_str(" DEFAULT ");
158                    definition.push_str(default);
159                }
160            }
161            Ok(definition)
162        })
163        .collect::<Result<Vec<_>, TableStoreError>>()?;
164
165    if let Some(version_column) = schema.version_column.as_deref() {
166        definitions.push(format!(
167            "{} {} NOT NULL DEFAULT 1",
168            quote_identifier(version_column),
169            sql_type(&ColumnType::UnsignedInteger, false, dialect)?
170        ));
171    }
172
173    definitions.push(format!(
174        "PRIMARY KEY ({})",
175        schema
176            .primary_key
177            .columns
178            .iter()
179            .map(|column| quote_identifier(column))
180            .collect::<Vec<_>>()
181            .join(", ")
182    ));
183
184    for column in &schema.columns {
185        if let Some(foreign_key) = &column.foreign_key {
186            definitions.push(format!(
187                "FOREIGN KEY ({}) REFERENCES {} ({})",
188                quote_identifier(&column.column_name),
189                quote_identifier(&foreign_key.table),
190                quote_identifier(&foreign_key.column)
191            ));
192        }
193    }
194
195    for foreign_key in &schema.foreign_keys {
196        let already_declared_on_column = schema.columns.iter().any(|column| {
197            column.column_name == foreign_key.column
198                && column.foreign_key.as_ref() == Some(foreign_key)
199        });
200        if already_declared_on_column {
201            continue;
202        }
203        definitions.push(format!(
204            "FOREIGN KEY ({}) REFERENCES {} ({})",
205            quote_identifier(&foreign_key.column),
206            quote_identifier(&foreign_key.table),
207            quote_identifier(&foreign_key.column)
208        ));
209    }
210
211    Ok(format!(
212        "CREATE TABLE IF NOT EXISTS {} (\n  {}\n);",
213        quote_identifier(&schema.table_name),
214        definitions.join(",\n  ")
215    ))
216}
217
218fn index_statements(schema: &TableSchema) -> impl Iterator<Item = String> + '_ {
219    schema.indexes.iter().map(|index| {
220        let name = index
221            .name
222            .clone()
223            .unwrap_or_else(|| format!("{}_{}_idx", schema.table_name, index.columns.join("_")));
224        let unique = if index.unique { "UNIQUE " } else { "" };
225        format!(
226            "CREATE {unique}INDEX IF NOT EXISTS {} ON {} ({});",
227            quote_identifier(&name),
228            quote_identifier(&schema.table_name),
229            index
230                .columns
231                .iter()
232                .map(|column| quote_identifier(column))
233                .collect::<Vec<_>>()
234                .join(", ")
235        )
236    })
237}
238
239fn sql_type(
240    column_type: &ColumnType,
241    jsonb: bool,
242    dialect: TableSqlDialect,
243) -> Result<&'static str, TableStoreError> {
244    let type_name = match (dialect, column_type) {
245        (TableSqlDialect::Sqlite, ColumnType::Text) => "TEXT",
246        (TableSqlDialect::Sqlite, ColumnType::Boolean) => "INTEGER",
247        (TableSqlDialect::Sqlite, ColumnType::Integer | ColumnType::UnsignedInteger) => "INTEGER",
248        (TableSqlDialect::Sqlite, ColumnType::Float) => "REAL",
249        (TableSqlDialect::Sqlite, ColumnType::Bytes) => "BLOB",
250        (TableSqlDialect::Sqlite, ColumnType::Json) => "TEXT",
251        (TableSqlDialect::Sqlite, ColumnType::Timestamp) => "TEXT",
252        (TableSqlDialect::Postgres, ColumnType::Text) => "text",
253        (TableSqlDialect::Postgres, ColumnType::Boolean) => "boolean",
254        (TableSqlDialect::Postgres, ColumnType::Integer | ColumnType::UnsignedInteger) => "bigint",
255        (TableSqlDialect::Postgres, ColumnType::Float) => "double precision",
256        (TableSqlDialect::Postgres, ColumnType::Bytes) => "bytea",
257        (TableSqlDialect::Postgres, ColumnType::Json) if jsonb => "jsonb",
258        (TableSqlDialect::Postgres, ColumnType::Json) => "jsonb",
259        (TableSqlDialect::Postgres, ColumnType::Timestamp) => "timestamptz",
260        (_, ColumnType::Unsupported(type_name)) => {
261            return Err(TableStoreError::Metadata(format!(
262                "unsupported table column type `{type_name}`"
263            )));
264        }
265    };
266    Ok(type_name)
267}
268
269fn quote_identifier(value: &str) -> String {
270    format!("\"{}\"", value.replace('"', "\"\""))
271}
272
273fn artifact_name(dialect: TableSqlDialect) -> &'static str {
274    match dialect {
275        TableSqlDialect::Sqlite => "sqlite-tables",
276        TableSqlDialect::Postgres => "postgres-tables",
277    }
278}
279
280pub fn bootstrap_result(registry: &TableSchemaRegistry) -> TableSchemaBootstrap {
281    TableSchemaBootstrap::new(registry.table_names().map(str::to_string))
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287    use crate::outbox::{outbox_message_schema, OUTBOX_MESSAGES_TABLE};
288    use crate::table::{ForeignKey, PrimaryKey, TableColumn, TableSchema};
289
290    #[test]
291    fn renders_outbox_table_schema_for_sqlite() {
292        let mut registry = TableSchemaRegistry::new();
293        registry
294            .register_schema(outbox_message_schema())
295            .expect("schema should register");
296
297        let artifact = generate_table_migration_artifacts(&registry, TableSqlDialect::Sqlite)
298            .expect("artifact should render")
299            .pop()
300            .expect("artifact should exist");
301
302        assert_eq!(artifact.name, "sqlite-tables");
303        assert!(artifact
304            .statements
305            .iter()
306            .any(|statement| statement.contains("CREATE TABLE IF NOT EXISTS \"outbox_messages\"")));
307        assert!(artifact
308            .statements
309            .iter()
310            .any(|statement| statement.contains("\"message_id\" TEXT NOT NULL")));
311        assert!(artifact
312            .statements
313            .iter()
314            .any(|statement| statement.contains("\"created_at\" TEXT NOT NULL")));
315        assert!(artifact.statements.iter().any(|statement| statement
316            .contains("\"updated_at\" TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP")));
317    }
318
319    #[test]
320    fn renders_outbox_table_schema_for_postgres_with_timestamp_columns() {
321        let mut registry = TableSchemaRegistry::new();
322        registry
323            .register_schema(outbox_message_schema())
324            .expect("schema should register");
325
326        let artifact = generate_table_migration_artifacts(&registry, TableSqlDialect::Postgres)
327            .expect("artifact should render")
328            .pop()
329            .expect("artifact should exist");
330
331        assert_eq!(artifact.name, "postgres-tables");
332        assert!(artifact
333            .statements
334            .iter()
335            .any(|statement| statement.contains("\"created_at\" timestamptz NOT NULL")));
336        assert!(artifact
337            .statements
338            .iter()
339            .any(|statement| statement.contains("\"claimed_until\" timestamptz")));
340        assert!(artifact.statements.iter().any(|statement| statement
341            .contains("\"updated_at\" timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP")));
342    }
343
344    #[test]
345    fn orders_parent_tables_before_foreign_key_dependents() {
346        let parent = TableSchema {
347            model_name: "Parent".into(),
348            table_name: "parents".into(),
349            columns: vec![TableColumn::new("parent_id", "parent_id", ColumnType::Text)],
350            primary_key: PrimaryKey::new(["parent_id"]),
351            version_column: None,
352            foreign_keys: Vec::new(),
353            indexes: Vec::new(),
354            relationships: Vec::new(),
355        };
356        let child = TableSchema {
357            model_name: "Child".into(),
358            table_name: "children".into(),
359            columns: vec![
360                TableColumn::new("child_id", "child_id", ColumnType::Text),
361                TableColumn {
362                    foreign_key: Some(ForeignKey::new("parents", "parent_id")),
363                    ..TableColumn::new("parent_id", "parent_id", ColumnType::Text)
364                },
365            ],
366            primary_key: PrimaryKey::new(["child_id"]),
367            version_column: None,
368            foreign_keys: Vec::new(),
369            indexes: Vec::new(),
370            relationships: Vec::new(),
371        };
372        let mut registry = TableSchemaRegistry::new();
373        registry.register_schema(child).expect("child registers");
374        registry.register_schema(parent).expect("parent registers");
375
376        let statements = table_schema_statements(&registry, TableSqlDialect::Postgres)
377            .expect("statements should render");
378        let parent_position = statements
379            .iter()
380            .position(|statement| statement.contains("CREATE TABLE IF NOT EXISTS \"parents\""))
381            .expect("parent statement should exist");
382        let child_position = statements
383            .iter()
384            .position(|statement| statement.contains("CREATE TABLE IF NOT EXISTS \"children\""))
385            .expect("child statement should exist");
386
387        assert!(parent_position < child_position);
388    }
389
390    #[test]
391    fn bootstrap_result_lists_registered_tables() {
392        let mut registry = TableSchemaRegistry::new();
393        registry
394            .register_schema(outbox_message_schema())
395            .expect("schema should register");
396
397        let result = bootstrap_result(&registry);
398
399        assert_eq!(
400            result.bootstrapped_tables,
401            vec![OUTBOX_MESSAGES_TABLE.to_string()]
402        );
403    }
404}
distributed/table/sql.rs

distributed/table/
sql.rs
