Add table option to skip empty updates

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

crates/core/src/crud_vtab.rs

@@ -5,6 +5,7 @@ use alloc::string::String;
 use core::ffi::{c_char, c_int, c_void};
 use core::slice;
 
+use const_format::formatcp;
 use sqlite::{Connection, ResultCode, Value};
 use sqlite_nostd as sqlite;
 use sqlite_nostd::ManagedStmt;
@@ -15,7 +16,7 @@ use crate::ext::SafeManagedStmt;
 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).
@@ -29,9 +30,13 @@ struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
     current_tx: Option<i64>,
-    insert_statement: Option<ManagedStmt>
+    insert_statement: Option<ManagedStmt>,
 }
 
+#[repr(transparent)]
+#[derive(Clone, Copy)]
+pub struct PowerSyncCrudFlags(pub u32);
+
 extern "C" fn connect(
     db: *mut sqlite::sqlite3,
     _aux: *mut c_void,
@@ -40,8 +45,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;
     }
 
@@ -54,7 +61,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,15 +76,22 @@ 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)?;
+    const SQL: &str = formatcp!("\
+WITH insertion (tx_id, data) AS (VALUES (?1, ?2))
+INSERT INTO ps_crud(tx_id, data)
+SELECT * FROM insertion WHERE (?3 & {}) OR data->>'op' != 'PATCH' OR EXISTS (SELECT 1 FROM json_each(data->'data'));
+    ", PowerSyncCrudFlags::FLAG_INCLUDE_EMPTY_UPDATE);
+
+    // language=SQLite
+    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")?;
+    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);
@@ -110,22 +124,31 @@ extern "C" fn rollback(vtab: *mut sqlite::vtab) -> c_int {
 }
 
 fn insert_operation(
-    vtab: *mut sqlite::vtab, data: &str) -> Result<(), SQLiteError> {
+    vtab: *mut sqlite::vtab,
+    data: &str,
+    flags: PowerSyncCrudFlags,
+) -> 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.bind_int(3, flags.0 as i32)?;
     statement.exec()?;
 
     Ok(())
 }
 
-
 extern "C" fn update(
     vtab: *mut sqlite::vtab,
     argc: c_int,
@@ -142,7 +165,13 @@ 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() {
+            // We don't ignore empty updates by default.
+            sqlite_nostd::ColumnType::Null => PowerSyncCrudFlags::default(),
+            _ => PowerSyncCrudFlags(args[3].int() as u32),
+        };
+
+        let result = insert_operation(vtab, data, flags);
         vtab_result(vtab, result)
     } else {
         // UPDATE - not supported
@@ -185,3 +214,26 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
 
     Ok(())
 }
+
+impl PowerSyncCrudFlags {
+    pub const FLAG_INCLUDE_EMPTY_UPDATE: u32 = 1 << 0;
+
+    pub fn set_include_empty_update(&mut self, value: bool) {
+        if value {
+            self.0 |= Self::FLAG_INCLUDE_EMPTY_UPDATE;
+        } else {
+            self.0 &= !Self::FLAG_INCLUDE_EMPTY_UPDATE;
+        }
+    }
+
+    pub fn has_include_empty_update(self) -> bool {
+        self.0 & Self::FLAG_INCLUDE_EMPTY_UPDATE != 0
+    }
+}
+
+impl Default for PowerSyncCrudFlags {
+    fn default() -> Self {
+        // For backwards-compatibility, we include empty updates by default.
+        return Self(Self::FLAG_INCLUDE_EMPTY_UPDATE);
+    }
+}

crates/core/src/util.rs

@@ -56,7 +56,8 @@ pub fn extract_table_info(
         json_extract(?1, '$.local_only') as local_only,
         json_extract(?1, '$.insert_only') as insert_only,
         json_extract(?1, '$.include_old') as include_old,
-        json_extract(?1, '$.include_metadata') as include_metadata",
+        json_extract(?1, '$.include_metadata') as include_metadata,
+        json_extract(?1, '$.ignore_empty_update') as ignore_empty_update",
     )?;
     statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
 

crates/core/src/views.rs

@@ -10,6 +10,7 @@ use sqlite::{Connection, Context, ResultCode, Value};
 use sqlite_nostd::{self as sqlite, ManagedStmt};
 
 use crate::create_sqlite_text_fn;
+use crate::crud_vtab::PowerSyncCrudFlags;
 use crate::error::{PSResult, SQLiteError};
 use crate::util::*;
 
@@ -215,6 +216,7 @@ fn powersync_trigger_update_sql_impl(
     // TODO: allow accepting a column list
     let include_old = statement.column_type(4)? == sqlite::ColumnType::Text;
     let include_metadata = statement.column_int(5)? != 0;
+    let ignore_empty_update = statement.column_int(6)? != 0;
 
     let quoted_name = quote_identifier(view_name);
     let internal_name = quote_internal_name(name, local_only);
@@ -242,6 +244,9 @@ fn powersync_trigger_update_sql_impl(
         metadata_fragment = "";
     }
 
+    let mut crud_flags: PowerSyncCrudFlags = PowerSyncCrudFlags::default();
+    crud_flags.set_include_empty_update(!ignore_empty_update);
+
     return if !local_only && !insert_only {
         let trigger = format!("\
 CREATE TRIGGER {:}
@@ -255,10 +260,10 @@ BEGIN
   UPDATE {:}
       SET data = {:}
       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({:}, {:})){:}{:}), {:});
   INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES({:}, NEW.id);
   INSERT OR REPLACE INTO ps_buckets(name, last_op, target_op) VALUES('$local', 0, {:});
-END", trigger_name, quoted_name, internal_name, json_fragment_new, type_string, json_fragment_old, json_fragment_new, old_fragment, metadata_fragment, type_string, MAX_OP_ID);
+END", trigger_name, quoted_name, internal_name, json_fragment_new, type_string, json_fragment_old, json_fragment_new, old_fragment, metadata_fragment, crud_flags.0, type_string, MAX_OP_ID);
         Ok(trigger)
     } else if local_only {
         let trigger = format!(

dart/test/crud_test.dart

@@ -65,6 +65,51 @@ void main() {
       expect(r5['diff'], equals('{"b":"test"}'));
     });
 
+    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));
+    });
+
     var runCrudTest = (int numberOfColumns) {
       var columns = [];
       for (var i = 0; i < numberOfColumns; i++) {

dart/test/utils/migration_fixtures.dart

@@ -439,7 +439,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")))), 1);
   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