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..b1c72ff 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; @@ -15,12 +14,16 @@ use crate::fix_data::apply_v035_fix; use crate::schema::inspection::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"\ diff --git a/crates/core/src/schema/common.rs b/crates/core/src/schema/common.rs index f133d44..c0d93d5 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, table_info::RestColumnIndex, + }, + 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,101 @@ 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![]; + let mut rest = match self { + SchemaTable::Json(_) => Some(("_rest", RestColumnIndex::default())), + SchemaTable::Raw { .. } => None, + }; + + buffer.push_str("INSERT INTO "); + let _ = buffer.identifier().write_str(table_name); + buffer.push_str(" (id"); + if let Some((column, _)) = rest { + let _ = write!(&mut buffer, ", {column}"); + } + + for column in self.column_names() { + buffer.comma(); + let _ = buffer.identifier().write_str(column); + } + buffer.push_str(") VALUES (?1"); + params.push(PendingStatementValue::Id); + if let Some((_, ref mut rest_index)) = rest { + params.push(PendingStatementValue::Rest); + buffer.push_str(", ?2"); + rest_index.rest_parameter_positions.push(1); // this is zero-indexed + } -impl<'a> Iterator for SchemaTableColumnIterator<'a> { - type Item = &'a str; + let data_start_index = if rest.is_some() { 3 } else { 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())); - fn next(&mut self) -> Option { - Some(match self { - Self::Json(iter) => &iter.next()?.name, - Self::Raw(iter) => iter.next()?.as_ref(), - }) + if let Some((_, ref mut index)) = rest { + index.named_parameters.insert(column.to_string()); + } + } + buffer.push_str(") ON CONFLICT (id) DO UPDATE SET "); + let mut do_update = buffer.comma_separated(); + + if let Some((column, _)) = rest { + let entry = do_update.element(); + let _ = write!(entry, "{column} = ?2"); + } + + // 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: rest.map(|e| e.1), + } + } + + /// 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 +182,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..3e76857 100644 --- a/crates/core/src/schema/inspection.rs +++ b/crates/core/src/schema/inspection.rs @@ -1,10 +1,14 @@ +use core::fmt::Write; + use alloc::borrow::ToOwned; use alloc::{format, vec}; use alloc::{string::String, vec::Vec}; use crate::error::Result; +use crate::schema::raw_table::InferredTableStructure; use crate::utils::SqlBuffer; use crate::utils::database::Database; +use crate::views::table_columns_to_json_object; /// An existing PowerSync-managed view that was found in the schema. #[derive(PartialEq)] @@ -12,7 +16,9 @@ pub struct ExistingView { /// The name of the view itself. pub name: String, /// SQL contents of the `CREATE VIEW` statement. - pub sql: String, + /// + /// This is not set for direct tables, which don't have a view. + pub sql: Option, /// SQL contents of all triggers implementing deletes by forwarding to /// `ps_data` and `ps_crud`. pub delete_trigger_sql: String, @@ -52,7 +58,7 @@ SELECT results.push(ExistingView { name, - sql, + sql: Some(sql), delete_trigger_sql: delete, insert_trigger_sql: insert, update_trigger_sql: update, @@ -68,9 +74,15 @@ SELECT Ok(()) } + pub fn delete_from_db(&self, db: Database) -> Result<()> { + Self::drop_by_name(db, &self.name) + } + pub fn create(&self, db: Database) -> Result<()> { - Self::drop_by_name(db, &self.name)?; - db.exec_safe_str(&self.sql)?; + if let Some(create_view) = &self.sql { + Self::drop_by_name(db, &self.name)?; + db.exec_safe_str(create_view)?; + } db.exec_safe_str(&self.delete_trigger_sql)?; db.exec_safe_str(&self.insert_trigger_sql)?; db.exec_safe_str(&self.update_trigger_sql)?; @@ -83,28 +95,44 @@ 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, + true, + )?), + }); + } } Ok(results) @@ -125,4 +153,29 @@ 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("powersync_json_merge("); + buffer.push_str(&table_columns_to_json_object( + &self.internal_name, + &schema.columns, + )?); + buffer.push_str(", _rest)"); + } 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..75aa052 100644 --- a/crates/core/src/schema/management.rs +++ b/crates/core/src/schema/management.rs @@ -15,6 +15,7 @@ use sqlite::{Connection, ResultCode, Value}; use crate::create_sqlite_text_fn; use crate::error::{PowerSyncError, Result}; +use crate::migrations::initialize_database; use crate::schema::inspection::{ExistingTable, ExistingView}; use crate::schema::table_info::Index; use crate::state::DatabaseState; @@ -27,7 +28,11 @@ use crate::views::{ use super::Schema; -fn update_tables(db: Database, schema: &Schema) -> Result<()> { +fn update_tables( + db: Database, + schema: &Schema, + existing_views: &mut BTreeMap<&str, &ExistingView>, +) -> Result<()> { let existing_tables = ExistingTable::list(db)?; let mut existing_tables = { let mut map = BTreeMap::new(); @@ -38,59 +43,97 @@ fn update_tables(db: Database, schema: &Schema) -> Result<()> { }; for table in &schema.tables { + let mut move_data_from = None::<&str>; + 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; + 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(&existing.internal_name); + } 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(_), false) => { + return Err(PowerSyncError::argument_error( + "Switching from direct to json-based tables is not yet implemented.", + )); + } + (Some(_), true) => { + // TODO: Consider migrations in schema tables. + } } } // New table. - let quoted_internal_name = SqlBuffer::quote_identifier(&table.internal_name()); + let mut create_table = SqlBuffer::default(); + + create_table.push_str("CREATE TABLE "); + table.write_name(&mut create_table); + _ = write!(&mut create_table, "(id TEXT PRIMARY KEY NOT NULL"); - db.exec_safe_str(&format!( - "CREATE TABLE {:}(id TEXT PRIMARY KEY NOT NULL, data TEXT)", - quoted_internal_name - ))?; + if table.direct { + create_table.push_str(", _rest TEXT /* ps-managed "); + if table.local_only() { + create_table.push_str("local-only "); + } + create_table.push_str("*/"); + + for column in &table.columns { + create_table.push_char(','); + let _ = create_table.identifier().write_str(&column.name); + let _ = write!(&mut create_table, " {}", column.type_name); + } + } else { + create_table.push_str(", data TEXT"); + } + create_table.push_str(");"); + db.exec_safe_str(&create_table.sql)?; - if !table.local_only() { + if let Some(old_json_table) = move_data_from { + table.direct_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 @@ -205,19 +248,17 @@ 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 view_sql = if table.direct { + None + } else { + Some(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)?; @@ -265,12 +306,20 @@ 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)?; + + let views = ExistingView::list(db)?; + let mut existing_views = { + let mut map = BTreeMap::new(); + for entry in &views { + map.insert(&*entry.name, entry); + } + map + }; - update_tables(db, &parsed_schema)?; + update_tables(db, &parsed_schema, &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("")) diff --git a/crates/core/src/schema/raw_table.rs b/crates/core/src/schema/raw_table.rs index ad146aa..57631b7 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 { @@ -30,8 +25,9 @@ impl InferredTableStructure { table_name: &str, db: Database, synced_columns: &Option, + is_direct: bool, ) -> 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,14 +35,21 @@ 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 && !filter.matches(name) { // This column isn't part of the synced columns, skip. + } else if is_direct && name == "_rest" { + // _rest column is an artifact of direct tables, skip. } else { - columns.push(name.to_string()); + columns.push(Column { + name: name.to_owned(), + type_name: column_type.to_owned(), + }); } } @@ -59,61 +62,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 +90,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 +99,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 +128,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 { @@ -191,27 +139,19 @@ impl SchemaCacheEntry { local_table_name, db, &table.schema.synced_columns, + false, )?; + 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 @@ -225,13 +165,29 @@ pub fn generate_raw_table_trigger( let local_table_name = table.require_table_name()?; let synced_columns = &table.schema.synced_columns; let resolved_table = - InferredTableStructure::read_from_database(local_table_name, db, synced_columns)?; + InferredTableStructure::read_from_database(local_table_name, db, synced_columns, false)?; let as_schema_table = SchemaTable::Raw { definition: table, 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.trigger_after(write, local_table_name); @@ -242,7 +198,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 +213,28 @@ 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 { + } 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); } } 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 +253,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..f745958 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::{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,91 @@ 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 direct_move_from_json( + &self, + db: Database, + json_table: &str, + ) -> Result<(), PowerSyncError> { + debug_assert!(self.direct); + + let mut source = SqlBuffer::new(); + source.push_str("SELECT id, data FROM "); + let _ = write!(source.identifier(), "{}", json_table); + + let source = db.prepare_v2(&source.sql)?; + self.direct_move_from_stmt(db, source) + } + + /// 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") + })?; + let rest = stmt.render_rest_object(json_object)?; + stmt.bind_for_put(id, data, Some(json_object), rest.as_ref())?; + 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, write: WriteType) -> Result { + debug_assert!(self.direct); + generate_schema_table_trigger( + &self.name, + SchemaTable::Json(self), + None, + &format!("{}_trigger_{}", self.name, write), + write, + ) + } } impl RawTable { @@ -303,6 +397,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, 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/sql_buffer.rs b/crates/core/src/utils/sql_buffer.rs index 6a9c97d..3c68e24 100644 --- a/crates/core/src/utils/sql_buffer.rs +++ b/crates/core/src/utils/sql_buffer.rs @@ -123,7 +123,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 +134,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(), )?; diff --git a/crates/core/src/view_admin.rs b/crates/core/src/view_admin.rs index cfe8b93..702fa16 100644 --- a/crates/core/src/view_admin.rs +++ b/crates/core/src/view_admin.rs @@ -12,7 +12,7 @@ use sqlite::{ResultCode, Value}; use crate::create_sqlite_text_fn; use crate::error::{PowerSyncError, Result}; -use crate::migrations::{LATEST_VERSION, powersync_migrate}; +use crate::migrations::{initialize_database, powersync_migrate}; use crate::schema::inspection::ExistingView; use crate::state::DatabaseState; use crate::utils::database::Database; @@ -34,8 +34,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 +49,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("")) } diff --git a/crates/core/src/views.rs b/crates/core/src/views.rs index a2c1b51..ffe2e3e 100644 --- a/crates/core/src/views.rs +++ b/crates/core/src/views.rs @@ -6,7 +6,7 @@ use core::fmt::{Write, from_fn}; use core::mem; use crate::error::{PowerSyncError, Result}; -use crate::schema::{ColumnFilter, SchemaTable, Table}; +use crate::schema::{Column, ColumnFilter, SchemaTable, Table}; use crate::utils::{InsertIntoCrud, SqlBuffer, WriteType}; pub fn powersync_view_sql(table_info: &Table) -> String { @@ -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(WriteType::Delete); + } + if table_info.options.flags.insert_only() { // Insert-only tables have no DELETE triggers return Ok(String::new()); @@ -117,6 +121,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(WriteType::Insert); + } + let name = &table_info.name; let view_name = table_info.view_name(); let local_only = table_info.options.flags.local_only(); @@ -132,7 +140,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 +176,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(WriteType::Update); + } + if table_info.options.flags.insert_only() { // Insert-only tables have no UPDATE triggers return Ok(String::new()); @@ -176,7 +188,6 @@ 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); @@ -190,8 +201,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 +217,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 +240,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 +270,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 +366,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/schema_test.dart b/dart/test/schema_test.dart index dfbd4ff..fd395f5 100644 --- a/dart/test/schema_test.dart +++ b/dart/test/schema_test.dart @@ -322,6 +322,158 @@ END''', test('#$i', () => testCase.testWith(db)); } }); + + group('direct tables', () { + Object schema({Map additionalOptions = const {}}) { + return { + 'tables': [ + { + 'name': 'users', + 'columns': [ + {'name': 'name', 'type': 'text'} + ], + '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', + '_rest': '{"other":3}', + }, + ]); + + 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, _rest TEXT /* ps-managed */,"name" text)', + ); + + 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 "users_trigger_DELETE" 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 "users_trigger_INSERT" 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 "users_trigger_UPDATE" 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', () { + group('from json to direct', () { + test('local-only', () { + replaceSchema(schema( + additionalOptions: {'local_only': true, 'direct': false})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: {'local_only': true})); + expect(db.select('SELECT * FROM users'), hasLength(1)); + }); + + test('local-only to synced', () { + replaceSchema(schema( + additionalOptions: {'local_only': true, 'direct': false})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: {})); + + // Migrating from local-only to synced tables deletes data + expect(db.select('SELECT * FROM users'), isEmpty); + }); + + test('synced', () { + replaceSchema(schema(additionalOptions: {'direct': false})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + replaceSchema(schema(additionalOptions: {})); + 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': false})); + db.execute( + 'INSERT INTO users (id, name) VALUES (?, ?)', ['id', 'name']); + + replaceSchema(schema(additionalOptions: {'local_only': true})); + // 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)); + }); + }); + + // todo: from json to direct + // todo: from direct to json + + // todo: add column + // todo: change column type + // todo: remove column + // todo: split columns + }); + }); }); } diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index b7c969b..211a205 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -2180,6 +2180,86 @@ 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', + '_rest': null, + } + ]); + + // 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