Merge pull request #66 from powersync-ja/skip-empty-updates

Add table option to skip empty updates

Author: simolus3

Cargo.lock

@@ -20,6 +20,7 @@ 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"] }
+const_format = "0.2.34"
 
 [dependencies.uuid]
 version = "1.4.1"

crates/core/Cargo.toml

@@ -2,6 +2,7 @@ extern crate alloc;
 
 use alloc::boxed::Box;
 use alloc::string::String;
+use const_format::formatcp;
 use core::ffi::{c_char, c_int, c_void};
 
 use sqlite::{Connection, ResultCode, Value};
@@ -11,10 +12,11 @@ use sqlite_nostd::ResultCode::NULL;
 
 use crate::error::SQLiteError;
 use crate::ext::SafeManagedStmt;
+use crate::schema::TableInfoFlags;
 use crate::vtab_util::*;
 
 // Structure:
-//   CREATE TABLE powersync_crud_(data TEXT);
+//   CREATE TABLE powersync_crud_(data TEXT, options INT HIDDEN);
 //
 // This is a insert-only virtual table. It generates transaction ids in ps_tx, and inserts data in
 // ps_crud(tx_id, data).
@@ -39,7 +41,10 @@ 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, options INT HIDDEN);",
+    ) {
         return rc as c_int;
     }
 
@@ -70,7 +75,16 @@ extern "C" fn disconnect(vtab: *mut sqlite::vtab) -> 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)?;
+    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
@@ -107,7 +121,11 @@ 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,
+    flags: TableInfoFlags,
+) -> Result<(), SQLiteError> {
     let tab = unsafe { &mut *(vtab.cast::<VirtualTable>()) };
     if tab.current_tx.is_none() {
         return Err(SQLiteError(
@@ -123,6 +141,7 @@ fn insert_operation(vtab: *mut sqlite::vtab, data: &str) -> Result<(), SQLiteErr
         .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(())
@@ -144,7 +163,11 @@ extern "C" fn update(
     } else if rowid.value_type() == sqlite::ColumnType::Null {
         // INSERT
         let data = args[2].text();
-        let result = insert_operation(vtab, data);
+        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);
         vtab_result(vtab, result)
     } else {
         // UPDATE - not supported

crates/core/src/crud_vtab.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/diff.rs

@@ -3,7 +3,9 @@ mod table_info;
 
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
-pub use table_info::{ColumnInfo, ColumnNameAndTypeStatement, DiffIncludeOld, TableInfo};
+pub use table_info::{
+    ColumnInfo, ColumnNameAndTypeStatement, DiffIncludeOld, TableInfo, TableInfoFlags,
+};
 
 pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     management::register(db)

crates/core/src/schema/mod.rs

@@ -28,7 +28,8 @@ impl TableInfo {
         json_extract(?1, '$.insert_only'),
         json_extract(?1, '$.include_old'),
         json_extract(?1, '$.include_metadata'),
-        json_extract(?1, '$.include_old_only_when_changed')",
+        json_extract(?1, '$.include_old_only_when_changed'),
+        json_extract(?1, '$.ignore_empty_update')",
         )?;
         statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
 
@@ -44,6 +45,7 @@ impl TableInfo {
             let insert_only = statement.column_int(3) != 0;
             let include_metadata = statement.column_int(5) != 0;
             let include_old_only_when_changed = statement.column_int(6) != 0;
+            let ignore_empty_update = statement.column_int(7) != 0;
 
             let mut flags = TableInfoFlags::default();
             flags = flags.set_flag(TableInfoFlags::LOCAL_ONLY, local_only);
@@ -53,7 +55,7 @@ impl TableInfo {
                 TableInfoFlags::INCLUDE_OLD_ONLY_WHEN_CHANGED,
                 include_old_only_when_changed,
             );
-
+            flags = flags.set_flag(TableInfoFlags::IGNORE_EMPTY_UPDATE, ignore_empty_update);
             flags
         };
 
@@ -98,13 +100,14 @@ pub enum DiffIncludeOld {
 
 #[derive(Clone, Copy)]
 #[repr(transparent)]
-pub struct TableInfoFlags(u32);
+pub struct TableInfoFlags(pub u32);
 
 impl TableInfoFlags {
     pub const LOCAL_ONLY: u32 = 1;
     pub const INSERT_ONLY: u32 = 2;
     pub const INCLUDE_METADATA: u32 = 4;
     pub const INCLUDE_OLD_ONLY_WHEN_CHANGED: u32 = 8;
+    pub const IGNORE_EMPTY_UPDATE: u32 = 16;
 
     pub const fn local_only(self) -> bool {
         self.0 & Self::LOCAL_ONLY != 0

crates/core/src/schema/table_info.rs

@@ -339,6 +339,8 @@ fn powersync_trigger_update_sql_impl(
             ""
         };
 
+        let flags = table_info.flags.0;
+
         let trigger = format!("\
 CREATE TRIGGER {trigger_name}
 INSTEAD OF UPDATE ON {quoted_name}
@@ -351,7 +353,7 @@ BEGIN
   UPDATE {internal_name}
       SET data = {json_fragment_new}
       WHERE id = NEW.id;
-  INSERT INTO powersync_crud_(data) VALUES(json_object('op', 'PATCH', 'type', {:}, 'id', NEW.id, 'data', json(powersync_diff({:}, {:})){:}{:}));
+  INSERT INTO powersync_crud_(data, options) VALUES(json_object('op', 'PATCH', 'type', {:}, 'id', NEW.id, 'data', json(powersync_diff({:}, {:})){:}{:}), {flags});
   INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES({type_string}, NEW.id);
   INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {MAX_OP_ID});
 END", type_string, json_fragment_old, json_fragment_new, old_fragment, metadata_fragment);

crates/core/src/views.rs

@@ -473,5 +473,50 @@ void main() {
         expect(op['metadata'], 'custom delete');
       });
     });
+
+    test('includes empty updates by default', () {
+      db
+        ..execute('select powersync_replace_schema(?)', [
+          json.encode({
+            'tables': [
+              {
+                'name': 'items',
+                'columns': [
+                  {'name': 'col', 'type': 'text'}
+                ],
+              }
+            ]
+          })
+        ])
+        ..execute(
+            'INSERT INTO items (id, col) VALUES (uuid(), ?)', ['new item'])
+        ..execute('UPDATE items SET col = LOWER(col)');
+
+      // Should record insert and update operation.
+      expect(db.select('SELECT * FROM ps_crud'), hasLength(2));
+    });
+
+    test('can ignore empty updates', () {
+      db
+        ..execute('select powersync_replace_schema(?)', [
+          json.encode({
+            'tables': [
+              {
+                'name': 'items',
+                'columns': [
+                  {'name': 'col', 'type': 'text'}
+                ],
+                'ignore_empty_update': true,
+              }
+            ]
+          })
+        ])
+        ..execute(
+            'INSERT INTO items (id, col) VALUES (uuid(), ?)', ['new item'])
+        ..execute('UPDATE items SET col = LOWER(col)');
+
+      // The update which didn't change any rows should not be recorded.
+      expect(db.select('SELECT * FROM ps_crud'), hasLength(1));
+    });
   });
 }

dart/test/crud_test.dart

@@ -558,7 +558,7 @@ BEGIN
   UPDATE "ps_data__lists"
       SET data = json_object('description', NEW."description")
       WHERE id = NEW.id;
-  INSERT INTO powersync_crud_(data) VALUES(json_object('op', 'PATCH', 'type', 'lists', 'id', NEW.id, 'data', json(powersync_diff(json_object('description', OLD."description"), json_object('description', NEW."description")))));
+  INSERT INTO powersync_crud_(data, options) VALUES(json_object('op', 'PATCH', 'type', 'lists', 'id', NEW.id, 'data', json(powersync_diff(json_object('description', OLD."description"), json_object('description', NEW."description")))), 0);
   INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES('lists', NEW.id);
   INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, 9223372036854775807);
 END