Merge pull request #87 from powersync-ja/raw-tables

Raw tables

Author: simolus3

crates/core/src/crud_vtab.rs

@@ -2,9 +2,10 @@ extern crate alloc;
 
 use alloc::boxed::Box;
 use alloc::string::String;
+use alloc::sync::Arc;
 use const_format::formatcp;
 use core::ffi::{c_char, c_int, c_void, CStr};
-use core::ptr::null_mut;
+use core::sync::atomic::Ordering;
 use serde::Serialize;
 use serde_json::value::RawValue;
 
@@ -15,6 +16,7 @@ use sqlite_nostd::{self as sqlite, ColumnType};
 use crate::error::SQLiteError;
 use crate::ext::SafeManagedStmt;
 use crate::schema::TableInfoFlags;
+use crate::state::DatabaseState;
 use crate::util::MAX_OP_ID;
 use crate::vtab_util::*;
 
@@ -41,6 +43,7 @@ struct VirtualTable {
     db: *mut sqlite::sqlite3,
     current_tx: Option<ActiveCrudTransaction>,
     is_simple: bool,
+    state: Arc<DatabaseState>,
 }
 
 struct ActiveCrudTransaction {
@@ -86,6 +89,15 @@ impl VirtualTable {
             .ok_or_else(|| SQLiteError(ResultCode::MISUSE, Some(String::from("No tx_id"))))?;
         let db = self.db;
 
+        if self.state.is_in_sync_local.load(Ordering::Relaxed) {
+            // Don't collect CRUD writes while we're syncing the local database - writes made here
+            // aren't writes we should upload.
+            // This normally doesn't happen because we insert directly into the data tables, but
+            // users might have custom raw tables used for sycing with triggers on them to call
+            // this function. And those specifically should not trigger here.
+            return Ok(());
+        }
+
         match &mut current_tx.mode {
             CrudTransactionMode::Manual(manual) => {
                 // Columns are (data TEXT, options INT HIDDEN)
@@ -258,7 +270,7 @@ fn prepare_lazy(
 
 extern "C" fn connect(
     db: *mut sqlite::sqlite3,
-    _aux: *mut c_void,
+    aux: *mut c_void,
     argc: c_int,
     argv: *const *const c_char,
     vtab: *mut *mut sqlite::vtab,
@@ -289,6 +301,14 @@ extern "C" fn connect(
                 pModule: core::ptr::null(),
                 zErrMsg: core::ptr::null_mut(),
             },
+            state: {
+                // Increase refcount - we can't use from_raw alone because we don't own the aux
+                // data (connect could be called multiple times).
+                let state = Arc::from_raw(aux as *mut DatabaseState);
+                let clone = state.clone();
+                core::mem::forget(state);
+                clone
+            },
             db,
             current_tx: None,
             is_simple,
@@ -380,20 +400,20 @@ static MODULE: sqlite_nostd::module = sqlite_nostd::module {
     xIntegrity: None,
 };
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
     sqlite::convert_rc(sqlite::create_module_v2(
         db,
         SIMPLE_NAME.as_ptr(),
         &MODULE,
-        null_mut(),
-        None,
+        Arc::into_raw(state.clone()) as *mut c_void,
+        Some(DatabaseState::destroy_arc),
     ))?;
     sqlite::convert_rc(sqlite::create_module_v2(
         db,
         MANUAL_NAME.as_ptr(),
         &MODULE,
-        null_mut(),
-        None,
+        Arc::into_raw(state) as *mut c_void,
+        Some(DatabaseState::destroy_arc),
     ))?;
 
     Ok(())

crates/core/src/lib.rs

@@ -9,9 +9,12 @@ extern crate alloc;
 
 use core::ffi::{c_char, c_int};
 
+use alloc::sync::Arc;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
+use crate::state::DatabaseState;
+
 mod bson;
 mod checkpoint;
 mod crud_vtab;
@@ -26,6 +29,7 @@ mod migrations;
 mod operations;
 mod operations_vtab;
 mod schema;
+mod state;
 mod sync;
 mod sync_local;
 mod util;
@@ -53,6 +57,8 @@ pub extern "C" fn sqlite3_powersync_init(
 }
 
 fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+    let state = Arc::new(DatabaseState::new());
+
     crate::version::register(db)?;
     crate::views::register(db)?;
     crate::uuid::register(db)?;
@@ -62,11 +68,12 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::view_admin::register(db)?;
     crate::checkpoint::register(db)?;
     crate::kv::register(db)?;
-    sync::register(db)?;
+    crate::state::register(db, state.clone())?;
+    sync::register(db, state.clone())?;
 
     crate::schema::register(db)?;
-    crate::operations_vtab::register(db)?;
-    crate::crud_vtab::register(db)?;
+    crate::operations_vtab::register(db, state.clone())?;
+    crate::crud_vtab::register(db, state)?;
 
     Ok(())
 }

crates/core/src/operations_vtab.rs

@@ -1,6 +1,7 @@
 extern crate alloc;
 
 use alloc::boxed::Box;
+use alloc::sync::Arc;
 use core::ffi::{c_char, c_int, c_void};
 
 use sqlite::{Connection, ResultCode, Value};
@@ -9,21 +10,23 @@ use sqlite_nostd as sqlite;
 use crate::operations::{
     clear_remove_ops, delete_bucket, delete_pending_buckets, insert_operation,
 };
+use crate::state::DatabaseState;
 use crate::sync_local::sync_local;
 use crate::vtab_util::*;
 
 #[repr(C)]
 struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
+    state: Arc<DatabaseState>,
 
     target_applied: bool,
     target_validated: bool,
 }
 
 extern "C" fn connect(
     db: *mut sqlite::sqlite3,
-    _aux: *mut c_void,
+    aux: *mut c_void,
     _argc: c_int,
     _argv: *const *const c_char,
     vtab: *mut *mut sqlite::vtab,
@@ -43,6 +46,14 @@ extern "C" fn connect(
                 zErrMsg: core::ptr::null_mut(),
             },
             db,
+            state: {
+                // Increase refcount - we can't use from_raw alone because we don't own the aux
+                // data (connect could be called multiple times).
+                let state = Arc::from_raw(aux as *mut DatabaseState);
+                let clone = state.clone();
+                core::mem::forget(state);
+                clone
+            },
             target_validated: false,
             target_applied: false,
         }));
@@ -83,7 +94,7 @@ extern "C" fn update(
             let result = insert_operation(db, args[3].text());
             vtab_result(vtab, result)
         } else if op == "sync_local" {
-            let result = sync_local(db, &args[3]);
+            let result = sync_local(&tab.state, db, &args[3]);
             if let Ok(result_row) = result {
                 unsafe {
                     *p_row_id = result_row;
@@ -139,8 +150,13 @@ static MODULE: sqlite_nostd::module = sqlite_nostd::module {
     xIntegrity: None,
 };
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
-    db.create_module_v2("powersync_operations", &MODULE, None, None)?;
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    db.create_module_v2(
+        "powersync_operations",
+        &MODULE,
+        Some(Arc::into_raw(state) as *mut c_void),
+        Some(DatabaseState::destroy_arc),
+    )?;
 
     Ok(())
 }

crates/core/src/schema/mod.rs

@@ -5,11 +5,15 @@ use alloc::vec::Vec;
 use serde::Deserialize;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
-pub use table_info::{DiffIncludeOld, Table, TableInfoFlags};
+pub use table_info::{
+    DiffIncludeOld, PendingStatement, PendingStatementValue, RawTable, Table, TableInfoFlags,
+};
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Default)]
 pub struct Schema {
-    tables: Vec<table_info::Table>,
+    pub tables: Vec<table_info::Table>,
+    #[serde(default)]
+    pub raw_tables: Vec<table_info::RawTable>,
 }
 
 pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {

crates/core/src/schema/table_info.rs

@@ -19,6 +19,13 @@ pub struct Table {
     pub flags: TableInfoFlags,
 }
 
+#[derive(Deserialize)]
+pub struct RawTable {
+    pub name: String,
+    pub put: PendingStatement,
+    pub delete: PendingStatement,
+}
+
 impl Table {
     pub fn from_json(text: &str) -> Result<Self, serde_json::Error> {
         serde_json::from_str(text)
@@ -229,3 +236,17 @@ impl<'de> Deserialize<'de> for TableInfoFlags {
         )
     }
 }
+
+#[derive(Deserialize)]
+pub struct PendingStatement {
+    pub sql: String,
+    /// This vec should contain an entry for each parameter in [sql].
+    pub params: Vec<PendingStatementValue>,
+}
+
+#[derive(Deserialize)]
+pub enum PendingStatementValue {
+    Id,
+    Column(String),
+    // TODO: Stuff like a raw object of put data?
+}

crates/core/src/state.rs

@@ -0,0 +1,74 @@
+use core::{
+    ffi::{c_int, c_void},
+    sync::atomic::{AtomicBool, Ordering},
+};
+
+use alloc::sync::Arc;
+use sqlite::{Connection, ResultCode};
+use sqlite_nostd::{self as sqlite, Context};
+
+/// State that is shared for a SQLite database connection after the core extension has been
+/// registered on it.
+///
+/// `init_extension` allocates an instance of this in an `Arc` that is shared as user-data for
+/// functions/vtabs that need access to it.
+pub struct DatabaseState {
+    pub is_in_sync_local: AtomicBool,
+}
+
+impl DatabaseState {
+    pub fn new() -> Self {
+        DatabaseState {
+            is_in_sync_local: AtomicBool::new(false),
+        }
+    }
+
+    pub fn sync_local_guard<'a>(&'a self) -> impl Drop + use<'a> {
+        self.is_in_sync_local
+            .compare_exchange(false, true, Ordering::Acquire, Ordering::Acquire)
+            .expect("should not be syncing already");
+
+        struct ClearOnDrop<'a>(&'a DatabaseState);
+
+        impl Drop for ClearOnDrop<'_> {
+            fn drop(&mut self) {
+                self.0.is_in_sync_local.store(false, Ordering::Release);
+            }
+        }
+
+        ClearOnDrop(self)
+    }
+
+    pub unsafe extern "C" fn destroy_arc(ptr: *mut c_void) {
+        drop(Arc::from_raw(ptr.cast::<DatabaseState>()));
+    }
+}
+
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    unsafe extern "C" fn func(
+        ctx: *mut sqlite::context,
+        _argc: c_int,
+        _argv: *mut *mut sqlite::value,
+    ) {
+        let data = ctx.user_data().cast::<DatabaseState>();
+        let data = unsafe { data.as_ref() }.unwrap();
+
+        ctx.result_int(if data.is_in_sync_local.load(Ordering::Relaxed) {
+            1
+        } else {
+            0
+        });
+    }
+
+    db.create_function_v2(
+        "powersync_in_sync_operation",
+        0,
+        0,
+        Some(Arc::into_raw(state) as *mut c_void),
+        Some(func),
+        None,
+        None,
+        Some(DatabaseState::destroy_arc),
+    )?;
+    Ok(())
+}

crates/core/src/sync/interface.rs

@@ -5,13 +5,16 @@ use alloc::borrow::Cow;
 use alloc::boxed::Box;
 use alloc::rc::Rc;
 use alloc::string::ToString;
+use alloc::sync::Arc;
 use alloc::{string::String, vec::Vec};
 use serde::{Deserialize, Serialize};
 use sqlite::{ResultCode, Value};
 use sqlite_nostd::{self as sqlite, ColumnType};
 use sqlite_nostd::{Connection, Context};
 
 use crate::error::SQLiteError;
+use crate::schema::Schema;
+use crate::state::DatabaseState;
 
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
@@ -22,6 +25,8 @@ pub struct StartSyncStream {
     /// Bucket parameters to include in the request when opening a sync stream.
     #[serde(default)]
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
+    #[serde(default)]
+    pub schema: Schema,
 }
 
 /// A request sent from a client SDK to the [SyncClient] with a `powersync_control` invocation.
@@ -118,7 +123,7 @@ struct SqlController {
     client: SyncClient,
 }
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
     extern "C" fn control(
         ctx: *mut sqlite::context,
         argc: c_int,
@@ -199,7 +204,7 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     }
 
     let controller = Box::new(SqlController {
-        client: SyncClient::new(db),
+        client: SyncClient::new(db, state),
     });
 
     db.create_function_v2(

crates/core/src/sync/mod.rs

@@ -1,3 +1,4 @@
+use alloc::sync::Arc;
 use sqlite_nostd::{self as sqlite, ResultCode};
 
 mod bucket_priority;
@@ -13,6 +14,8 @@ mod sync_status;
 pub use bucket_priority::BucketPriority;
 pub use checksum::Checksum;
 
-pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
-    interface::register(db)
+use crate::state::DatabaseState;
+
+pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(), ResultCode> {
+    interface::register(db, state)
 }