Add simpler crud vtab

simolus3 · simolus3 · commit b28bb395ef28 · 2025-06-11T15:55:19.000+02:00
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
@@ -17,7 +17,7 @@ sqlite_nostd = { workspace=true }
 bytes = { version = "1.4", default-features = false }
 num-traits = { version = "0.2.15", default-features = false }
 num-derive = "0.3"
-serde_json = { version = "1.0", default-features = false, features = ["alloc"] }
+serde_json = { version = "1.0", default-features = false, features = ["alloc", "raw_value"] }
 serde = { version = "1.0", default-features = false, features = ["alloc", "derive", "rc"] }
 const_format = "0.2.34"
 futures-lite = { version = "2.6.0", default-features = false, features = ["alloc"] }
diff --git a/crates/core/src/crud_vtab.rs b/crates/core/src/crud_vtab.rs
@@ -5,25 +5,31 @@ use alloc::string::String;
 use const_format::formatcp;
 use core::ffi::{c_char, c_int, c_void, CStr};
 use core::ptr::null_mut;
+use serde::Serialize;
+use serde_json::value::RawValue;
 
 use sqlite::{Connection, ResultCode, Value};
-use sqlite_nostd as sqlite;
 use sqlite_nostd::ManagedStmt;
-use sqlite_nostd::ResultCode::NULL;
+use sqlite_nostd::{self as sqlite, ColumnType};
 
 use crate::error::SQLiteError;
 use crate::ext::SafeManagedStmt;
 use crate::schema::TableInfoFlags;
+use crate::util::MAX_OP_ID;
 use crate::vtab_util::*;
 
-const MANUAL_NAME: &CStr = c"powersync_crud";
+const MANUAL_NAME: &CStr = c"powersync_crud_";
 const SIMPLE_NAME: &CStr = c"powersync_crud";
 
 // Structure:
 //   CREATE TABLE powersync_crud_(data TEXT, options INT HIDDEN);
+//   CREATE TABLE powersync_crud(op TEXT, id TEXT, data TEXT old_values TEXT, metadata TEXT);
 //
 // This is a insert-only virtual table. It generates transaction ids in ps_tx, and inserts data in
 // ps_crud(tx_id, data).
+// The second form (without the trailing underscore) takes the data to insert as individual
+// components and constructs the data to insert into `ps_crud` internally. It will also update
+// `ps_updated_rows` and the `$local` bucket.
 //
 // Using a virtual table like this allows us to hook into xBegin, xCommit and xRollback to automatically
 // increment transaction ids. These are only called when powersync_crud_ is used as part of a transaction,
@@ -33,8 +39,159 @@ const SIMPLE_NAME: &CStr = c"powersync_crud";
 struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
-    current_tx: Option<i64>,
-    insert_statement: Option<ManagedStmt>,
+    current_tx: Option<ActiveCrudTransaction>,
+    is_simple: bool,
+}
+
+struct ActiveCrudTransaction {
+    tx_id: i64,
+    mode: CrudTransactionMode,
+}
+
+enum CrudTransactionMode {
+    Manual {
+        stmt: ManagedStmt,
+    },
+    Simple {
+        stmt: ManagedStmt,
+        set_updated_rows: ManagedStmt,
+        update_local_bucket: ManagedStmt,
+    },
+}
+
+impl VirtualTable {
+    fn value_to_json<'a>(value: &'a *mut sqlite::value) -> Option<&'a RawValue> {
+        match value.value_type() {
+            ColumnType::Text => {
+                Some(unsafe {
+                    // Safety: RawValue is a transparent type wrapping a str. We assume that it
+                    // contains valid JSON.
+                    core::mem::transmute::<&'a str, &'a RawValue>(value.text())
+                })
+            }
+            _ => None,
+        }
+    }
+
+    fn handle_insert(&self, args: &[*mut sqlite::value]) -> Result<(), SQLiteError> {
+        let current_tx = self
+            .current_tx
+            .as_ref()
+            .ok_or_else(|| SQLiteError(ResultCode::MISUSE, Some(String::from("No tx_id"))))?;
+
+        match &current_tx.mode {
+            CrudTransactionMode::Manual { stmt } => {
+                // Columns are (data TEXT, options INT HIDDEN)
+                let data = args[0].text();
+                let flags = match args[1].value_type() {
+                    sqlite_nostd::ColumnType::Null => TableInfoFlags::default(),
+                    _ => TableInfoFlags(args[1].int() as u32),
+                };
+
+                stmt.bind_int64(1, current_tx.tx_id)?;
+                stmt.bind_text(2, data, sqlite::Destructor::STATIC)?;
+                stmt.bind_int(3, flags.0 as i32)?;
+                stmt.exec()?;
+            }
+            CrudTransactionMode::Simple {
+                stmt,
+                set_updated_rows,
+                update_local_bucket,
+            } => {
+                // Columns are (op TEXT, id TEXT, type TEXT, data TEXT, old_values TEXT, metadata TEXT)
+                let metadata = args[5];
+                let op = args[0].text();
+                let id = args[1].text();
+                let row_type = args[2].text();
+
+                #[derive(Serialize)]
+                struct CrudEntry<'a> {
+                    op: &'a str,
+                    id: &'a str,
+                    #[serde(rename = "type")]
+                    row_type: &'a str,
+                    data: Option<&'a RawValue>,
+                    old: Option<&'a RawValue>,
+                    metadata: Option<&'a str>,
+                }
+
+                // First, we insert into ps_crud like the manual vtab would too. We have to create
+                // the JSON out of the individual components for that.
+                stmt.bind_int64(1, current_tx.tx_id)?;
+                let serialized = serde_json::to_string(&CrudEntry {
+                    op,
+                    id,
+                    row_type,
+                    data: Self::value_to_json(&args[3]),
+                    old: Self::value_to_json(&args[4]),
+                    metadata: if metadata.value_type() == ColumnType::Text {
+                        Some(metadata.text())
+                    } else {
+                        None
+                    },
+                })?;
+                stmt.bind_text(2, &serialized, sqlite::Destructor::STATIC)?;
+                stmt.exec()?;
+
+                // However, we also set ps_updated_rows and update the $local bucket
+                set_updated_rows.bind_text(1, row_type, sqlite::Destructor::STATIC)?;
+                set_updated_rows.bind_text(2, id, sqlite::Destructor::STATIC)?;
+                set_updated_rows.exec()?;
+                update_local_bucket.exec()?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn begin(&mut self) -> Result<(), SQLiteError> {
+        let db = self.db;
+
+        // language=SQLite
+        let statement =
+            db.prepare_v2("UPDATE ps_tx SET next_tx = next_tx + 1 WHERE id = 1 RETURNING next_tx")?;
+        let tx_id = if statement.step()? == ResultCode::ROW {
+            statement.column_int64(0) - 1
+        } else {
+            return Err(SQLiteError::from(ResultCode::ABORT));
+        };
+
+        self.current_tx = Some(ActiveCrudTransaction {
+            tx_id,
+            mode: if self.is_simple {
+                CrudTransactionMode::Simple {
+                    // language=SQLite
+                    stmt: db.prepare_v3("INSERT INTO ps_crud(tx_id, data) VALUES (?, ?)", 0)?,
+                    // language=SQLite
+                    set_updated_rows: db.prepare_v3(
+                        "INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?, ?)",
+                        0,
+                    )?,
+                    update_local_bucket: db.prepare_v3(formatcp!("INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {MAX_OP_ID})"), 0)?,
+                }
+            } else {
+                const SQL: &str = formatcp!(
+                    "\
+WITH insertion (tx_id, data) AS (VALUES (?1, ?2))
+INSERT INTO ps_crud(tx_id, data)
+SELECT * FROM insertion WHERE (NOT (?3 & {})) OR data->>'op' != 'PATCH' OR data->'data' != '{{}}';
+    ",
+                    TableInfoFlags::IGNORE_EMPTY_UPDATE
+                );
+
+                let insert_statement = db.prepare_v3(SQL, 0)?;
+                CrudTransactionMode::Manual {
+                    stmt: insert_statement,
+                }
+            },
+        });
+
+        Ok(())
+    }
+
+    fn end_transaction(&mut self) {
+        self.current_tx = None;
+    }
 }
 
 extern "C" fn connect(
@@ -54,7 +211,7 @@ extern "C" fn connect(
     let is_simple = name == SIMPLE_NAME;
 
     let sql = if is_simple {
-        "CREATE TABLE powersync_crud(op TEXT, id TEXT, data TEXT old_values TEXT, metadata TEXT);"
+        "CREATE TABLE powersync_crud(op TEXT, id TEXT, type TEXT, data TEXT, old_values TEXT, metadata TEXT);"
     } else {
         "CREATE TABLE powersync_crud_(data TEXT, options INT HIDDEN);"
     };
@@ -72,7 +229,7 @@ extern "C" fn connect(
             },
             db,
             current_tx: None,
-            insert_statement: None,
+            is_simple,
         }));
         *vtab = tab.cast::<sqlite::vtab>();
         let _ = sqlite::vtab_config(db, 0);
@@ -87,81 +244,25 @@ extern "C" fn disconnect(vtab: *mut sqlite::vtab) -> c_int {
     ResultCode::OK as c_int
 }
 
-fn begin_impl(tab: &mut VirtualTable) -> Result<(), SQLiteError> {
-    let db = tab.db;
-
-    const SQL: &str = formatcp!(
-        "\
-WITH insertion (tx_id, data) AS (VALUES (?1, ?2))
-INSERT INTO ps_crud(tx_id, data)
-SELECT * FROM insertion WHERE (NOT (?3 & {})) OR data->>'op' != 'PATCH' OR data->'data' != '{{}}';
-    ",
-        TableInfoFlags::IGNORE_EMPTY_UPDATE
-    );
-
-    let insert_statement = db.prepare_v3(SQL, 0)?;
-    tab.insert_statement = Some(insert_statement);
-
-    // language=SQLite
-    let statement =
-        db.prepare_v2("UPDATE ps_tx SET next_tx = next_tx + 1 WHERE id = 1 RETURNING next_tx")?;
-    if statement.step()? == ResultCode::ROW {
-        let tx_id = statement.column_int64(0) - 1;
-        tab.current_tx = Some(tx_id);
-    } else {
-        return Err(SQLiteError::from(ResultCode::ABORT));
-    }
-
-    Ok(())
-}
-
 extern "C" fn begin(vtab: *mut sqlite::vtab) -> c_int {
     let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
-    let result = begin_impl(tab);
+    let result = tab.begin();
     vtab_result(vtab, result)
 }
 
 extern "C" fn commit(vtab: *mut sqlite::vtab) -> c_int {
     let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
-    tab.current_tx = None;
-    tab.insert_statement = None;
+    tab.end_transaction();
     ResultCode::OK as c_int
 }
 
 extern "C" fn rollback(vtab: *mut sqlite::vtab) -> c_int {
     let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
-    tab.current_tx = None;
-    tab.insert_statement = None;
+    tab.end_transaction();
     // ps_tx will be rolled back automatically
     ResultCode::OK as c_int
 }
 
-fn insert_operation(
-    vtab: *mut sqlite::vtab,
-    data: &str,
-    flags: TableInfoFlags,
-) -> Result<(), SQLiteError> {
-    let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
-    if tab.current_tx.is_none() {
-        return Err(SQLiteError(
-            ResultCode::MISUSE,
-            Some(String::from("No tx_id")),
-        ));
-    }
-    let current_tx = tab.current_tx.unwrap();
-    // language=SQLite
-    let statement = tab
-        .insert_statement
-        .as_ref()
-        .ok_or(SQLiteError::from(NULL))?;
-    statement.bind_int64(1, current_tx)?;
-    statement.bind_text(2, data, sqlite::Destructor::STATIC)?;
-    statement.bind_int(3, flags.0 as i32)?;
-    statement.exec()?;
-
-    Ok(())
-}
-
 extern "C" fn update(
     vtab: *mut sqlite::vtab,
     argc: c_int,
@@ -177,12 +278,8 @@ extern "C" fn update(
         ResultCode::MISUSE as c_int
     } else if rowid.value_type() == sqlite::ColumnType::Null {
         // INSERT
-        let data = args[2].text();
-        let flags = match args[3].value_type() {
-            sqlite_nostd::ColumnType::Null => TableInfoFlags::default(),
-            _ => TableInfoFlags(args[3].int() as u32),
-        };
-        let result = insert_operation(vtab, data, flags);
+        let tab = unsafe { &*(vtab.cast::<VirtualTable>()) };
+        let result = tab.handle_insert(&args[2..]);
         vtab_result(vtab, result)
     } else {
         // UPDATE - not supported
diff --git a/dart/test/crud_test.dart b/dart/test/crud_test.dart
