diff --git a/crates/core/src/json_util.rs b/crates/core/src/json_util.rs index 2d3fa04..6585181 100644 --- a/crates/core/src/json_util.rs +++ b/crates/core/src/json_util.rs @@ -6,8 +6,8 @@ use core::ffi::c_int; use crate::constants::SUBTYPE_JSON; use crate::create_sqlite_text_fn; use crate::error::{PowerSyncError, Result}; -use powersync_sqlite_nostd as sqlite; use powersync_sqlite_nostd::bindings::{SQLITE_RESULT_SUBTYPE, SQLITE_SUBTYPE}; +use powersync_sqlite_nostd::{self as sqlite, ColumnType}; use powersync_sqlite_nostd::{Connection, Context, Value}; use sqlite::ResultCode; @@ -38,6 +38,10 @@ fn powersync_json_merge_impl( } let mut result = String::from("{"); for arg in args { + if arg.value_type() == ColumnType::Null { + continue; + } + let chunk = arg.text(); if chunk.is_empty() || !chunk.starts_with('{') || !chunk.ends_with('}') { return Err(PowerSyncError::argument_error("Expected json object")); diff --git a/crates/core/src/migrations.rs b/crates/core/src/migrations.rs index 3a8a58e..504dc20 100644 --- a/crates/core/src/migrations.rs +++ b/crates/core/src/migrations.rs @@ -4,7 +4,6 @@ use alloc::format; use alloc::string::{String, ToString}; use alloc::vec::Vec; -use powersync_sqlite_nostd::Context; use powersync_sqlite_nostd::{self as sqlite, Destructor}; use serde::Serialize; use serde_json::json; @@ -12,15 +11,19 @@ use sqlite::ResultCode; use crate::error::{PowerSyncError, Result}; use crate::fix_data::apply_v035_fix; -use crate::schema::inspection::ExistingView; +use crate::schema::inspection::{ExistingTable, ExistingView}; use crate::sync::BucketPriority; use crate::utils::database::Database; +use crate::utils::verify_in_transaction; pub const LATEST_VERSION: i32 = 14; -pub fn powersync_migrate(ctx: *mut sqlite::context, target_version: i32) -> Result<()> { - let local_db = Database::from(ctx.db_handle()); +pub fn initialize_database(db: Database) -> Result<()> { + verify_in_transaction(db)?; + powersync_migrate(db, LATEST_VERSION) +} +pub fn powersync_migrate(local_db: Database, target_version: i32) -> Result<()> { // language=SQLite local_db.exec_safe( c"\ @@ -170,7 +173,8 @@ VALUES(4, // Down migrations are less common, so we're okay about that breaking // in some cases. - for mut view in ExistingView::list(local_db)? { + let tables = ExistingTable::list(local_db)?; + for mut view in ExistingView::list(local_db, &tables)? { view.delete_trigger_sql = String::default(); view.update_trigger_sql = String::default(); view.insert_trigger_sql = String::default(); diff --git a/crates/core/src/schema/common.rs b/crates/core/src/schema/common.rs index f133d44..592869e 100644 --- a/crates/core/src/schema/common.rs +++ b/crates/core/src/schema/common.rs @@ -1,10 +1,18 @@ -use core::slice; +use core::fmt::Write; -use alloc::{string::String, vec::Vec}; +use alloc::{ + string::{String, ToString}, + vec, + vec::Vec, +}; use serde::Deserialize; -use crate::schema::{ - Column, CommonTableOptions, RawTable, Table, raw_table::InferredTableStructure, +use crate::{ + schema::{ + Column, CommonTableOptions, PendingStatement, PendingStatementValue, RawTable, Table, + raw_table::InferredTableStructure, + }, + utils::SqlBuffer, }; /// Utility to wrap both PowerSync-managed JSON tables and raw tables (with their schema snapshot @@ -18,6 +26,17 @@ pub enum SchemaTable<'a> { } impl<'a> SchemaTable<'a> { + /// The type name used for the table when referenced in `ps_crud`, `ps_oplog` and other tables. + pub fn name(&self) -> &str { + match self { + SchemaTable::Json(table) => &table.name, + SchemaTable::Raw { + definition, + schema: _, + } => &definition.name, + } + } + pub fn common_options(&self) -> &CommonTableOptions { match self { Self::Json(table) => &table.options, @@ -28,37 +47,80 @@ impl<'a> SchemaTable<'a> { } } - /// Iterates over defined column names in this table (not including the `id` column). - pub fn column_names(&self) -> impl Iterator { + pub fn columns(&self) -> &'a [Column] { match self { - Self::Json(table) => SchemaTableColumnIterator::Json(table.columns.iter()), + Self::Json(table) => &table.columns, Self::Raw { definition: _, schema, - } => SchemaTableColumnIterator::Raw(schema.columns.iter()), + } => &schema.columns, } } -} -impl<'a> From<&'a Table> for SchemaTable<'a> { - fn from(value: &'a Table) -> Self { - Self::Json(value) + /// Iterates over defined column names in this table (not including the `id` column). + pub fn column_names(&self) -> impl Iterator { + self.columns().iter().map(|c| &*c.name) } -} -enum SchemaTableColumnIterator<'a> { - Json(slice::Iter<'a, Column>), - Raw(slice::Iter<'a, String>), -} + /// Generates a statement of the form `INSERT INTO $tbl ($cols) VALUES (?, ...) ON CONFLICT (id) + /// DO UPDATE SET ...` for the sync client. + pub fn infer_put_stmt(&self, table_name: &str) -> PendingStatement { + let mut buffer = SqlBuffer::new(); + let mut params = vec![]; -impl<'a> Iterator for SchemaTableColumnIterator<'a> { - type Item = &'a str; + buffer.push_str("INSERT INTO "); + let _ = buffer.identifier().write_str(table_name); + buffer.push_str(" (id"); - fn next(&mut self) -> Option { - Some(match self { - Self::Json(iter) => &iter.next()?.name, - Self::Raw(iter) => iter.next()?.as_ref(), - }) + for column in self.column_names() { + buffer.comma(); + let _ = buffer.identifier().write_str(column); + } + buffer.push_str(") VALUES (?1"); + params.push(PendingStatementValue::Id); + + let data_start_index = 2; + for (i, column) in self.column_names().enumerate() { + buffer.comma(); + let _ = write!(&mut buffer, "?{}", i + data_start_index); + params.push(PendingStatementValue::Column(column.to_string())); + } + buffer.push_str(") ON CONFLICT (id) DO UPDATE SET "); + let mut do_update = buffer.comma_separated(); + + // Generate an "x" = ? for all synced columns to update them without affecting local-only + // columns. + for (i, column) in self.column_names().enumerate() { + let entry = do_update.element(); + let _ = entry.identifier().write_str(column); + let _ = write!(entry, " = ?{}", i + data_start_index); + } + + PendingStatement { + sql: buffer.sql, + params, + named_parameters_index: None, + } + } + + /// Generates a statement of the form `DELETE FROM $tbl WHERE id = ?` for the sync client. + pub fn infer_delete_stmt(&self, table_name: &str) -> PendingStatement { + let mut buffer = SqlBuffer::new(); + buffer.push_str("DELETE FROM "); + let _ = buffer.identifier().write_str(table_name); + buffer.push_str(" WHERE id = ?"); + + PendingStatement { + sql: buffer.sql, + params: vec![PendingStatementValue::Id], + named_parameters_index: None, + } + } +} + +impl<'a> From<&'a Table> for SchemaTable<'a> { + fn from(value: &'a Table) -> Self { + Self::Json(value) } } @@ -99,3 +161,61 @@ impl<'de> Deserialize<'de> for ColumnFilter { Ok(Self::from(Vec::::deserialize(deserializer)?)) } } +#[cfg(test)] +mod test { + use alloc::{string::ToString, vec}; + use core::assert_matches; + + use crate::schema::{ + Column, PendingStatementValue, RawTable, SchemaTable, raw_table::InferredTableStructure, + table_info::RawTableSchema, + }; + + #[test] + fn infer_sync_statements() { + let raw_table = RawTable { + name: "users".to_string(), + schema: RawTableSchema::default(), + put: None, + delete: None, + clear: None, + }; + let structure = InferredTableStructure { + columns: vec![ + Column { + name: "foo".to_string(), + type_name: "TEXT".to_string(), + }, + Column { + name: "bar".to_string(), + type_name: "TEXT".to_string(), + }, + ], + }; + let schema_table = SchemaTable::Raw { + definition: &raw_table, + schema: &structure, + }; + + let put = schema_table.infer_put_stmt("tbl"); + assert_eq!( + put.sql, + r#"INSERT INTO "tbl" (id, "foo", "bar") VALUES (?1, ?2, ?3) ON CONFLICT (id) DO UPDATE SET "foo" = ?2, "bar" = ?3"# + ); + assert_eq!(put.params.len(), 3); + assert_matches!(put.params[0], PendingStatementValue::Id); + assert_matches!( + put.params[1], + PendingStatementValue::Column(ref name) if name == "foo" + ); + assert_matches!( + put.params[2], + PendingStatementValue::Column(ref name) if name == "bar" + ); + + let delete = schema_table.infer_delete_stmt("tbl"); + assert_eq!(delete.sql, r#"DELETE FROM "tbl" WHERE id = ?"#); + assert_eq!(delete.params.len(), 1); + assert_matches!(delete.params[0], PendingStatementValue::Id); + } +} diff --git a/crates/core/src/schema/inspection.rs b/crates/core/src/schema/inspection.rs index 09b72f7..286825f 100644 --- a/crates/core/src/schema/inspection.rs +++ b/crates/core/src/schema/inspection.rs @@ -1,18 +1,23 @@ +use core::fmt::Write; + use alloc::borrow::ToOwned; -use alloc::{format, vec}; +use alloc::string::ToString; +use alloc::vec; use alloc::{string::String, vec::Vec}; +use powersync_sqlite_nostd::Destructor; use crate::error::Result; -use crate::utils::SqlBuffer; +use crate::schema::Table; +use crate::schema::raw_table::InferredTableStructure; use crate::utils::database::Database; +use crate::utils::{SqlBuffer, WriteType}; +use crate::views::table_columns_to_json_object; /// An existing PowerSync-managed view that was found in the schema. #[derive(PartialEq)] pub struct ExistingView { /// The name of the view itself. - pub name: String, - /// SQL contents of the `CREATE VIEW` statement. - pub sql: String, + pub key: ViewKey, /// SQL contents of all triggers implementing deletes by forwarding to /// `ps_data` and `ps_crud`. pub delete_trigger_sql: String, @@ -22,55 +27,126 @@ pub struct ExistingView { pub update_trigger_sql: String, } +#[derive(PartialEq)] +pub enum ViewKey { + JsonTable { + /// The name of the view itself. + name: String, + /// SQL contents of the `CREATE VIEW` statement. + sql: String, + }, + DirectTable { + /// The name of the direct table for which this view has been created. + table_name: String, + }, +} + impl ExistingView { - pub fn list(db: Database) -> Result> { + pub fn list(db: Database, existing_tables: &[ExistingTable]) -> Result> { let mut results = vec![]; - let stmt = db.prepare_v2(" -SELECT - view.name, - view.sql, - ifnull(group_concat(trigger1.sql, ';\n' ORDER BY trigger1.name DESC), ''), - ifnull(trigger2.sql, ''), - ifnull(trigger3.sql, '') - FROM sqlite_master view - LEFT JOIN sqlite_master trigger1 - ON trigger1.tbl_name = view.name AND trigger1.type = 'trigger' AND trigger1.name GLOB 'ps_view_delete*' - LEFT JOIN sqlite_master trigger2 - ON trigger2.tbl_name = view.name AND trigger2.type = 'trigger' AND trigger2.name GLOB 'ps_view_insert*' - LEFT JOIN sqlite_master trigger3 - ON trigger3.tbl_name = view.name AND trigger3.type = 'trigger' AND trigger3.name GLOB 'ps_view_update*' - WHERE view.type = 'view' AND view.sql GLOB '*-- powersync-auto-generated' - GROUP BY view.name; - ")?; - while stmt.step()? { - let name = stmt.column_text(0)?.to_owned(); - let sql = stmt.column_text(1)?.to_owned(); - let delete = stmt.column_text(2)?.to_owned(); - let insert = stmt.column_text(3)?.to_owned(); - let update = stmt.column_text(4)?.to_owned(); - - results.push(ExistingView { - name, - sql, - delete_trigger_sql: delete, - insert_trigger_sql: insert, - update_trigger_sql: update, - }); + let find_triggers = db.prepare_v2( + "SELECT name, sql FROM sqlite_schema WHERE type = 'trigger' AND tbl_name = ? ORDER BY name DESC", + )?; + let find_views = db.prepare_v2("SELECT name, sql FROM sqlite_schema WHERE type = 'view' AND sql GLOB '*-- powersync-auto-generated'")?; + + let complete_triggers = |key: ViewKey| -> Result { + find_triggers.bind_text(1, &key.name(), Destructor::STATIC)?; + + let mut insert_trigger_sql = String::new(); + let mut update_trigger_sql = String::new(); + let mut delete_trigger_sql = String::new(); + + while find_triggers.step()? { + let trigger_name = find_triggers.column_text(0)?; + let trigger_sql = find_triggers.column_text(1)?; + + let stmt = if trigger_name.starts_with("ps_view_delete") { + &mut delete_trigger_sql + } else if trigger_name.starts_with("ps_view_insert") { + &mut insert_trigger_sql + } else if trigger_name.starts_with("ps_view_update") { + &mut update_trigger_sql + } else { + continue; + }; + + if !stmt.is_empty() { + stmt.push_str(";\n"); + } + + stmt.push_str(trigger_sql); + } + + find_triggers.reset()?; + Ok(ExistingView { + key, + delete_trigger_sql, + insert_trigger_sql, + update_trigger_sql, + }) + }; + + while find_views.step()? { + let name = find_views.column_text(0)?.to_owned(); + let sql = find_views.column_text(1)?.to_owned(); + + let key = ViewKey::JsonTable { name, sql }; + results.push(complete_triggers(key)?); + } + + for table in existing_tables { + if table.direct.is_some() { + // Direct tables don't have a view, but we still want to collect associated + // triggers. + let key = ViewKey::DirectTable { + table_name: table.name.clone(), + }; + results.push(complete_triggers(key)?); + } } Ok(results) } pub fn drop_by_name(db: Database, name: &str) -> Result<()> { - let q = format!("DROP VIEW IF EXISTS {:}", SqlBuffer::quote_identifier(name)); - db.exec_safe_str(&q)?; + let mut buffer = SqlBuffer::new(); + buffer.drop("VIEW", true, name); + + db.exec_safe_str(&buffer.sql)?; + Ok(()) + } + + pub fn delete_from_db(&self, db: Database) -> Result<()> { + match &self.key { + ViewKey::JsonTable { name, .. } => { + Self::drop_by_name(db, &name)?; + } + ViewKey::DirectTable { table_name } => { + // For json tables, dropping the view also drops the triggers. For direct tables + // where we only want to remove triggers, we need to drop them by name manually. + for write in WriteType::VALUES { + let mut buffer = SqlBuffer::new(); + buffer.drop( + "TRIGGER", + true, + &Table::crud_trigger_name(table_name, *write).to_string(), + ); + + db.exec_safe_str(&buffer.sql)?; + } + } + } + Ok(()) } pub fn create(&self, db: Database) -> Result<()> { - Self::drop_by_name(db, &self.name)?; - db.exec_safe_str(&self.sql)?; + self.delete_from_db(db)?; + + if let ViewKey::JsonTable { sql, .. } = &self.key { + db.exec_safe_str(sql)?; + } db.exec_safe_str(&self.delete_trigger_sql)?; db.exec_safe_str(&self.insert_trigger_sql)?; db.exec_safe_str(&self.update_trigger_sql)?; @@ -79,32 +155,56 @@ SELECT } } +impl ViewKey { + pub fn name(&self) -> &str { + match &self { + ViewKey::JsonTable { name, .. } => name, + ViewKey::DirectTable { table_name } => table_name, + } + } +} + pub struct ExistingTable { pub name: String, pub internal_name: String, pub local_only: bool, + pub direct: Option, } impl ExistingTable { pub fn list(db: Database) -> Result> { + Self::list_filtered(db, false) + } + + pub fn list_filtered(db: Database, ignore_direct: bool) -> Result> { let mut results = vec![]; - let stmt = db.prepare_v2( - " -SELECT name FROM sqlite_master WHERE type = 'table' AND name GLOB 'ps_data_*'; - ", - )?; + let stmt = db.prepare_v2("SELECT name, sql FROM sqlite_master WHERE type = 'table';")?; while stmt.step()? { let internal_name = stmt.column_text(0)?; - let Some((name, local_only)) = Self::external_name(internal_name) else { + let Ok(sql) = stmt.column_text(1) else { continue; }; - results.push(ExistingTable { - internal_name: internal_name.to_owned(), - name: name.to_owned(), - local_only: local_only, - }); + if let Some((name, local_only)) = Self::external_name(internal_name) { + results.push(ExistingTable { + internal_name: internal_name.to_owned(), + name: name.to_owned(), + local_only: local_only, + direct: None, + }); + } else if sql.contains("/* ps-managed") && !ignore_direct { + results.push(ExistingTable { + internal_name: internal_name.to_owned(), + name: internal_name.to_owned(), + local_only: sql.contains("local-only"), + direct: Some(InferredTableStructure::read_from_database( + internal_name, + db, + &None, + )?), + }); + } } Ok(results) @@ -125,4 +225,27 @@ SELECT name FROM sqlite_master WHERE type = 'table' AND name GLOB 'ps_data_*'; None } } + + pub fn move_into_ps_untyped(&self, db: Database) -> Result<()> { + if self.local_only { + return Ok(()); + } + + let mut buffer = SqlBuffer::new(); + buffer.push_str("INSERT INTO ps_untyped(type, id, data) SELECT ?, id, "); + + if let Some(ref schema) = self.direct { + buffer.push_str(&table_columns_to_json_object( + &self.internal_name, + &schema.columns, + )?); + } else { + buffer.push_str("data"); + } + + buffer.push_str(" FROM "); + let _ = buffer.identifier().write_str(&self.internal_name); + + db.exec_text(&buffer.sql, &self.name) + } } diff --git a/crates/core/src/schema/management.rs b/crates/core/src/schema/management.rs index 90f5d8c..c109a58 100644 --- a/crates/core/src/schema/management.rs +++ b/crates/core/src/schema/management.rs @@ -1,123 +1,285 @@ extern crate alloc; use alloc::borrow::ToOwned; +use alloc::collections::BTreeSet; use alloc::collections::btree_map::BTreeMap; use alloc::rc::Rc; -use alloc::string::String; +use alloc::string::{String, ToString}; use alloc::vec::Vec; use alloc::{format, vec}; use core::ffi::c_int; use core::fmt::Write; -use powersync_sqlite_nostd as sqlite; use powersync_sqlite_nostd::Context; +use powersync_sqlite_nostd::{self as sqlite, Destructor}; use sqlite::{Connection, ResultCode, Value}; use crate::create_sqlite_text_fn; use crate::error::{PowerSyncError, Result}; -use crate::schema::inspection::{ExistingTable, ExistingView}; -use crate::schema::table_info::Index; +use crate::migrations::initialize_database; +use crate::schema::inspection::{ExistingTable, ExistingView, ViewKey}; +use crate::schema::raw_table::InferredTableStructure; +use crate::schema::table_info::{CreateTableStatement, Index, JsonDataSource}; +use crate::schema::{Column, Table}; use crate::state::DatabaseState; use crate::utils::database::Database; use crate::utils::{SqlBuffer, verify_in_transaction}; use crate::views::{ powersync_trigger_delete_sql, powersync_trigger_insert_sql, powersync_trigger_update_sql, - powersync_view_sql, + powersync_view_sql, table_columns_to_json_object, }; use super::Schema; -fn update_tables(db: Database, schema: &Schema) -> Result<()> { - let existing_tables = ExistingTable::list(db)?; +fn update_tables( + db: Database, + schema: &Schema, + existing_tables: &[ExistingTable], + existing_views: &mut BTreeMap<&str, &ExistingView>, +) -> Result<()> { let mut existing_tables = { let mut map = BTreeMap::new(); - for table in &existing_tables { + for table in existing_tables { map.insert(&*table.name, table); } map }; for table in &schema.tables { + let mut move_data_from = None::; + if let Some(existing) = existing_tables.remove(&*table.name) { - if existing.local_only != table.local_only() { - // Migrating between local-only and synced tables. This works by deleting - // existing and re-creating the table from scratch. We can re-create first and - // delete the old table afterwards because they have a different name - // (local-only tables have a ps_data_local prefix). - - // To delete the old existing table in the end. - existing_tables.insert(&existing.name, existing); - } else { - // Compatible table exists already, nothing to do. - continue; + // Migrate between JSON-based and direct tables. + match (&existing.direct, table.direct) { + (None, false) => { + // JSON-based table before and now. We might have to migrate between synced and + // local-only tables. + if existing.local_only != table.local_only() { + // Migrating between local-only and synced tables. This works by deleting + // existing and re-creating the table from scratch. We can re-create first + // and delete the old table afterwards because they have a different name + // (local-only tables have a ps_data_local prefix). + + // To delete the old existing table in the end. + existing_tables.insert(&existing.name, existing); + } else { + // Compatible table exists already, nothing to do. + continue; + } + } + (None, true) => { + // When migrating from JSON-based to direct tables, there are four cases to + // consider: + // 1. Local-only to direct local-only: We copy data; delete the old table. + // 2. Local-only to synced: Delete old table, copy from ps_untyped for new. + // 3. Synced to local-only: Move old into ps_untyped; create new from scratch. + // 4. Synced to synced: Copy data; delete old table. + if existing.local_only == table.local_only() { + // Case 1 or 4. + move_data_from = Some(JsonDataSource { + table: &existing.internal_name, + fragment: None, + }); + } else { + // Case 2 and 3 is the default, we'll delete the old table in the end which + // moves to ps_untyped if necessary. + } + + // To delete the existing table in the end. + existing_tables.insert(&existing.name, existing); + + // The direct table we create conflicts with the view. So delete that one first. + if let Some(old_view) = existing_views.remove(&*existing.name) { + old_view.delete_from_db(db)?; + } + } + (Some(old_direct), false) => { + // The four cases to consider here match those from the other direction. + if existing.local_only == table.local_only() { + let json = table_columns_to_json_object( + &existing.internal_name, + &old_direct.columns, + )?; + + move_data_from = Some(JsonDataSource { + table: &existing.name, + fragment: Some(json), + }) + } else { + // Also switching synced / local-only state. We'll delete data for this, no + // need to copy. + } + + // To delete the old table in the end. + existing_tables.insert(&existing.name, existing); + } + (Some(previous), true) => { + if existing.local_only != table.local_only() { + // Unlike with json-based tables where local and synced tables have + // different names, here we need to drop the old table first. + existing_views.remove(existing.name.as_str()); + + if !existing.local_only { + existing.move_into_ps_untyped(db)?; + } + + let mut buffer = SqlBuffer::new(); + buffer.drop("TABLE", false, &existing.internal_name); + db.exec_safe_str(&buffer.sql)?; + } else { + // Otherwise compatible tables might still have different columns, which + // requires a migration for direct tables. + direct_table_migration(db, previous, table, existing_views)?; + continue; + } + } } } // New table. - let quoted_internal_name = SqlBuffer::quote_identifier(&table.internal_name()); + let create_table = { + let mut create = CreateTableStatement::from(table); + if table.direct { + for column in &table.columns { + create.push_any_column(&column.name); + } + } - db.exec_safe_str(&format!( - "CREATE TABLE {:}(id TEXT PRIMARY KEY NOT NULL, data TEXT)", - quoted_internal_name - ))?; + create.finish() + }; + db.exec_safe_str(&create_table.sql)?; - if !table.local_only() { + if let Some(ref old_json_table) = move_data_from { + table.move_from_json(db, old_json_table)?; + } else if !table.local_only() { // MOVE data if any - db.exec_text( - &format!( - "INSERT INTO {:}(id, data) - SELECT id, data - FROM ps_untyped - WHERE type = ?", - quoted_internal_name - ), - &table.name, - )?; - - // language=SQLite - db.exec_text("DELETE FROM ps_untyped WHERE type = ?", &table.name)?; + table.move_from_ps_untyped(db)?; } } // Remaining tables need to be dropped. But first, we want to move their contents to // ps_untyped. for remaining in existing_tables.values() { - if !remaining.local_only { - db.exec_text( - &format!( - "INSERT INTO ps_untyped(type, id, data) SELECT ?, id, data FROM {:}", - SqlBuffer::quote_identifier(&remaining.internal_name) - ), - &remaining.name, - )?; - } + remaining.move_into_ps_untyped(db)?; } // We cannot have any open queries on sqlite_master at the point that we drop tables, otherwise // we get "table is locked" errors. for remaining in existing_tables.values() { - let q = format!( - "DROP TABLE {:}", - SqlBuffer::quote_identifier(&remaining.internal_name) - ); - db.exec_safe_str(&q)?; + let mut buffer = SqlBuffer::new(); + buffer.drop("TABLE", false, &remaining.internal_name); + db.exec_safe_str(&buffer.sql)?; } Ok(()) } -fn create_index_stmt(table_name: &str, index_name: &str, index: &Index) -> String { +fn direct_table_migration( + db: Database, + old: &InferredTableStructure, + new: &Table, + existing_views: &mut BTreeMap<&str, &ExistingView>, +) -> Result<()> { + debug_assert!(new.direct); + + struct ExistingColumn<'a> { + column: &'a Column, + found_in_old: bool, + } + + let mut new_columns: Vec<_> = new + .columns + .iter() + .map(|column| ExistingColumn { + column, + found_in_old: false, + }) + .collect(); + new_columns.sort_by(|a, b| a.column.name.cmp(&b.column.name)); + + let mut deleted_columns = vec![]; + + for old_column in &old.columns { + let Ok(new_column_index) = + new_columns.binary_search_by(|probe| probe.column.name.cmp(&old_column.name)) + else { + deleted_columns.push(old_column); + continue; + }; + + let new_column = &mut new_columns[new_column_index]; + new_column.found_in_old = true; + + // For found columns, the type doesn't matter as we generate ANY types for all of them. + } + + new_columns.retain(|c| !c.found_in_old); + + if new_columns.is_empty() && deleted_columns.is_empty() { + return Ok(()); // Nothing to migrate. + } + + // Migrate the direct table. SQLite validates associated triggers and indexes on ALTER TABLE + // statements, so we drop those first. A subsequent update_indexes and update_views call will + // create them again. + { + let stmt = + db.prepare_v2("SELECT name FROM sqlite_schema WHERE type = 'index' AND sql IS NOT NULL AND tbl_name = ?")?; + stmt.bind_text(1, &new.name, Destructor::STATIC)?; + + while stmt.step()? { + let index_name = stmt.column_text(0)?; + + let mut stmt = SqlBuffer::new(); + stmt.drop("INDEX", false, &index_name); + db.exec_safe_str(&stmt.sql)?; + } + + if let Some(old_triggers) = existing_views.remove(new.name.as_str()) { + old_triggers.delete_from_db(db)?; + } + } + + // Add new columns, drop old ones + for new_column in new_columns { + let mut stmt = SqlBuffer::new(); + stmt.alter_table(&new.name); + stmt.add_column(&new_column.column.name, "ANY"); + db.exec_safe_str(&stmt.sql)?; + } + + for dropped_column in deleted_columns { + let mut stmt = SqlBuffer::new(); + stmt.alter_table(&new.name); + stmt.drop("COLUMN", false, &dropped_column.name); + db.exec_safe_str(&stmt.sql)?; + } + + Ok(()) +} + +fn create_index_stmt(table: &Table, index_name: &str, index: &Index) -> String { let mut sql = SqlBuffer::new(); sql.push_str("CREATE INDEX "); let _ = sql.identifier().write_str(&index_name); + if table.direct { + // We use this to identify old indexes to remove them. This is only required for direct + // tables, for json tables we use the ps_data prefix. + sql.push_str("/* ps-managed */"); + } sql.push_str(" ON "); - let _ = sql.identifier().write_str(&table_name); + table.write_name(&mut sql); sql.push_char('('); { let mut sql = sql.comma_separated(); for indexed_column in &index.columns { let sql = sql.element(); - sql.json_extract_and_cast("data", &indexed_column.name, &indexed_column.type_name); + + if table.direct { + let _ = sql.identifier().write_str(&indexed_column.name); + } else { + sql.json_extract_and_cast("data", &indexed_column.name, &indexed_column.type_name); + } if !indexed_column.ascending { sql.push_str(" DESC"); @@ -131,7 +293,7 @@ fn create_index_stmt(table_name: &str, index_name: &str, index: &Index) -> Strin fn update_indexes(db: Database, schema: &Schema) -> Result<()> { let mut statements: Vec = alloc::vec![]; - let mut expected_index_names: Vec = vec![]; + let mut expected_index_names: BTreeSet = Default::default(); { // In a block so that the statement is finalized before dropping indexes @@ -158,41 +320,34 @@ fn update_indexes(db: Database, schema: &Schema) -> Result<()> { result }; - let sql = create_index_stmt(&table_name, &index_name, index); + let sql = create_index_stmt(&table, &index_name, index); if existing_sql.is_none() { statements.push(sql); } else if existing_sql != Some(&sql) { - statements.push(format!( - "DROP INDEX {}", - SqlBuffer::quote_identifier(&index_name) - )); + let mut drop_stmt = SqlBuffer::new(); + drop_stmt.drop("INDEX", false, &index_name); + + statements.push(drop_stmt.sql); statements.push(sql); } - expected_index_names.push(index_name); + expected_index_names.insert(index_name); } } - // In a block so that the statement is finalized before dropping indexes // language=SQLite let statement = db.prepare_v2( "\ -SELECT - sqlite_master.name as index_name - FROM sqlite_master - WHERE sqlite_master.type = 'index' - AND sqlite_master.name GLOB 'ps_data_*' - AND sqlite_master.name NOT IN (SELECT value FROM json_each(?)) -", +SELECT name FROM sqlite_master + WHERE type = 'index' + AND (name GLOB 'ps_data_*' OR sqlite_master.sql GLOB '* ps-managed *')", )?; - let json_names = serde_json::to_string(&expected_index_names) - .map_err(PowerSyncError::as_argument_error)?; - statement.bind_text(1, &json_names, sqlite::Destructor::STATIC)?; while statement.step()? { let name = statement.column_text(0)?; - - statements.push(format!("DROP INDEX {}", SqlBuffer::quote_identifier(name))); + if !expected_index_names.contains(name) { + statements.push(format!("DROP INDEX {}", SqlBuffer::quote_identifier(name))); + } } } @@ -205,26 +360,27 @@ SELECT Ok(()) } -fn update_views(db: Database, schema: &Schema) -> Result<()> { - // First, find all existing views and index them by name. - let existing = ExistingView::list(db)?; - let mut existing = { - let mut map = BTreeMap::new(); - for entry in &existing { - map.insert(&*entry.name, entry); - } - map - }; - +fn update_views( + db: Database, + schema: &Schema, + existing: &mut BTreeMap<&str, &ExistingView>, +) -> Result<()> { for table in &schema.tables { - let view_sql = powersync_view_sql(table); let delete_trigger_sql = powersync_trigger_delete_sql(table)?; let insert_trigger_sql = powersync_trigger_insert_sql(table)?; let update_trigger_sql = powersync_trigger_update_sql(table)?; let wanted_view = ExistingView { - name: table.view_name().to_owned(), - sql: view_sql, + key: if table.direct { + ViewKey::DirectTable { + table_name: table.name.to_string(), + } + } else { + ViewKey::JsonTable { + name: table.view_name().to_owned(), + sql: powersync_view_sql(table), + } + }, delete_trigger_sql, insert_trigger_sql, update_trigger_sql, @@ -243,7 +399,7 @@ fn update_views(db: Database, schema: &Schema) -> Result<()> { // Delete old views. for remaining in existing.values() { - ExistingView::drop_by_name(db, &remaining.name)?; + remaining.delete_from_db(db)?; } Ok(()) @@ -265,12 +421,21 @@ fn powersync_replace_schema_impl( let parsed_schema = serde_json::from_str::(schema).map_err(PowerSyncError::as_argument_error)?; - // language=SQLite - db.exec_safe(c"SELECT powersync_init()")?; + initialize_database(db)?; - update_tables(db, &parsed_schema)?; + let existing_tables = ExistingTable::list(db)?; + let views: Vec = ExistingView::list(db, &existing_tables)?; + let mut existing_views = { + let mut map = BTreeMap::new(); + for entry in &views { + map.insert(entry.key.name(), entry); + } + map + }; + + update_tables(db, &parsed_schema, &existing_tables, &mut existing_views)?; update_indexes(db, &parsed_schema)?; - update_views(db, &parsed_schema)?; + update_views(db, &parsed_schema, &mut existing_views)?; state.set_schema(parsed_schema); Ok(String::from("")) @@ -304,14 +469,26 @@ pub fn register( mod test { use alloc::{string::ToString, vec}; - use crate::schema::table_info::{Index, IndexedColumn}; + use crate::schema::{ + Table, + table_info::{Index, IndexedColumn}, + }; use super::create_index_stmt; #[test] fn test_create_index() { + let table = Table { + name: "table".to_string(), + view_name_override: None, + columns: Default::default(), + indexes: Default::default(), + options: Default::default(), + direct: false, + }; + let stmt = create_index_stmt( - "table", + &table, "index", &Index { name: "unused".to_string(), @@ -332,7 +509,7 @@ mod test { assert_eq!( stmt, - r#"CREATE INDEX "index" ON "table"(CAST(json_extract(data, '$.a') as text), CAST(json_extract(data, '$.b') as integer) DESC)"# + r#"CREATE INDEX "index" ON "ps_data__table"(CAST(json_extract(data, '$.a') as text), CAST(json_extract(data, '$.b') as integer) DESC)"# ) } } diff --git a/crates/core/src/schema/raw_table.rs b/crates/core/src/schema/raw_table.rs index ad146aa..d15f2c8 100644 --- a/crates/core/src/schema/raw_table.rs +++ b/crates/core/src/schema/raw_table.rs @@ -4,25 +4,20 @@ use core::{ }; use alloc::{ - collections::btree_map::BTreeMap, - format, - rc::Rc, - string::{String, ToString}, - vec, + borrow::ToOwned, collections::btree_map::BTreeMap, format, rc::Rc, string::String, vec, vec::Vec, }; use powersync_sqlite_nostd::Destructor; use crate::{ error::{PowerSyncError, Result}, - schema::{ColumnFilter, PendingStatement, PendingStatementValue, RawTable, SchemaTable}, + schema::{Column, ColumnFilter, PendingStatement, RawTable, SchemaTable}, utils::{InsertIntoCrud, SqlBuffer, WriteType, database::Database}, views::table_columns_to_json_object, }; pub struct InferredTableStructure { - pub name: String, - pub columns: Vec, + pub columns: Vec, } impl InferredTableStructure { @@ -31,7 +26,7 @@ impl InferredTableStructure { db: Database, synced_columns: &Option, ) -> Result { - let stmt = db.prepare_v2("select name from pragma_table_info(?)")?; + let stmt = db.prepare_v2("select name, type from pragma_table_info(?)")?; stmt.bind_text(1, table_name, Destructor::STATIC)?; let mut has_id_column = false; @@ -39,6 +34,8 @@ impl InferredTableStructure { while stmt.step()? { let name = stmt.column_text(0)?; + let column_type = stmt.column_text(1)?; + if name == "id" { has_id_column = true; } else if let Some(filter) = synced_columns @@ -46,7 +43,10 @@ impl InferredTableStructure { { // This column isn't part of the synced columns, skip. } else { - columns.push(name.to_string()); + columns.push(Column { + name: name.to_owned(), + type_name: column_type.to_owned(), + }); } } @@ -59,61 +59,7 @@ impl InferredTableStructure { "Table {table_name} has no id column." ))) } else { - Ok(Self { - name: table_name.to_string(), - columns, - }) - } - } - - /// Generates a statement of the form `INSERT INTO $tbl ($cols) VALUES (?, ...) ON CONFLICT (id) - /// DO UPDATE SET ...` for the sync client. - pub fn infer_put_stmt(&self) -> PendingStatement { - let mut buffer = SqlBuffer::new(); - let mut params = vec![]; - - buffer.push_str("INSERT INTO "); - let _ = buffer.identifier().write_str(&self.name); - buffer.push_str(" (id"); - for column in &self.columns { - buffer.comma(); - let _ = buffer.identifier().write_str(column); - } - buffer.push_str(") VALUES (?1"); - params.push(PendingStatementValue::Id); - for (i, column) in self.columns.iter().enumerate() { - buffer.comma(); - let _ = write!(&mut buffer, "?{}", i + 2); - params.push(PendingStatementValue::Column(column.clone())); - } - buffer.push_str(") ON CONFLICT (id) DO UPDATE SET "); - let mut do_update = buffer.comma_separated(); - // Generated an "x" = ? for all synced columns to update them without affecting local-only - // columns. - for (i, column) in self.columns.iter().enumerate() { - let entry = do_update.element(); - let _ = entry.identifier().write_str(column); - let _ = write!(entry, " = ?{}", i + 2); - } - - PendingStatement { - sql: buffer.sql, - params, - named_parameters_index: None, - } - } - - /// Generates a statement of the form `DELETE FROM $tbl WHERE id = ?` for the sync client. - pub fn infer_delete_stmt(&self) -> PendingStatement { - let mut buffer = SqlBuffer::new(); - buffer.push_str("DELETE FROM "); - let _ = buffer.identifier().write_str(&self.name); - buffer.push_str(" WHERE id = ?"); - - PendingStatement { - sql: buffer.sql, - params: vec![PendingStatementValue::Id], - named_parameters_index: None, + Ok(Self { columns }) } } } @@ -141,7 +87,7 @@ impl InferredSchemaCache { schema_version: usize, tbl: &RawTable, ) -> Result> { - self.with_entry(db, schema_version, tbl, SchemaCacheEntry::put) + self.with_entry(db, schema_version, tbl, |entry| entry.put_stmt.clone()) } pub fn infer_delete_statement( @@ -150,7 +96,7 @@ impl InferredSchemaCache { schema_version: usize, tbl: &RawTable, ) -> Result> { - self.with_entry(db, schema_version, tbl, SchemaCacheEntry::delete) + self.with_entry(db, schema_version, tbl, |entry| entry.delete_stmt.clone()) } fn with_entry( @@ -179,9 +125,8 @@ impl InferredSchemaCache { pub struct SchemaCacheEntry { schema_version: usize, - structure: InferredTableStructure, - put_stmt: Option>, - delete_stmt: Option>, + pub put_stmt: Rc, + pub delete_stmt: Rc, } impl SchemaCacheEntry { @@ -192,26 +137,17 @@ impl SchemaCacheEntry { db, &table.schema.synced_columns, )?; + let schema_table = SchemaTable::Raw { + definition: table, + schema: &structure, + }; Ok(Self { schema_version, - structure, - put_stmt: None, - delete_stmt: None, + put_stmt: Rc::new(schema_table.infer_put_stmt(local_table_name)), + delete_stmt: Rc::new(schema_table.infer_delete_stmt(local_table_name)), }) } - - fn put(&mut self) -> Rc { - self.put_stmt - .get_or_insert_with(|| Rc::new(self.structure.infer_put_stmt())) - .clone() - } - - fn delete(&mut self) -> Rc { - self.delete_stmt - .get_or_insert_with(|| Rc::new(self.structure.infer_delete_stmt())) - .clone() - } } /// Generates a `CREATE TRIGGER` statement to capture writes on raw tables and to forward them to @@ -232,8 +168,24 @@ pub fn generate_raw_table_trigger( schema: &resolved_table, }; + generate_schema_table_trigger( + local_table_name, + as_schema_table, + synced_columns.as_ref(), + trigger_name, + write, + ) +} + +pub fn generate_schema_table_trigger( + local_table_name: &str, + table: SchemaTable, + synced_columns: Option<&ColumnFilter>, + trigger_name: &str, + write: WriteType, +) -> Result { let mut buffer = SqlBuffer::new(); - buffer.create_trigger("", trigger_name); + buffer.create_trigger(trigger_name); buffer.trigger_after(write, local_table_name); // Skip the trigger for writes during sync_local, these aren't crud writes. buffer.push_str("WHEN NOT powersync_in_sync_operation()"); @@ -242,7 +194,7 @@ pub fn generate_raw_table_trigger( buffer.push_str(" AND\n("); // If we have a filter for synced columns (instead of syncing all of them), we want to add // additional WHEN clauses to enesure the trigger runs for updates on those columns only. - for (i, name) in as_schema_table.column_names().enumerate() { + for (i, name) in table.column_names().enumerate() { if i != 0 { buffer.push_str(" OR "); } @@ -257,25 +209,30 @@ pub fn generate_raw_table_trigger( } buffer.push_str(" BEGIN\n"); + let flags = table.common_options().flags; + let mut has_stmt = false; - if table.schema.options.flags.insert_only() { + if flags.insert_only() { if write != WriteType::Insert { // Prevent illegal writes to a table marked as insert-only by raising errors here. buffer.push_str("SELECT RAISE(FAIL, 'Unexpected update on insert-only table');\n"); - } else { + has_stmt = true; + } else if !flags.local_only() { // Insert-only tables use manual CRUD writes so they don't block incoming data. - let fragment = table_columns_to_json_object("NEW", &as_schema_table)?; - buffer.powersync_crud_manual_put(&table.name, &fragment); + let fragment = table_columns_to_json_object("NEW", table.columns())?; + buffer.powersync_crud_manual_put(table.name(), &fragment); + has_stmt = true; } } else { if write == WriteType::Update { // Updates must not change the id. buffer.check_id_not_changed(); + has_stmt = true; } - let json_fragment_new = table_columns_to_json_object("NEW", &as_schema_table)?; + let json_fragment_new = table_columns_to_json_object("NEW", table.columns())?; let json_fragment_old = if write == WriteType::Update { - Some(table_columns_to_json_object("OLD", &as_schema_table)?) + Some(table_columns_to_json_object("OLD", table.columns())?) } else { None }; @@ -294,61 +251,31 @@ pub fn generate_raw_table_trigger( write!(f, ", {json_fragment_new}))") }); - buffer.insert_into_powersync_crud(InsertIntoCrud { - op: write, - table: &as_schema_table, - id_expr: if write == WriteType::Delete { - "OLD.id" - } else { - "NEW.id" - }, - type_name: &table.name, - data: match write { - // There is no data for deleted rows. - WriteType::Delete => None, - _ => Some(&write_data), - }, - metadata: None::<&'static str>, - })?; + if !flags.local_only() { + has_stmt = true; + buffer.insert_into_powersync_crud(InsertIntoCrud { + op: write, + table: &table, + id_expr: if write == WriteType::Delete { + "OLD.id" + } else { + "NEW.id" + }, + type_name: table.name(), + data: match write { + // There is no data for deleted rows. + WriteType::Delete => None, + _ => Some(&write_data), + }, + metadata: None::<&'static str>, + })?; + } + } + + if !has_stmt { + return Ok(Default::default()); } buffer.trigger_end(); Ok(buffer.sql) } - -#[cfg(test)] -mod test { - use alloc::{string::ToString, vec}; - use core::assert_matches; - - use crate::schema::{PendingStatementValue, raw_table::InferredTableStructure}; - - #[test] - fn infer_sync_statements() { - let structure = InferredTableStructure { - name: "tbl".to_string(), - columns: vec!["foo".to_string(), "bar".to_string()], - }; - - let put = structure.infer_put_stmt(); - assert_eq!( - put.sql, - r#"INSERT INTO "tbl" (id, "foo", "bar") VALUES (?1, ?2, ?3) ON CONFLICT (id) DO UPDATE SET "foo" = ?2, "bar" = ?3"# - ); - assert_eq!(put.params.len(), 3); - assert_matches!(put.params[0], PendingStatementValue::Id); - assert_matches!( - put.params[1], - PendingStatementValue::Column(ref name) if name == "foo" - ); - assert_matches!( - put.params[2], - PendingStatementValue::Column(ref name) if name == "bar" - ); - - let delete = structure.infer_delete_stmt(); - assert_eq!(delete.sql, r#"DELETE FROM "tbl" WHERE id = ?"#); - assert_eq!(delete.params.len(), 1); - assert_matches!(delete.params[0], PendingStatementValue::Id); - } -} diff --git a/crates/core/src/schema/table_info.rs b/crates/core/src/schema/table_info.rs index fc46b9d..a292c6a 100644 --- a/crates/core/src/schema/table_info.rs +++ b/crates/core/src/schema/table_info.rs @@ -1,11 +1,18 @@ +use core::fmt::Write; + use alloc::rc::Rc; use alloc::string::ToString; use alloc::vec; use alloc::{collections::btree_set::BTreeSet, format, string::String, vec::Vec}; +use powersync_sqlite_nostd::Destructor; use serde::{Deserialize, de::Visitor}; use crate::error::PowerSyncError; -use crate::schema::ColumnFilter; +use crate::schema::raw_table::generate_schema_table_trigger; +use crate::schema::{ColumnFilter, SchemaTable}; +use crate::sync::PreparedPendingStatement; +use crate::utils::database::{Database, Statement}; +use crate::utils::{CrudTriggerName, SqlBuffer, WriteType}; #[derive(Deserialize)] pub struct Table { @@ -17,6 +24,8 @@ pub struct Table { pub indexes: Vec, #[serde(flatten)] pub options: CommonTableOptions, + #[serde(default)] + pub direct: bool, } /// Options shared between regular and raw tables. @@ -78,6 +87,124 @@ impl Table { format!("ps_data__{:}", self.name) } } + + pub fn move_from_ps_untyped(&self, db: Database) -> Result<(), PowerSyncError> { + let direct = self.direct; + + let mut delete_stmt = SqlBuffer::new(); + delete_stmt.push_str("DELETE FROM ps_untyped WHERE type = ?"); + + if direct { + let _ = delete_stmt.write_str(" RETURNING id, data"); + let source = db.prepare_v2(&delete_stmt.sql)?; + source.bind_text(1, &self.name, Destructor::STATIC)?; + + self.direct_move_from_stmt(db, source)?; + } else { + let mut stmt = SqlBuffer::default(); + stmt.push_str("INSERT INTO "); + self.write_name(&mut stmt); + let _ = stmt.write_str(" (id, data) SELECT id, data FROM ps_untyped WHERE type = ?"); + let _ = db.exec_text(&stmt.sql, &self.name); + db.exec_text(&delete_stmt.sql, &self.name)?; + } + + Ok(()) + } + + pub fn move_from_json( + &self, + db: Database, + json: &JsonDataSource, + ) -> Result<(), PowerSyncError> { + let mut source = SqlBuffer::new(); + // For direct tables, create a SELECT statement returning id and json data we then parse via + // direct_move_from_stmt. For json tables, we directly generate an INSERT INTO SELECT + // statement. + let direct = self.direct; + + if !direct { + source.push_str("INSERT INTO "); + source.quote_internal_name(&self.name, self.local_only()); + source.push_char(' '); + } + + source.push_str("SELECT id, "); + if let Some(ref json_fragment) = json.fragment { + source.push_str(json_fragment); + } else { + source.push_str("data "); + } + source.push_str("FROM "); + let _ = write!(source.identifier(), "{}", json.table); + + let source = db.prepare_v2(&source.sql)?; + + if direct { + self.direct_move_from_stmt(db, source) + } else { + source.exec() + } + } + + /// For direct tables, copies data from a prepared statement returning id and data. + fn direct_move_from_stmt(&self, db: Database, source: Statement) -> Result<(), PowerSyncError> { + debug_assert!(self.direct); + + // Copying into direct tables reqires extracting from JSON. This essentially replays a + // sync_local step for the table, using a custom source. + let stmt = Rc::new(SchemaTable::Json(self).infer_put_stmt(&self.name)); + let stmt = PreparedPendingStatement::prepare(db, stmt)?; + + while source.step()? { + let id = source.column_text(0)?; + let data = source.column_text(1)?; + + let parsed: serde_json::Value = + serde_json::from_str(data).map_err(PowerSyncError::json_local_error)?; + let json_object = parsed.as_object().ok_or_else(|| { + PowerSyncError::argument_error("expected oplog data to be an object") + })?; + stmt.bind_for_put(id, data, Some(json_object), None)?; + stmt.exec(&self.name, id, Some(&data))?; + } + + Ok(()) + } + + pub fn write_name(&self, buffer: &mut SqlBuffer) { + if self.direct { + // Direct tables don't have views, so use the name of the table directly. + let _ = buffer.identifier().write_str(&self.name); + } else { + buffer.quote_internal_name(&self.name, self.local_only()); + } + } + + pub fn generate_direct_trigger( + &self, + mut trigger_name: Option, + write: WriteType, + ) -> Result { + debug_assert!(self.direct); + + generate_schema_table_trigger( + &self.name, + SchemaTable::Json(self), + None, + trigger_name + .get_or_insert_with(|| Self::crud_trigger_name(&self.name, write).to_string()), + write, + ) + } + + pub fn crud_trigger_name<'a>(name: &'a str, write: WriteType) -> CrudTriggerName<'a> { + CrudTriggerName { + write, + name_suffix: "", + view_name: name, + } + } } impl RawTable { @@ -92,6 +219,11 @@ impl RawTable { } } +pub struct JsonDataSource<'a> { + pub table: &'a str, + pub fragment: Option, +} + #[derive(Deserialize)] pub struct Column { pub name: String, @@ -303,6 +435,7 @@ pub struct PendingStatement { pub named_parameters_index: Option, } +#[derive(Default)] pub struct RestColumnIndex { /// All column names referenced by this statement. pub named_parameters: BTreeSet, @@ -370,3 +503,50 @@ pub enum PendingStatementValue { /// The full JSON object for the row, as received from the PowerSync service. Row, } + +pub struct CreateTableStatement { + create_table: SqlBuffer, + direct: bool, +} + +impl From<&Table> for CreateTableStatement { + fn from(value: &Table) -> Self { + let mut create_table = SqlBuffer::new(); + + create_table.push_str("CREATE TABLE "); + value.write_name(&mut create_table); + create_table.push_str("(id TEXT PRIMARY KEY NOT NULL"); + + if value.direct { + create_table.push_str(" /* ps-managed "); + if value.local_only() { + create_table.push_str("local-only "); + } + create_table.push_str("*/"); + } else { + create_table.push_str(", data TEXT"); + } + + Self { + create_table, + direct: value.direct, + } + } +} + +impl CreateTableStatement { + pub fn push_any_column(&mut self, name: &str) { + self.create_table.push_char(','); + self.create_table.column_definition(name, "ANY"); + } + + pub fn finish(mut self) -> SqlBuffer { + self.create_table.push_char(')'); + if self.direct { + self.create_table.push_str(" STRICT"); + } + + self.create_table.push_char(';'); + self.create_table + } +} diff --git a/crates/core/src/sync/mod.rs b/crates/core/src/sync/mod.rs index 874eb29..2c150de 100644 --- a/crates/core/src/sync/mod.rs +++ b/crates/core/src/sync/mod.rs @@ -19,6 +19,7 @@ pub use checksum::Checksum; use crate::state::DatabaseState; pub use streaming_sync::SyncClient; +pub use sync_local::PreparedPendingStatement; pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { interface::register(db, state) diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index fb3c449..01a3ac7 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -11,7 +11,8 @@ use serde::ser::SerializeMap; use crate::error::{PowerSyncError, Result}; use crate::schema::inspection::ExistingTable; use crate::schema::{ - InferredSchemaCache, PendingStatement, PendingStatementValue, RawTable, Schema, + InferredSchemaCache, PendingStatement, PendingStatementValue, RawTable, Schema, SchemaTable, + Table, }; use crate::state::DatabaseState; use crate::sync::BucketPriority; @@ -124,7 +125,6 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' "expected oplog data to be an object", ) })?; - let rest = stmt.render_rest_object(json_object)?; stmt.bind_for_put(id, data, Some(json_object), rest.as_ref())?; stmt.exec(type_name, id, Some(&data))?; @@ -340,6 +340,15 @@ impl<'a> ParsedDatabaseSchema<'a> { } fn add_from_schema(&mut self, schema: &'a Schema) { + for regular in &schema.tables { + if regular.direct && !regular.local_only() { + self.tables.insert( + regular.name.clone(), + ParsedSchemaTable::new(TableDefinition::Direct(regular)), + ); + } + } + for raw in &schema.raw_tables { self.tables.insert( raw.name.clone(), @@ -349,9 +358,11 @@ impl<'a> ParsedDatabaseSchema<'a> { } fn add_from_db(&mut self, db: Database) -> Result<()> { - let tables = ExistingTable::list(db)?; + // Ignore direct tables here, we can rely on them being added via add_from_schema. + // TODO: Remove this function, SDKs should always pass the used schema when they connect. + let tables = ExistingTable::list_filtered(db, true)?; for table in tables { - if !table.local_only { + if !table.local_only && !self.tables.contains_key(&table.name) { let visible_name = table.name; self.tables.insert( @@ -420,6 +431,9 @@ impl<'a> ParsedSchemaTable<'a> { named_parameters_index: None, }) } + TableDefinition::Direct(table) => { + Rc::new(SchemaTable::Json(table).infer_put_stmt(&table.name)) + } }) }) } @@ -448,6 +462,9 @@ impl<'a> ParsedSchemaTable<'a> { named_parameters_index: None, }) } + TableDefinition::Direct(table) => { + Rc::new(SchemaTable::Json(table).infer_delete_stmt(&table.name)) + } }) }) } @@ -456,12 +473,13 @@ impl<'a> ParsedSchemaTable<'a> { enum TableDefinition<'a> { Raw(&'a RawTable), JsonView { local_table: String }, + Direct(&'a Table), } -struct PreparedPendingStatement { +pub struct PreparedPendingStatement { stmt: Statement, - definition: Rc, needs_parsed_json: bool, + definition: Rc, } impl PreparedPendingStatement { diff --git a/crates/core/src/utils/mod.rs b/crates/core/src/utils/mod.rs index 6a11893..62b1aef 100644 --- a/crates/core/src/utils/mod.rs +++ b/crates/core/src/utils/mod.rs @@ -6,7 +6,7 @@ use core::{cmp::Ordering, fmt::Display, hash::Hash}; use alloc::{boxed::Box, string::String}; use serde::Serialize; use serde_json::value::RawValue; -pub use sql_buffer::{InsertIntoCrud, SqlBuffer, WriteType}; +pub use sql_buffer::{CrudTriggerName, InsertIntoCrud, SqlBuffer, WriteType}; use crate::{ error::{PowerSyncError, RawPowerSyncError}, diff --git a/crates/core/src/utils/sql_buffer.rs b/crates/core/src/utils/sql_buffer.rs index 6a9c97d..0b4a51e 100644 --- a/crates/core/src/utils/sql_buffer.rs +++ b/crates/core/src/utils/sql_buffer.rs @@ -69,9 +69,9 @@ impl SqlBuffer { let _ = write!(str, "$.{s}"); } - pub fn create_trigger(&mut self, prefix: &str, view_name: &str) { + pub fn create_trigger(&mut self, name: impl Display) { self.push_str("CREATE TRIGGER "); - self.quote_identifier_prefixed(prefix, view_name); + let _ = write!(self.identifier(), "{}", name); self.push_char(' '); } @@ -107,6 +107,33 @@ impl SqlBuffer { ); } + pub fn alter_table(&mut self, table: &str) { + self.push_str("ALTER TABLE "); + let _ = self.identifier().write_str(table); + self.push_char(' '); + } + + pub fn drop(&mut self, _type: &str, if_exists: bool, name: &str) { + self.push_str("DROP "); + self.push_str(_type); + if if_exists { + self.push_str(" IF EXISTS"); + } + self.push_char(' '); + let _ = self.identifier().write_str(name); + } + + pub fn add_column(&mut self, name: &str, type_name: &str) { + self.push_str("ADD COLUMN "); + self.column_definition(name, type_name); + } + + pub fn column_definition(&mut self, name: &str, type_name: &str) { + let _ = self.identifier().write_str(&name); + self.push_char(' '); + self.push_str(&type_name); + } + /// Writes an `INSERT INTO powersync_crud` statement. pub fn insert_into_powersync_crud( &mut self, @@ -123,7 +150,7 @@ impl SqlBuffer { Some(include_old) => { let old_values = table_columns_to_json_object_with_filter( "OLD", - insert.table, + insert.table.columns(), include_old.column_filter(), )?; @@ -134,7 +161,7 @@ impl SqlBuffer { // only include the powersync_diff of columns matched by the filter. let filtered_new_fragment = table_columns_to_json_object_with_filter( "NEW", - insert.table, + insert.table.columns(), include_old.column_filter(), )?; @@ -320,6 +347,8 @@ pub enum WriteType { } impl WriteType { + pub const VALUES: &[WriteType] = &[WriteType::Insert, WriteType::Update, WriteType::Delete]; + pub fn ps_crud_op_type(&self) -> &'static str { match self { WriteType::Insert => "PUT", @@ -357,6 +386,28 @@ impl FromStr for WriteType { } } +pub struct CrudTriggerName<'a> { + pub write: WriteType, + pub name_suffix: &'a str, + pub view_name: &'a str, +} + +impl<'a> Display for CrudTriggerName<'a> { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + write!( + f, + "{}{}_{}", + match self.write { + WriteType::Insert => "ps_view_insert", + WriteType::Update => "ps_view_update", + WriteType::Delete => "ps_view_delete", + }, + self.name_suffix, + self.view_name + ) + } +} + #[cfg(test)] mod test { use super::SqlBuffer; diff --git a/crates/core/src/view_admin.rs b/crates/core/src/view_admin.rs index cfe8b93..11bda86 100644 --- a/crates/core/src/view_admin.rs +++ b/crates/core/src/view_admin.rs @@ -2,8 +2,7 @@ extern crate alloc; use alloc::format; use alloc::rc::Rc; -use alloc::string::{String, ToString}; -use alloc::vec::Vec; +use alloc::string::String; use core::ffi::{c_int, c_void}; use powersync_sqlite_nostd as sqlite; @@ -12,8 +11,8 @@ use sqlite::{ResultCode, Value}; use crate::create_sqlite_text_fn; use crate::error::{PowerSyncError, Result}; -use crate::migrations::{LATEST_VERSION, powersync_migrate}; -use crate::schema::inspection::ExistingView; +use crate::migrations::{initialize_database, powersync_migrate}; +use crate::schema::inspection::{ExistingTable, ExistingView}; use crate::state::DatabaseState; use crate::utils::database::Database; use crate::utils::{SqlBuffer, verify_in_transaction}; @@ -34,8 +33,7 @@ extern "C" fn powersync_drop_view( fn powersync_init_impl(ctx: *mut sqlite::context, _args: &[*mut sqlite::value]) -> Result { let db = Database::from(ctx.db_handle()); - verify_in_transaction(db)?; - powersync_migrate(ctx, LATEST_VERSION)?; + initialize_database(db)?; Ok(String::from("")) } @@ -50,7 +48,7 @@ fn powersync_test_migration_impl( verify_in_transaction(db)?; let target_version = args[0].int(); - powersync_migrate(ctx, target_version)?; + powersync_migrate(db, target_version)?; Ok(String::from("")) } @@ -89,25 +87,18 @@ DELETE FROM ps_stream_subscriptions; )?; clear_has_synced(local_db)?; - let table_glob = if flags.clear_local() { - "ps_data_*" - } else { - "ps_data__*" - }; - - let tables_stmt = local_db - .prepare_v2("SELECT name FROM sqlite_master WHERE type='table' AND name GLOB ?1")?; - tables_stmt.bind_text(1, table_glob, sqlite::Destructor::STATIC)?; - - let mut tables: Vec = alloc::vec![]; + // Pretend to be in a sync_local step when clearing raw and direct tables. For json-based tables + // we delete from underlying ps_data__ tables to sidestep crud triggers, but some tables have + // triggers directly on the table. + let _skip_crud = state.sync_local_guard(); - while tables_stmt.step()? { - let name = tables_stmt.column_text(0)?; - tables.push(name.to_string()); - } + let existing_tables = ExistingTable::list(local_db)?; + for table in &existing_tables { + if !flags.clear_local() && table.local_only { + continue; + } - for name in tables { - let quoted = SqlBuffer::quote_identifier(&name); + let quoted = SqlBuffer::quote_identifier(&table.internal_name); // The first delete statement deletes a single row, to trigger an update notification for the table. // The second delete statement uses the truncate optimization to delete the remainder of the data. let delete_sql = format!( @@ -120,11 +111,6 @@ DELETE FROM {table};", } if let Some(schema) = state.view_schema() { - // Pretend to be in a sync_local step when clearing raw tables. Similar to the case above - // where we delete from the underlying table to sidestep the CRUD trigger, we don't want - // triggers on raw tables to record this delete in ps_crud. - let _skip_crud = state.sync_local_guard(); - for raw_table in &schema.raw_tables { if let Some(stmt) = &raw_table.clear { local_db diff --git a/crates/core/src/views.rs b/crates/core/src/views.rs index a2c1b51..71f92c1 100644 --- a/crates/core/src/views.rs +++ b/crates/core/src/views.rs @@ -6,8 +6,8 @@ use core::fmt::{Write, from_fn}; use core::mem; use crate::error::{PowerSyncError, Result}; -use crate::schema::{ColumnFilter, SchemaTable, Table}; -use crate::utils::{InsertIntoCrud, SqlBuffer, WriteType}; +use crate::schema::{Column, ColumnFilter, SchemaTable, Table}; +use crate::utils::{CrudTriggerName, InsertIntoCrud, SqlBuffer, WriteType}; pub fn powersync_view_sql(table_info: &Table) -> String { let name = &table_info.name; @@ -59,6 +59,10 @@ pub fn powersync_view_sql(table_info: &Table) -> String { } pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { + if table_info.direct { + return table_info.generate_direct_trigger(None, WriteType::Delete); + } + if table_info.options.flags.insert_only() { // Insert-only tables have no DELETE triggers return Ok(String::new()); @@ -70,7 +74,7 @@ pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { let as_schema_table = SchemaTable::from(table_info); let mut sql = SqlBuffer::new(); - sql.create_trigger("ps_view_delete_", view_name); + sql.create_trigger(Table::crud_trigger_name(view_name, WriteType::Delete)); sql.trigger_instead_of(WriteType::Delete, view_name); sql.push_str("BEGIN\n"); // First, forward to internal data table. @@ -95,7 +99,11 @@ pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { sql.trigger_end(); sql.push_str(";\n"); - sql.create_trigger("ps_view_delete2_", view_name); + sql.create_trigger(CrudTriggerName { + write: WriteType::Delete, + name_suffix: "2", + view_name, + }); sql.trigger_instead_of(WriteType::Update, view_name); sql.push_str("WHEN NEW._deleted IS TRUE BEGIN DELETE FROM "); sql.quote_internal_name(name, local_only); @@ -117,6 +125,10 @@ pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { } pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result { + if table_info.direct { + return table_info.generate_direct_trigger(None, WriteType::Insert); + } + let name = &table_info.name; let view_name = table_info.view_name(); let local_only = table_info.options.flags.local_only(); @@ -124,7 +136,7 @@ pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result { let as_schema_table = SchemaTable::from(table_info); let mut sql = SqlBuffer::new(); - sql.create_trigger("ps_view_insert_", view_name); + sql.create_trigger(Table::crud_trigger_name(view_name, WriteType::Insert)); sql.trigger_instead_of(WriteType::Insert, view_name); sql.push_str("BEGIN\n"); @@ -132,7 +144,7 @@ pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result { sql.check_id_valid(); } - let json_fragment = table_columns_to_json_object("NEW", &as_schema_table)?; + let json_fragment = table_columns_to_json_object("NEW", &table_info.columns)?; if insert_only { // This is using the manual powersync_crud_ instead of powersync_crud because insert-only @@ -168,6 +180,10 @@ pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result { } pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { + if table_info.direct { + return table_info.generate_direct_trigger(None, WriteType::Update); + } + if table_info.options.flags.insert_only() { // Insert-only tables have no UPDATE triggers return Ok(String::new()); @@ -176,10 +192,9 @@ pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { let name = &table_info.name; let view_name = table_info.view_name(); let local_only = table_info.options.flags.local_only(); - let as_schema_table = SchemaTable::from(table_info); let mut sql = SqlBuffer::new(); - sql.create_trigger("ps_view_update_", view_name); + sql.create_trigger(Table::crud_trigger_name(view_name, WriteType::Update)); sql.trigger_instead_of(WriteType::Update, view_name); // If we're supposed to include metadata, we support UPDATE ... SET _deleted = TRUE with @@ -190,8 +205,8 @@ pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { sql.push_str("BEGIN\n"); sql.check_id_not_changed(); - let json_fragment_new = table_columns_to_json_object("NEW", &as_schema_table)?; - let json_fragment_old = table_columns_to_json_object("OLD", &as_schema_table)?; + let json_fragment_new = table_columns_to_json_object("NEW", &table_info.columns)?; + let json_fragment_old = table_columns_to_json_object("OLD", &table_info.columns)?; // UPDATE {internal_name} SET data = {json_fragment_new} WHERE id = NEW.id; sql.push_str("UPDATE "); @@ -206,7 +221,7 @@ pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { sql.insert_into_powersync_crud(InsertIntoCrud { op: WriteType::Update, id_expr: "NEW.id", - table: &as_schema_table, + table: &SchemaTable::Json(table_info), type_name: name, data: Some(&from_fn(|f| { write!( @@ -229,16 +244,13 @@ pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { /// Given a query returning column names, return a JSON object fragment for a trigger. /// /// Example output with prefix "NEW": "json_object('id', NEW.id, 'name', NEW.name, 'age', NEW.age)". -pub fn table_columns_to_json_object<'a>( - prefix: &str, - table: &'a SchemaTable<'a>, -) -> Result { - table_columns_to_json_object_with_filter(prefix, table, None) +pub fn table_columns_to_json_object(prefix: &str, columns: &[Column]) -> Result { + table_columns_to_json_object_with_filter(prefix, columns, None) } pub fn table_columns_to_json_object_with_filter<'a>( prefix: &str, - table: &'a SchemaTable<'a>, + columns: &[Column], filter: Option<&'a ColumnFilter>, ) -> Result { // floor(SQLITE_MAX_FUNCTION_ARG / 2). @@ -262,8 +274,7 @@ pub fn table_columns_to_json_object_with_filter<'a>( buffer.sql } - let mut columns = table.column_names(); - while let Some(name) = columns.next() { + for Column { name, type_name: _ } in columns { if let Some(filter) = filter && !filter.matches(name) { @@ -359,13 +370,14 @@ mod test { ], indexes: vec![], options: Default::default(), + direct: false, }; } #[test] fn test_json_object_fragment() { - let fragment = - table_columns_to_json_object("NEW", &(&test_table()).into()).expect("should generate"); + let columns = &test_table().columns; + let fragment = table_columns_to_json_object("NEW", columns).expect("should generate"); assert_eq!( fragment, diff --git a/dart/test/crud_test.dart b/dart/test/crud_test.dart index d8a9b1b..ba92c4f 100644 --- a/dart/test/crud_test.dart +++ b/dart/test/crud_test.dart @@ -981,6 +981,29 @@ INSERT INTO ps_kv(key, value) VALUES }); }); + test('can clear direct tables', () { + db.executeInTx('SELECT powersync_replace_schema(?)', [ + json.encode({ + 'tables': [ + { + 'name': 'users', + 'columns': [ + {'name': 'name', 'type': 'text'}, + ], + 'direct': true, + } + ] + }) + ]); + + db.execute( + 'INSERT INTO users (id, name) VALUES (uuid(), ?)', ['test user']); + + db.executeInTx('SELECT powersync_clear(0)'); + expect(db.select('SELECT * FROM users'), isEmpty); + expect(db.select('SELECT * FROM ps_crud'), isEmpty); + }); + group('transaction ids', () { setUp(() { db.executeInTx('select powersync_replace_schema(?)', [ diff --git a/dart/test/schema_test.dart b/dart/test/schema_test.dart index dfbd4ff..ad7d09f 100644 --- a/dart/test/schema_test.dart +++ b/dart/test/schema_test.dart @@ -322,6 +322,347 @@ END''', test('#$i', () => testCase.testWith(db)); } }); + + group('direct tables', () { + Object schema({ + Map additionalOptions = const {}, + List additionalColumns = const [], + }) { + return { + 'tables': [ + { + 'name': 'users', + 'columns': [ + {'name': 'name', 'type': 'text'}, + ...additionalColumns, + ], + 'direct': true, + ...additionalOptions, + } + ] + }; + } + + void replaceSchema(Object schema) { + db.executeInTx( + 'SELECT powersync_replace_schema(?)', [json.encode(schema)]); + } + + test('create', () { + replaceSchema({'tables': []}); + db.execute('INSERT INTO ps_untyped (type, id, data) VALUES (?, ?, ?)', [ + 'users', + 'user-id', + json.encode({'name': 'Name', 'other': 3}) + ]); + replaceSchema(schema()); + + expect(db.select('SELECT * FROM users'), [ + { + 'id': 'user-id', + 'name': 'Name', + }, + ]); + + final createTable = db.select( + 'SELECT sql FROM sqlite_schema WHERE type = ? AND tbl_name = ?', + ['table', 'users'], + )[0].columnAt(0); + expect( + createTable, + 'CREATE TABLE "users"(id TEXT PRIMARY KEY NOT NULL /* ps-managed */,"name" ANY) STRICT', + ); + + final triggers = db + .select( + 'SELECT sql FROM sqlite_schema WHERE type = ? AND tbl_name = ? ORDER BY name', + ['trigger', 'users'], + ) + .map((r) => r['sql']) + .toList(); + + expect(triggers, [ + r''' +CREATE TRIGGER "ps_view_delete_users" AFTER DELETE ON "users" FOR EACH ROW WHEN NOT powersync_in_sync_operation() BEGIN +INSERT INTO powersync_crud(op,id,type) VALUES ('DELETE', OLD.id, 'users'); +END''', + r''' +CREATE TRIGGER "ps_view_insert_users" AFTER INSERT ON "users" FOR EACH ROW WHEN NOT powersync_in_sync_operation() BEGIN +INSERT INTO powersync_crud(op,id,type,data) VALUES ('PUT', NEW.id, 'users', json(powersync_diff('{}', json_object('name', powersync_strip_subtype(NEW."name"))))); +END''', + r''' +CREATE TRIGGER "ps_view_update_users" AFTER UPDATE ON "users" FOR EACH ROW WHEN NOT powersync_in_sync_operation() BEGIN +SELECT CASE WHEN (OLD.id != NEW.id) THEN RAISE (FAIL, 'Cannot update id') END; +INSERT INTO powersync_crud(op,id,type,data,options) VALUES ('PATCH', NEW.id, 'users', json(powersync_diff(json_object('name', powersync_strip_subtype(OLD."name")), json_object('name', powersync_strip_subtype(NEW."name")))), 0); +END''' + ]); + }); + + test('local-only', () { + replaceSchema(schema(additionalOptions: {'local_only': true})); + + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + expect(db.select('SELECT * FROM ps_crud'), isEmpty); + }); + + test('remove from schema', () { + replaceSchema(schema()); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + db.executeInTx('SELECT powersync_replace_schema(?)', [ + json.encode({'tables': []}) + ]); + + expect(db.select('SELECT * FROM ps_untyped'), [ + {'type': 'users', 'id': 'id', 'data': '{"name":"name"}'} + ]); + + expect( + db.select( + 'SELECT * FROM sqlite_schema WHERE type = ?', ['trigger']), + isEmpty); + }); + + group('migrate', () { + test('unchanged', () { + final usedSchema = schema(additionalOptions: { + 'indexes': [ + { + 'name': 'test', + 'columns': [ + {'name': 'name', 'type': 'text', 'ascending': true}, + ] + } + ] + }); + + replaceSchema(usedSchema); + + final [versionBefore] = db.select('PRAGMA schema_version'); + replaceSchema(usedSchema); + final [versionAfter] = db.select('PRAGMA schema_version'); + + expect(versionAfter, versionBefore); + }); + + // Test migrating from json to direct tables (and vice versa). + for (final startDirect in [false, true]) { + final fromDesc = startDirect ? 'direct' : 'json'; + final toDesc = startDirect ? 'json' : 'direct'; + final endDirect = !startDirect; + + group('from $fromDesc to $toDesc', () { + test('local-only', () { + replaceSchema(schema(additionalOptions: { + 'local_only': true, + 'direct': startDirect + })); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: { + 'local_only': true, + 'direct': endDirect + })); + expect(db.select('SELECT * FROM users'), hasLength(1)); + }); + + test('local-only to synced', () { + replaceSchema(schema(additionalOptions: { + 'local_only': true, + 'direct': startDirect + })); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: {'direct': endDirect})); + + // Migrating from local-only to synced tables deletes data + expect(db.select('SELECT * FROM users'), isEmpty); + }); + + test('synced', () { + replaceSchema(schema(additionalOptions: {'direct': startDirect})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: {'direct': endDirect})); + expect(db.select('SELECT * FROM users'), hasLength(1)); + expect(db.select('SELECT * FROM ps_crud'), hasLength(1)); + }); + + test('synced to local-only', () { + replaceSchema(schema(additionalOptions: {'direct': startDirect})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + + replaceSchema(schema(additionalOptions: { + 'local_only': true, + 'direct': endDirect + })); + // Data should be deleted when changing to a local-only table, + // previous crud entry is still there. + expect(db.select('SELECT * FROM users'), isEmpty); + expect(db.select('SELECT * FROM ps_crud'), hasLength(1)); + }); + }); + } + + test('from synced to local', () { + replaceSchema(schema()); + db.execute('INSERT INTO users (id, name) VALUES (?, ?)', + ['synced-id', 'name']); + + replaceSchema(schema(additionalOptions: {'local_only': true})); + + expect(db.select('SELECT * FROM ps_untyped'), hasLength(1)); + expect(db.select('SELECT * FROM ps_crud'), hasLength(1)); + expect(db.select('SELECT * FROM users'), isEmpty); + + // A second write on the now local-only table should not be recorded. + db.execute( + 'INSERT INTO users (id, name) VALUES (uuid(), ?)', ['name']); + expect(db.select('SELECT * FROM ps_crud'), hasLength(1)); + }); + + test('from local to synced', () { + replaceSchema(schema(additionalOptions: {'local_only': true})); + db.execute( + 'INSERT INTO users (id, name) VALUES (uuid(), ?)', ['local']); + + // Migrate to synced table. Because the previous local write would + // never get uploaded, this clears local data. + replaceSchema(schema()); + expect(db.select('SELECT * FROM users'), isEmpty); + }); + + test('adding columns', () { + replaceSchema(schema()); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + + replaceSchema(schema(additionalColumns: [ + {'name': 'new-1', 'type': 'text'}, + {'name': 'new-2', 'type': 'integer'}, + ])); + + expect(db.select('SELECT * FROM users'), [ + { + 'id': 'id', + 'name': 'name', + 'new-1': null, + 'new-2': null, + } + ]); + }); + + group('index', () { + final indexes = { + 'indexes': [ + { + 'name': 'test', + 'columns': [ + {'name': 'name', 'type': 'text', 'ascending': true}, + ] + } + ] + }; + + test('add', () { + replaceSchema(schema()); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + + replaceSchema(schema(additionalOptions: indexes)); + expect( + db.select( + 'SELECT sql FROM sqlite_schema WHERE type = ? AND tbl_name = ? AND sql IS NOT NULL', + ['index', 'users']), + [ + { + 'sql': + 'CREATE INDEX "ps_data__users__test"/* ps-managed */ ON "users"("name")' + } + ], + ); + }); + + test('remove', () { + replaceSchema(schema(additionalOptions: indexes)); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + + replaceSchema(schema()); + expect( + db.select( + 'SELECT sql FROM sqlite_schema WHERE type = ? AND tbl_name = ? AND sql IS NOT NULL', + ['index', 'users']), + isEmpty); + }); + }); + + test('change column type', () { + replaceSchema(schema(additionalColumns: [ + {'name': 'additional', 'type': 'text'} + ])); + db.execute( + 'INSERT INTO users (id, name, additional) VALUES (?, ?, ?)', + ['id', 'name', 'text']); + + replaceSchema(schema(additionalColumns: [ + {'name': 'additional', 'type': 'integer'} + ])); + + expect(db.select('SELECT * FROM users'), [ + { + 'id': 'id', + 'name': 'name', + 'additional': 'text', + } + ]); + }); + + test('remove column', () { + replaceSchema(schema(additionalColumns: [ + {'name': 'additional', 'type': 'text'} + ])); + db.execute( + 'INSERT INTO users (id, name, additional) VALUES (?, ?, ?)', + ['id', 'name', 'text']); + replaceSchema(schema(additionalColumns: [])); + + expect(db.select('SELECT * FROM users'), [ + { + 'id': 'id', + 'name': 'name', + } + ]); + }); + + test('multiple column migrations at once', () { + replaceSchema(schema(additionalColumns: [ + {'name': 'removed', 'type': 'text'}, + {'name': 'changed-type', 'type': 'text'}, + ])); + db.execute( + 'INSERT INTO users (id, name, removed, "changed-type") VALUES (?, ?, ?, ?)', + ['id', 'name', 'removed', 'changed-type'], + ); + + replaceSchema(schema(additionalColumns: [ + {'name': 'added', 'type': 'text'}, + {'name': 'changed-type', 'type': 'integer'}, + ])); + + expect(db.select('SELECT * FROM users'), [ + { + 'id': 'id', + 'name': 'name', + 'changed-type': 'changed-type', + 'added': null, + } + ]); + }); + }); + }); }); } diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index b7c969b..e693cca 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -2180,6 +2180,85 @@ CREATE TRIGGER users_ref_delete }); }); + group('direct tables', () { + Object schema({Map additionalOptions = const {}}) { + return { + 'tables': [ + { + 'name': 'users', + 'columns': [ + {'name': 'name', 'type': 'text'} + ], + 'direct': true, + ...additionalOptions, + } + ] + }; + } + + test('smoke test', () { + db.executeInTx( + 'SELECT powersync_replace_schema(?)', [json.encode(schema())]); + invokeControl('start', json.encode({'schema': schema()})); + + // Insert + pushCheckpoint(buckets: [bucketDescription('a')]); + pushSyncData( + 'a', + '1', + 'my_user', + 'PUT', + {'name': 'First user'}, + objectType: 'users', + ); + pushCheckpointComplete(); + + final users = db.select('SELECT * FROM users;'); + expect(users, [ + { + 'id': 'my_user', + 'name': 'First user', + } + ]); + + // Delete + pushCheckpoint(buckets: [bucketDescription('a')]); + pushSyncData( + 'a', + '1', + 'my_user', + 'REMOVE', + null, + objectType: 'users', + ); + pushCheckpointComplete(); + + expect(db.select('SELECT * FROM users'), isEmpty); + }); + + test('local only', () { + final localOnlySchema = schema(additionalOptions: {'local_only': true}); + + db.executeInTx( + 'SELECT powersync_replace_schema(?)', [json.encode(localOnlySchema)]); + invokeControl('start', json.encode({'schema': localOnlySchema})); + + // Insert + pushCheckpoint(buckets: [bucketDescription('a')]); + pushSyncData( + 'a', + '1', + 'my_user', + 'PUT', + {'name': 'First user'}, + objectType: 'users', + ); + pushCheckpointComplete(); + + expect(db.select('SELECT * FROM ps_untyped'), hasLength(1)); + }); + }); + test('can close database while iteration is active', () { // The sync client caches prepared statements, we need to ensure those are // freed when we close the connection since SQLite would keep files open