Add new schema options

packages/powersync_core/lib/src/crud.dart

@@ -3,6 +3,8 @@ import 'dart:convert';
 import 'package:collection/collection.dart';
 import 'package:powersync_core/sqlite3_common.dart' as sqlite;
 
+import 'schema.dart';
+
 /// A batch of client-side changes.
 class CrudBatch {
   /// List of client-side changes.
@@ -68,6 +70,14 @@ class CrudEntry {
   /// ID of the changed row.
   final String id;
 
+  /// An optional metadata string attached to this entry at the time the write
+  /// has been issued.
+  ///
+  /// For tables where [Table.trackMetadata] is enabled, a hidden `_metadata`
+  /// column is added to this table that can be used during updates to attach
+  /// a hint to the update thas is preserved here.
+  final String? metadata;
+
   /// Data associated with the change.
   ///
   /// For PUT, this is contains all non-null columns of the row.
@@ -77,8 +87,22 @@ class CrudEntry {
   /// For DELETE, this is null.
   final Map<String, dynamic>? opData;
 
-  CrudEntry(this.clientId, this.op, this.table, this.id, this.transactionId,
-      this.opData);
+  /// Old values before an update.
+  ///
+  /// This is only tracked for tables for which this has been enabled by setting
+  /// the [Table.trackPreviousValues].
+  final Map<String, dynamic>? oldData;
+
+  CrudEntry(
+    this.clientId,
+    this.op,
+    this.table,
+    this.id,
+    this.transactionId,
+    this.opData, {
+    this.oldData,
+    this.metadata,
+  });
 
   factory CrudEntry.fromRow(sqlite.Row row) {
     final data = jsonDecode(row['data'] as String);
@@ -89,6 +113,8 @@ class CrudEntry {
       data['id'] as String,
       row['tx_id'] as int,
       data['data'] as Map<String, Object?>?,
+      oldData: data['old'] as Map<String, Object?>?,
+      metadata: data['metadata'] as String?,
     );
   }
 
@@ -100,7 +126,9 @@ class CrudEntry {
       'type': table,
       'id': id,
       'tx_id': transactionId,
-      'data': opData
+      'data': opData,
+      'metadata': metadata,
+      'old': oldData,
     };
   }
 

packages/powersync_core/lib/src/schema.dart

@@ -1,3 +1,4 @@
+import 'crud.dart';
 import 'schema_logic.dart';
 
 /// The schema used by the database.
@@ -26,8 +27,29 @@ class Schema {
   }
 }
 
+/// Options to include old values in [CrudEntry] for update statements.
+///
+/// These options are enabled by passing them to a non-local [Table]
+/// constructor.
+final class TrackPreviousValuesOptions {
+  /// A filter of column names for which updates should be tracked.
+  ///
+  /// When set to a non-null value, columns not included in this list will not
+  /// appear in [CrudEntry.oldData]. By default, all columns are included.
+  final List<String>? columnFilter;
+
+  /// Whether to only include old values when they were changed by an update,
+  /// instead of always including all old values.
+  final bool onlyWhenChanged;
+
+  const TrackPreviousValuesOptions(
+      {this.columnFilter, this.onlyWhenChanged = false});
+}
+
 /// A single table in the schema.
 class Table {
+  static const _maxNumberOfColumns = 1999;
+
   /// The synced table name, matching sync rules.
   final String name;
 
@@ -37,20 +59,34 @@ class Table {
   /// List of indexes.
   final List<Index> indexes;
 
-  /// Whether the table only exists only.
+  /// Whether to add a hidden `_metadata` column that will be enabled for
+  /// updates to attach custom information about writes that will be reported
+  /// through [CrudEntry.metadata].
+  final bool trackMetadata;
+
+  /// Whether to track old values of columns for [CrudEntry.oldData].
+  ///
+  /// See [TrackPreviousValuesOptions] for details.
+  final TrackPreviousValuesOptions? trackPreviousValues;
+
+  /// Whether the table only exists locally.
   final bool localOnly;
 
   /// Whether this is an insert-only table.
   final bool insertOnly;
 
+  /// Whether an `UPDATE` statement that doesn't change any values should be
+  /// ignored when creating CRUD entries.
+  final bool ignoreEmptyUpdates;
+
   /// Override the name for the view
   final String? _viewNameOverride;
 
   /// powersync-sqlite-core limits the number of columns
   /// per table to 1999, due to internal SQLite limits.
   ///
   /// In earlier versions this was limited to 63.
-  final int maxNumberOfColumns = 1999;
+  final int maxNumberOfColumns = _maxNumberOfColumns;
 
   /// Internal use only.
   ///
@@ -66,9 +102,16 @@ class Table {
   /// Create a synced table.
   ///
   /// Local changes are recorded, and remote changes are synced to the local table.
-  const Table(this.name, this.columns,
-      {this.indexes = const [], String? viewName, this.localOnly = false})
-      : insertOnly = false,
+  const Table(
+    this.name,
+    this.columns, {
+    this.indexes = const [],
+    String? viewName,
+    this.localOnly = false,
+    this.ignoreEmptyUpdates = false,
+    this.trackMetadata = false,
+    this.trackPreviousValues,
+  })  : insertOnly = false,
         _viewNameOverride = viewName;
 
   /// Create a table that only exists locally.
@@ -78,6 +121,9 @@ class Table {
       {this.indexes = const [], String? viewName})
       : localOnly = true,
         insertOnly = false,
+        trackMetadata = false,
+        trackPreviousValues = null,
+        ignoreEmptyUpdates = false,
         _viewNameOverride = viewName;
 
   /// Create a table that only supports inserts.
@@ -88,8 +134,14 @@ class Table {
   ///
   /// SELECT queries on the table will always return 0 rows.
   ///
-  const Table.insertOnly(this.name, this.columns, {String? viewName})
-      : localOnly = false,
+  const Table.insertOnly(
+    this.name,
+    this.columns, {
+    String? viewName,
+    this.ignoreEmptyUpdates = false,
+    this.trackMetadata = false,
+    this.trackPreviousValues,
+  })  : localOnly = false,
         insertOnly = true,
         indexes = const [],
         _viewNameOverride = viewName;
@@ -106,9 +158,9 @@ class Table {
 
   /// Check that there are no issues in the table definition.
   void validate() {
-    if (columns.length > maxNumberOfColumns) {
+    if (columns.length > _maxNumberOfColumns) {
       throw AssertionError(
-          "Table $name has more than $maxNumberOfColumns columns, which is not supported");
+          "Table $name has more than $_maxNumberOfColumns columns, which is not supported");
     }
 
     if (invalidSqliteCharacters.hasMatch(name)) {
@@ -121,6 +173,14 @@ class Table {
           "Invalid characters in view name: $_viewNameOverride");
     }
 
+    if (trackMetadata && localOnly) {
+      throw AssertionError("Local-only tables can't track metadata");
+    }
+
+    if (trackPreviousValues != null && localOnly) {
+      throw AssertionError("Local-only tables can't track old values");
+    }
+
     Set<String> columnNames = {"id"};
     for (var column in columns) {
       if (column.name == 'id') {
@@ -168,7 +228,13 @@ class Table {
         'local_only': localOnly,
         'insert_only': insertOnly,
         'columns': columns,
-        'indexes': indexes.map((e) => e.toJson(this)).toList(growable: false)
+        'indexes': indexes.map((e) => e.toJson(this)).toList(growable: false),
+        'ignore_empty_update': ignoreEmptyUpdates,
+        'include_metadata': trackMetadata,
+        if (trackPreviousValues case final trackPreviousValues?) ...{
+          'include_old': trackPreviousValues.columnFilter ?? true,
+          'include_old_only_when_changed': trackPreviousValues.onlyWhenChanged,
+        },
       };
 }
 

packages/powersync_core/test/crud_test.dart

@@ -139,6 +139,7 @@ void main() {
 
     test('INSERT-only tables', () async {
       await powersync.disconnectAndClear();
+      await powersync.close();
       powersync = await testUtils.setupPowerSync(
           path: path,
           schema: const Schema([
@@ -269,5 +270,98 @@ void main() {
       await tx2.complete();
       expect(await powersync.getNextCrudTransaction(), equals(null));
     });
+
+    test('include metadata', () async {
+      await powersync.updateSchema(Schema([
+        Table(
+          'lists',
+          [Column.text('name')],
+          trackMetadata: true,
+        )
+      ]));
+
+      await powersync.execute(
+          'INSERT INTO lists (id, name, _metadata) VALUES (uuid(), ?, ?)',
+          ['entry', 'so meta']);
+
+      final batch = await powersync.getNextCrudTransaction();
+      expect(batch!.crud[0].metadata, 'so meta');
+    });
+
+    test('include old values', () async {
+      await powersync.updateSchema(Schema([
+        Table(
+          'lists',
+          [Column.text('name'), Column.text('content')],
+          trackPreviousValues: TrackPreviousValuesOptions(),
+        )
+      ]));
+
+      await powersync.execute(
+          'INSERT INTO lists (id, name, content) VALUES (uuid(), ?, ?)',
+          ['entry', 'content']);
+      await powersync.execute('DELETE FROM ps_crud;');
+      await powersync.execute('UPDATE lists SET name = ?;', ['new name']);
+
+      final batch = await powersync.getNextCrudTransaction();
+      expect(batch!.crud[0].oldData, {'name': 'entry', 'content': 'content'});
+    });
+
+    test('include old values with column filter', () async {
+      await powersync.updateSchema(Schema([
+        Table(
+          'lists',
+          [Column.text('name'), Column.text('content')],
+          trackPreviousValues:
+              TrackPreviousValuesOptions(columnFilter: ['name']),
+        )
+      ]));
+
+      await powersync.execute(
+          'INSERT INTO lists (id, name, content) VALUES (uuid(), ?, ?)',
+          ['name', 'content']);
+      await powersync.execute('DELETE FROM ps_crud;');
+      await powersync.execute('UPDATE lists SET name = ?, content = ?',
+          ['new name', 'new content']);
+
+      final batch = await powersync.getNextCrudTransaction();
+      expect(batch!.crud[0].oldData, {'name': 'name'});
+    });
+
+    test('include old values when changed', () async {
+      await powersync.updateSchema(Schema([
+        Table(
+          'lists',
+          [Column.text('name'), Column.text('content')],
+          trackPreviousValues:
+              TrackPreviousValuesOptions(onlyWhenChanged: true),
+        )
+      ]));
+
+      await powersync.execute(
+          'INSERT INTO lists (id, name, content) VALUES (uuid(), ?, ?)',
+          ['name', 'content']);
+      await powersync.execute('DELETE FROM ps_crud;');
+      await powersync.execute('UPDATE lists SET name = ?', ['new name']);
+
+      final batch = await powersync.getNextCrudTransaction();
+      expect(batch!.crud[0].oldData, {'name': 'name'});
+    });
+
+    test('ignore empty update', () async {
+      await powersync.updateSchema(Schema([
+        Table(
+          'lists',
+          [Column.text('name')],
+          ignoreEmptyUpdates: true,
+        )
+      ]));
+
+      await powersync
+          .execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['name']);
+      await powersync.execute('DELETE FROM ps_crud;');
+      await powersync.execute('UPDATE lists SET name = ?;', ['name']);
+      expect(await powersync.getNextCrudTransaction(), isNull);
+    });
   });
 }