Expand support for update metadata and tracking old values

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "sqlite-rs-embedded"]
 	path = sqlite-rs-embedded
-	url = https://github.com/vlcn-io/sqlite-rs-embedded.git
+	url = https://github.com/powersync-ja/sqlite-rs-embedded.git

crates/core/Cargo.toml

@@ -19,6 +19,7 @@ num-traits = { version = "0.2.15", default-features = false }
 num-derive = "0.3"
 serde_json = { version = "1.0", default-features = false, features = ["alloc"] }
 serde = { version = "1.0", default-features = false, features = ["alloc", "derive"] }
+streaming-iterator = { version = "0.1.9", default-features = false, features = ["alloc"] }
 
 [dependencies.uuid]
 version = "1.4.1"

crates/core/src/checkpoint.rs

@@ -4,7 +4,6 @@ use alloc::format;
 use alloc::string::String;
 use alloc::vec::Vec;
 use core::ffi::c_int;
-use core::slice;
 
 use serde::{Deserialize, Serialize};
 use serde_json as json;
@@ -59,9 +58,9 @@ GROUP BY bucket_list.bucket",
     while statement.step()? == ResultCode::ROW {
         let name = statement.column_text(0)?;
         // checksums with column_int are wrapped to i32 by SQLite
-        let add_checksum = statement.column_int(1)?;
-        let oplog_checksum = statement.column_int(2)?;
-        let expected_checksum = statement.column_int(3)?;
+        let add_checksum = statement.column_int(1);
+        let oplog_checksum = statement.column_int(2);
+        let expected_checksum = statement.column_int(3);
 
         // wrapping add is like +, but safely overflows
         let checksum = oplog_checksum.wrapping_add(add_checksum);

crates/core/src/crud_vtab.rs

@@ -3,7 +3,6 @@ extern crate alloc;
 use alloc::boxed::Box;
 use alloc::string::String;
 use core::ffi::{c_char, c_int, c_void};
-use core::slice;
 
 use sqlite::{Connection, ResultCode, Value};
 use sqlite_nostd as sqlite;
@@ -29,7 +28,7 @@ struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
     current_tx: Option<i64>,
-    insert_statement: Option<ManagedStmt>
+    insert_statement: Option<ManagedStmt>,
 }
 
 extern "C" fn connect(
@@ -40,8 +39,7 @@ extern "C" fn connect(
     vtab: *mut *mut sqlite::vtab,
     _err: *mut *mut c_char,
 ) -> c_int {
-    if let Err(rc) = sqlite::declare_vtab(db, "CREATE TABLE powersync_crud_(data TEXT);")
-    {
+    if let Err(rc) = sqlite::declare_vtab(db, "CREATE TABLE powersync_crud_(data TEXT);") {
         return rc as c_int;
     }
 
@@ -54,7 +52,7 @@ extern "C" fn connect(
             },
             db,
             current_tx: None,
-            insert_statement: None
+            insert_statement: None,
         }));
         *vtab = tab.cast::<sqlite::vtab>();
         let _ = sqlite::vtab_config(db, 0);
@@ -69,17 +67,17 @@ 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;
 
     let insert_statement = db.prepare_v3("INSERT INTO ps_crud(tx_id, data) VALUES (?1, ?2)", 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")?;
+    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;
+        let tx_id = statement.column_int64(0) - 1;
         tab.current_tx = Some(tx_id);
     } else {
         return Err(SQLiteError::from(ResultCode::ABORT));
@@ -109,23 +107,27 @@ extern "C" fn rollback(vtab: *mut sqlite::vtab) -> c_int {
     ResultCode::OK as c_int
 }
 
-fn insert_operation(
-    vtab: *mut sqlite::vtab, data: &str) -> Result<(), SQLiteError> {
+fn insert_operation(vtab: *mut sqlite::vtab, data: &str) -> 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"))));
+        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))?;
+    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.exec()?;
 
     Ok(())
 }
 
-
 extern "C" fn update(
     vtab: *mut sqlite::vtab,
     argc: c_int,
@@ -178,6 +180,7 @@ static MODULE: sqlite_nostd::module = sqlite_nostd::module {
     xRelease: None,
     xRollbackTo: None,
     xShadowName: None,
+    xIntegrity: None,
 };
 
 pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {

crates/core/src/diff.rs

@@ -3,7 +3,6 @@ extern crate alloc;
 use alloc::format;
 use alloc::string::{String, ToString};
 use core::ffi::c_int;
-use core::slice;
 
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
@@ -20,11 +19,20 @@ fn powersync_diff_impl(
 ) -> Result<String, SQLiteError> {
     let data_old = args[0].text();
     let data_new = args[1].text();
+    let ignore_removed = args.get(2).map_or(false, |v| v.int() != 0);
 
-    diff_objects(data_old, data_new)
+    diff_objects_with_options(data_old, data_new, ignore_removed)
 }
 
-pub fn diff_objects(data_old: &str, data_new: &str) -> Result<String, SQLiteError> {
+/// Returns a JSON object containing entries from [data_new] that are not present in [data_old].
+///
+/// When [ignore_removed_columns] is set, columns that are present in [data_old] but not in
+/// [data_new] will not be present in the returned object. Otherwise, they will be set to `null`.
+fn diff_objects_with_options(
+    data_old: &str,
+    data_new: &str,
+    ignore_removed_columns: bool,
+) -> Result<String, SQLiteError> {
     let v_new: json::Value = json::from_str(data_new)?;
     let v_old: json::Value = json::from_str(data_old)?;
 
@@ -39,9 +47,11 @@ pub fn diff_objects(data_old: &str, data_new: &str) -> Result<String, SQLiteErro
         }
 
         // Add missing nulls to left
-        for key in right.keys() {
-            if !left.contains_key(key) {
-                left.insert(key.clone(), json::Value::Null);
+        if !ignore_removed_columns {
+            for key in right.keys() {
+                if !left.contains_key(key) {
+                    left.insert(key.clone(), json::Value::Null);
+                }
             }
         }
 
@@ -76,13 +86,28 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
         None,
     )?;
 
+    db.create_function_v2(
+        "powersync_diff",
+        3,
+        sqlite::UTF8 | sqlite::DETERMINISTIC,
+        None,
+        Some(powersync_diff),
+        None,
+        None,
+        None,
+    )?;
+
     Ok(())
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    fn diff_objects(data_old: &str, data_new: &str) -> Result<String, SQLiteError> {
+        diff_objects_with_options(data_old, data_new, false)
+    }
+
     #[test]
     fn basic_diff_test() {
         assert_eq!(diff_objects("{}", "{}").unwrap(), "{}");

crates/core/src/json_merge.rs

@@ -3,7 +3,6 @@ extern crate alloc;
 use alloc::format;
 use alloc::string::{String, ToString};
 use core::ffi::c_int;
-use core::slice;
 
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;

crates/core/src/kv.rs

@@ -3,7 +3,6 @@ extern crate alloc;
 use alloc::format;
 use alloc::string::{String, ToString};
 use core::ffi::c_int;
-use core::slice;
 
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;

crates/core/src/lib.rs

@@ -25,7 +25,7 @@ mod macros;
 mod migrations;
 mod operations;
 mod operations_vtab;
-mod schema_management;
+mod schema;
 mod sync_local;
 mod sync_types;
 mod util;
@@ -62,7 +62,7 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::checkpoint::register(db)?;
     crate::kv::register(db)?;
 
-    crate::schema_management::register(db)?;
+    crate::schema::register(db)?;
     crate::operations_vtab::register(db)?;
     crate::crud_vtab::register(db)?;