chore: Add comprehensive merge benchmarks (#71)

benchmark/pubspec.yaml

@@ -14,6 +14,8 @@ dependencies:
   serverpod_database: 3.5.0-beta.9
   serverpod_offline_sync_server:
     path: ../packages/serverpod_offline_sync_server
+  serverpod_offline_sync_shared:
+    path: ../packages/serverpod_offline_sync_shared
   serverpod_offline_sync_test_client:
     path: ../test/serverpod_offline_sync_test_client
 

benchmark/run.dart

@@ -1,5 +1,6 @@
 import 'utils/benchmark.dart';
 import 'utils/conversion.dart';
+import 'utils/merge.dart';
 import 'utils/results.dart';
 import 'utils/runner.dart';
 
@@ -116,6 +117,51 @@ Future<void> main(List<String> args) async {
     runningInCI: runningInCI,
   );
 
+  if (!runningInCI) print('\n${'=' * 60}');
+  print('\n## Merge Benchmark ($rowsString changes)\n');
+  print(
+    [
+      'This benchmark measures the time to apply remote CRDT changes to ',
+      'the local database through the sync merge path.',
+    ].join(runningInCI ? '' : '\n'),
+  );
+  if (!runningInCI) print('-' * 60);
+
+  final mergeBenchmarks = <MergeScenarioBenchmark>[
+    for (final operation in MergeOperation.values)
+      TypesMergeBenchmark(
+        'merge (${operation.name})',
+        operation: operation,
+        changeCount: rowCount,
+      ),
+    UniqueMergeBenchmark('merge (unique conflict)', changeCount: rowCount),
+    for (final operation in FkChainOperation.values)
+      FkChainMergeBenchmark(
+        'merge (fk chain ${operation.name})',
+        operation: operation,
+        changeCount: rowCount,
+      ),
+  ];
+
+  for (final benchmark in mergeBenchmarks) {
+    final mergeResult = await runWithProgress(
+      'Running ${benchmark.name} benchmark',
+      benchmark.measureMerge,
+      skipProgress: runningInCI,
+    );
+
+    printMergeImpact(
+      MergeBenchmarkResults(
+        title: benchmark.resultTitle,
+        batchDescription: benchmark.batchDescription,
+        average: mergeResult.averageMicroseconds,
+        averageQueries: mergeResult.averageQueries,
+        changeCount: benchmark.changesPerBatch,
+      ),
+      runningInCI: runningInCI,
+    );
+  }
+
   print('\n${'-' * 60}');
   print('✅ Benchmark complete!\n');
 }

benchmark/utils/benchmark.dart

@@ -203,7 +203,7 @@ class TypesTableBenchmark extends AsyncBenchmarkBase {
       'offline_sync_benchmark_$name.db',
     );
     _dbFile = File(dbPath);
-    _deleteDatabaseFiles(_dbFile);
+    deleteDatabaseFiles(_dbFile);
 
     _plainSession = await Client(_clientUrl).createSession(
       _dbFile.path,
@@ -234,22 +234,8 @@ class TypesTableBenchmark extends AsyncBenchmarkBase {
     }
   }
 
-  Future<double> _measurePreparedCycles(int minimumMillis) async {
+  Future<double> _measurePreparedCycles(int minimumMillis) =>
-    final minimumMicros = minimumMillis * 1000;
+      measurePreparedCycles(minimumMillis, prepare: _prepareCycle, run: run);
-    final watch = Stopwatch()..start();
-    var totalTimedMicros = 0.0;
-    var timedIterations = 0;
-
-    while (watch.elapsedMicroseconds < minimumMicros) {
-      await _prepareCycle();
-      final sw = Stopwatch()..start();
-      await run();
-      totalTimedMicros += sw.elapsedMicroseconds;
-      timedIterations++;
-    }
-
-    return totalTimedMicros / timedIterations;
-  }
 
   Future<void> _captureDatabaseSize() async {
     try {
@@ -335,7 +321,7 @@ class TypesTableBenchmark extends AsyncBenchmarkBase {
     await _captureDatabaseSize();
     await _plainSession.close();
     _hasOpenSession = false;
-    _deleteDatabaseFiles(_dbFile);
+    deleteDatabaseFiles(_dbFile);
   }
 
   int measureStorage() {
@@ -406,6 +392,30 @@ class TypesTableStorageBenchmark extends TypesTableBenchmark {
   }
 }
 
+/// Repeats untimed [prepare] plus timed [run] cycles until at least
+/// [minimumMillis] of wall time has elapsed, returning the average timed
+/// microseconds per cycle.
+Future<double> measurePreparedCycles(
+  int minimumMillis, {
+  required Future<void> Function() prepare,
+  required Future<void> Function() run,
+}) async {
+  final minimumMicros = minimumMillis * 1000;
+  final watch = Stopwatch()..start();
+  var totalTimedMicros = 0.0;
+  var timedIterations = 0;
+
+  while (watch.elapsedMicroseconds < minimumMicros) {
+    await prepare();
+    final sw = Stopwatch()..start();
+    await run();
+    totalTimedMicros += sw.elapsedMicroseconds;
+    timedIterations++;
+  }
+
+  return totalTimedMicros / timedIterations;
+}
+
 /// Bytes used by SQLite for [mainDb], including WAL sidecars (often omitted by
 /// `mainDb.lengthSync()` alone).
 int _sqliteDbFootprintBytes(File mainDb) {
@@ -421,7 +431,8 @@ int _sqliteDbFootprintBytes(File mainDb) {
   return total;
 }
 
-void _deleteDatabaseFiles(File mainDb) {
+/// Deletes the SQLite database [mainDb] and its WAL sidecar files.
+void deleteDatabaseFiles(File mainDb) {
   if (mainDb.existsSync()) {
     mainDb.deleteSync();
   }

benchmark/utils/merge.dart

@@ -0,0 +1,511 @@
+// Merge benchmark harness — measures the sync merge apply path.
+
+import 'dart:io';
+import 'dart:math';
+import 'dart:typed_data';
+
+import 'package:benchmark_harness/benchmark_harness.dart';
+import 'package:path/path.dart' as p;
+import 'package:serverpod_database/serverpod_database.dart';
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_shared/serverpod_offline_sync_shared.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+
+import 'benchmark.dart';
+import 'conversion.dart';
+import 'query_counter.dart';
+import 'tables.dart';
+
+enum MergeOperation { insert, update, delete, mixed }
+
+enum FkChainOperation { insert, delete }
+
+/// Measurement of a merge scenario, averaged over the timed merge batches.
+typedef MergeMeasurement = ({
+  double averageMicroseconds,
+  double averageQueries,
+});
+
+/// The number of changes of each kind in a mixed batch of [changeCount].
+({int inserts, int updates, int deletes}) mixedMergeComposition(
+  int changeCount,
+) {
+  final updates = changeCount ~/ 3;
+  final deletes = changeCount ~/ 3;
+  return (
+    inserts: changeCount - updates - deletes,
+    updates: updates,
+    deletes: deletes,
+  );
+}
+
+/// Base harness for benchmarks of the merge path ([CrdtDatabase.mergeChanges]).
+///
+/// Changes are attributed to a single simulated remote node whose HLCs always
+/// advance, so every merged change is newer than the local state and is
+/// effectively applied (no stale-change short circuits). All database work is
+/// routed through a [QueryCountingDatabase] so scenarios also report how many
+/// queries each merged batch issues.
+abstract class MergeScenarioBenchmark extends AsyncBenchmarkBase {
+  MergeScenarioBenchmark(super.name);
+
+  static const _warmupMillis = 100;
+  static const _measurementMillis = 2000;
+
+  static const _clientUrl = 'http://localhost:8081/';
+
+  late final File _dbFile;
+  late final ClientDatabaseSession _plainSession;
+  late final QueryCountingDatabase _countingDb;
+  late CrdtDatabaseSession _crdtSession;
+  var _hasOpenSession = false;
+
+  final UuidValue _userId = const Uuid().v7obj();
+
+  /// Clock of the simulated remote node the merged changes come from.
+  Hlc _remoteHlc = Hlc.now(const Uuid().v7obj());
+
+  int _valueSeq = 0;
+
+  CrdtMergeSet _mergeSet = [];
+
+  int _timedQueries = 0;
+  int _timedRuns = 0;
+
+  /// Scenario title used in the results header, e.g. `INSERT`.
+  String get resultTitle;
+
+  /// Number of merge changes in each timed batch.
+  int get changesPerBatch;
+
+  /// Optional batch composition note printed with the results.
+  String? get batchDescription => null;
+
+  /// Builds the next [_mergeSet] (untimed), seeding prerequisite rows if needed.
+  Future<void> prepareCycle();
+
+  /// Seeds state once after the session is created, before the first cycle.
+  Future<void> onSetup() async {}
+
+  Hlc _nextRemoteHlc() => _remoteHlc = _remoteHlc.increment();
+
+  int _nextSeq() => _valueSeq++;
+
+  CrdtMergeInsert insertChangeFor(TableRow<UuidValue?> row) {
+    final hlc = _nextRemoteHlc();
+    return CrdtMergeInsert(
+      tableName: row.table.tableName,
+      uuidRowId: row.id!,
+      uuidNodeId: hlc.nodeId,
+      hlcDatetime: hlc.datetime,
+      hlcCounter: hlc.counter,
+      data: row,
+    );
+  }
+
+  CrdtMergeUpdate updateChangeFor(
+    TableRow<UuidValue?> row,
+    String columnName,
+    Object? value,
+  ) {
+    final hlc = _nextRemoteHlc();
+    return CrdtMergeUpdate(
+      tableName: row.table.tableName,
+      uuidRowId: row.id!,
+      uuidNodeId: hlc.nodeId,
+      hlcDatetime: hlc.datetime,
+      hlcCounter: hlc.counter,
+      columnName: columnName,
+      value: value,
+    );
+  }
+
+  CrdtMergeDelete deleteChangeFor(TableRow<UuidValue?> row) {
+    final hlc = _nextRemoteHlc();
+    return CrdtMergeDelete(
+      tableName: row.table.tableName,
+      uuidRowId: row.id!,
+      uuidNodeId: hlc.nodeId,
+      hlcDatetime: hlc.datetime,
+      hlcCounter: hlc.counter,
+      clFlag: 2,
+      reason: CrdtDataDeletedReason.userDelete,
+    );
+  }
+
+  /// Seeds [rows] through the merge path so they exist locally with remote
+  /// CRDT metadata, like rows previously received from the remote node.
+  Future<void> mergeSeedRows(List<TableRow<UuidValue?>> rows) {
+    return _crdtSession.db.mergeChanges(
+      [for (final row in rows) insertChangeFor(row)],
+      userId: _userId,
+    );
+  }
+
+  @override
+  Future<void> setup() async {
+    _valueSeq = 0;
+    _mergeSet = [];
+    _timedQueries = 0;
+    _timedRuns = 0;
+    final dbPath = p.join(
+      Directory.systemTemp.path,
+      'offline_sync_benchmark_$name.db',
+    );
+    _dbFile = File(dbPath);
+    deleteDatabaseFiles(_dbFile);
+
+    _plainSession = await Client(_clientUrl).createSession(
+      _dbFile.path,
+      isDebugMode: true,
+    );
+    _hasOpenSession = true;
+
+    await clearUserTables(_plainSession);
+    _countingDb = QueryCountingDatabase(_plainSession.db);
+    _crdtSession = CrdtDatabaseSession(
+      _countingDb,
+      syncTables: benchmarkSyncTables,
+    );
+    await _crdtSession.db.initialize();
+
+    await onSetup();
+  }
+
+  Future<MergeMeasurement> measureMerge() async {
+    await setup();
+    try {
+      await measurePreparedCycles(
+        _warmupMillis,
+        prepare: prepareCycle,
+        run: run,
+      );
+      _timedQueries = 0;
+      _timedRuns = 0;
+      final averageMicroseconds = await measurePreparedCycles(
+        _measurementMillis,
+        prepare: prepareCycle,
+        run: run,
+      );
+      return (
+        averageMicroseconds: averageMicroseconds,
+        averageQueries: _timedQueries / _timedRuns,
+      );
+    } finally {
+      await teardown();
+    }
+  }
+
+  @override
+  Future<void> run() async {
+    final queriesBefore = _countingDb.queryCount;
+    await _crdtSession.db.mergeChanges(_mergeSet, userId: _userId);
+    _timedQueries += _countingDb.queryCount - queriesBefore;
+    _timedRuns++;
+  }
+
+  @override
+  Future<void> teardown() async {
+    if (!_hasOpenSession) {
+      return;
+    }
+    await _plainSession.close();
+    _hasOpenSession = false;
+    deleteDatabaseFiles(_dbFile);
+  }
+}
+
+/// Benchmarks plain merge operations using the wide [Types] row, like
+/// [TypesTableBenchmark] does for local mutations.
+class TypesMergeBenchmark extends MergeScenarioBenchmark {
+  TypesMergeBenchmark(
+    super.name, {
+    required this.operation,
+    required this.changeCount,
+  });
+
+  /// Minimal blob column payload (shared; column stays non-null with zero bytes).
+  static final ByteData _emptyBlob = ByteData(0);
+
+  final MergeOperation operation;
+  final int changeCount;
+
+  List<Types> _seededRows = [];
+
+  @override
+  String get resultTitle => operation.name.toUpperCase();
+
+  @override
+  int get changesPerBatch => changeCount;
+
+  @override
+  String? get batchDescription {
+    if (operation != MergeOperation.mixed) return null;
+    final composition = mixedMergeComposition(changeCount);
+    return '${formatter0.format(composition.inserts)} inserts + '
+        '${formatter0.format(composition.updates)} updates + '
+        '${formatter0.format(composition.deletes)} deletes';
+  }
+
+  Types _createTypesRow(int i, DateTime baseTimestamp) {
+    return Types(
+      id: const Uuid().v7obj(),
+      aBool: i.isEven,
+      aDateTime: baseTimestamp.add(Duration(seconds: i)),
+      aText: 'Text $i',
+      anInt: i,
+      anInt64: BigInt.from(i),
+      aReal: i.toDouble(),
+      aBlob: _emptyBlob,
+      anEnum: null,
+      optionalText: null,
+      optionalUuid: const Uuid().v7obj(),
+    );
+  }
+
+  List<Types> _createTypesRows(int count) {
+    final baseTimestamp = DateTime.now();
+    final start = _valueSeq;
+    _valueSeq += count;
+    return List.generate(
+      count,
+      (j) => _createTypesRow(start + j, baseTimestamp),
+    );
+  }
+
+  CrdtMergeUpdate _typesUpdateChangeFor(Types row) {
+    final i = _nextSeq();
+    final (columnName, value) = switch (i % 3) {
+      0 => (Types.t.aText.columnName, 'Updated $i' as Object?),
+      1 => (Types.t.anInt.columnName, i as Object?),
+      _ => (Types.t.aReal.columnName, i.toDouble() as Object?),
+    };
+    return updateChangeFor(row, columnName, value);
+  }
+
+  Future<List<Types>> _mergeInsertRows(int count) async {
+    final rows = _createTypesRows(count);
+    await mergeSeedRows(rows);
+    return rows;
+  }
+
+  @override
+  Future<void> onSetup() async {
+    if (operation == MergeOperation.update) {
+      _seededRows = await _mergeInsertRows(changeCount);
+    }
+  }
+
+  @override
+  Future<void> prepareCycle() async {
+    switch (operation) {
+      case MergeOperation.insert:
+        _mergeSet = [
+          for (final row in _createTypesRows(changeCount)) insertChangeFor(row),
+        ];
+      case MergeOperation.update:
+        _mergeSet = [
+          for (final row in _seededRows) _typesUpdateChangeFor(row),
+        ];
+      case MergeOperation.delete:
+        final rows = await _mergeInsertRows(changeCount);
+        _mergeSet = [for (final row in rows) deleteChangeFor(row)];
+      case MergeOperation.mixed:
+        final composition = mixedMergeComposition(changeCount);
+        final existingRows = await _mergeInsertRows(
+          composition.updates + composition.deletes,
+        );
+        _mergeSet = [
+          for (final row in _createTypesRows(composition.inserts)) insertChangeFor(row),
+          for (final row in existingRows.take(composition.updates))
+            _typesUpdateChangeFor(row),
+          for (final row in existingRows.skip(composition.updates))
+            deleteChangeFor(row),
+        ];
+    }
+  }
+}
+
+/// Benchmarks merging inserts that all collide with existing rows on unique
+/// columns, forcing the unique conflict resolver to release every loser.
+///
+/// Half of the batch collides on a text unique column ([Unique.name], released
+/// with a suffix) and half on a UUID unique column ([UniqueUuid.value],
+/// released with a synthetic UUID).
+class UniqueMergeBenchmark extends MergeScenarioBenchmark {
+  UniqueMergeBenchmark(super.name, {required this.changeCount});
+
+  final int changeCount;
+
+  int get _uuidCount => changeCount ~/ 2;
+  int get _textCount => changeCount - _uuidCount;
+
+  @override
+  String get resultTitle => 'UNIQUE CONFLICT';
+
+  @override
+  int get changesPerBatch => changeCount;
+
+  @override
+  String get batchDescription =>
+      '${formatter0.format(_textCount)} text + '
+      '${formatter0.format(_uuidCount)} uuid conflicting inserts';
+
+  @override
+  Future<void> prepareCycle() async {
+    final textRows = [
+      for (var j = 0; j < _textCount; j++)
+        Unique(id: const Uuid().v7obj(), name: 'name ${_nextSeq()}'),
+    ];
+    final uuidRows = [
+      for (var j = 0; j < _uuidCount; j++)
+        UniqueUuid(id: const Uuid().v7obj(), value: const Uuid().v7obj()),
+    ];
+    await mergeSeedRows([...textRows, ...uuidRows]);
+
+    // New row ids reusing the seeded unique values: every insert conflicts.
+    _mergeSet = [
+      for (final row in textRows)
+        insertChangeFor(Unique(id: const Uuid().v7obj(), name: row.name)),
+      for (final row in uuidRows)
+        insertChangeFor(
+          UniqueUuid(id: const Uuid().v7obj(), value: row.value),
+        ),
+    ];
+  }
+}
+
+/// Benchmarks merges over the FK chain model graph, where foreign key
+/// invariants force existence checks per change and a foreign key projection
+/// pass over the affected tables after the batch.
+///
+/// Each family is 9 related rows spanning cascade, set-null and restrict
+/// `onDelete` actions across three levels:
+///
+/// root ← cascadeMiddle ← restrictBlocker ← {cascade, set-null} children
+///                      ← setNullMiddle ← {cascade, restrict, set-null} children
+class FkChainMergeBenchmark extends MergeScenarioBenchmark {
+  FkChainMergeBenchmark(
+    super.name, {
+    required this.operation,
+    required int changeCount,
+  }) : familyCount = max(1, changeCount ~/ rowsPerFamily);
+
+  /// Rows created by [_createFamily].
+  static const rowsPerFamily = 9;
+
+  /// Timed delete changes per family (root and restrict blocker).
+  static const _deletesPerFamily = 2;
+
+  final FkChainOperation operation;
+  final int familyCount;
+
+  @override
+  String get resultTitle => 'FK CHAIN ${operation.name.toUpperCase()}';
+
+  @override
+  int get changesPerBatch => switch (operation) {
+    FkChainOperation.insert => familyCount * rowsPerFamily,
+    FkChainOperation.delete => familyCount * _deletesPerFamily,
+  };
+
+  String get _familiesLabel =>
+      '${formatter0.format(familyCount)} '
+      '${familyCount == 1 ? 'family' : 'families'}';
+
+  @override
+  String get batchDescription => switch (operation) {
+    FkChainOperation.insert => '$_familiesLabel × $rowsPerFamily related rows',
+    FkChainOperation.delete =>
+      'root + restrict blocker deletes across '
+          '$_familiesLabel of $rowsPerFamily rows',
+  };
+
+  /// Creates one family of related rows in dependency order, so parents are
+  /// merged before the children referencing them.
+  _FkChainFamily _createFamily(int i) {
+    final root = FkChainRoot(id: const Uuid().v7obj(), name: 'root $i');
+    final cascadeMiddle = FkChainCascadeMiddle(
+      id: const Uuid().v7obj(),
+      name: 'cascade middle $i',
+      rootId: root.id,
+    );
+    final restrictBlocker = FkChainRestrictBlocker(
+      id: const Uuid().v7obj(),
+      name: 'restrict blocker $i',
+      cascadeMiddleId: cascadeMiddle.id,
+    );
+    final setNullMiddle = FkChainSetNullMiddle(
+      id: const Uuid().v7obj(),
+      name: 'set-null middle $i',
+      cascadeMiddleId: cascadeMiddle.id,
+    );
+    final rows = <TableRow<UuidValue?>>[
+      root,
+      cascadeMiddle,
+      restrictBlocker,
+      setNullMiddle,
+      FkChainMiddleCascadeChild(
+        id: const Uuid().v7obj(),
+        name: 'middle cascade child $i',
+        restrictBlockerId: restrictBlocker.id,
+      ),
+      FkChainMiddleSetNullChild(
+        id: const Uuid().v7obj(),
+        name: 'middle set-null child $i',
+        restrictBlockerId: restrictBlocker.id,
+      ),
+      FkChainSetNullCascadeChild(
+        id: const Uuid().v7obj(),
+        name: 'set-null cascade child $i',
+        setNullMiddleId: setNullMiddle.id,
+      ),
+      FkChainSetNullRestrictChild(
+        id: const Uuid().v7obj(),
+        name: 'set-null restrict child $i',
+        setNullMiddleId: setNullMiddle.id,
+      ),
+      FkChainSetNullSetNullChild(
+        id: const Uuid().v7obj(),
+        name: 'set-null set-null child $i',
+        setNullMiddleId: setNullMiddle.id,
+      ),
+    ];
+    return (root: root, restrictBlocker: restrictBlocker, rows: rows);
+  }
+
+  List<_FkChainFamily> _createFamilies() {
+    return List.generate(familyCount, (_) => _createFamily(_nextSeq()));
+  }
+
+  @override
+  Future<void> prepareCycle() async {
+    switch (operation) {
+      case FkChainOperation.insert:
+        _mergeSet = [
+          for (final family in _createFamilies())
+            for (final row in family.rows) insertChangeFor(row),
+        ];
+      case FkChainOperation.delete:
+        final families = _createFamilies();
+        await mergeSeedRows([
+          for (final family in families) ...family.rows,
+        ]);
+
+        // Deleting the restrict blocker first unblocks the cascade from the
+        // root through the middle tables, projecting the whole family.
+        _mergeSet = [
+          for (final family in families) ...[
+            deleteChangeFor(family.restrictBlocker),
+            deleteChangeFor(family.root),
+          ],
+        ];
+    }
+  }
+}
+
+typedef _FkChainFamily = ({
+  FkChainRoot root,
+  FkChainRestrictBlocker restrictBlocker,
+  List<TableRow<UuidValue?>> rows,
+});

benchmark/utils/query_counter.dart

@@ -0,0 +1,375 @@
+// Query-counting Database proxy used to report merge query volumes.
+
+import 'package:serverpod_database/serverpod_database.dart';
+
+/// Database proxy that counts every database operation it delegates.
+///
+/// Each counted operation maps to one query issued by the caller (finds,
+/// inserts, updates, deletes, locks and unsafe statements). Transaction
+/// control ([transaction] itself) and metadata getters are not counted.
+class QueryCountingDatabase implements Database {
+  /// Creates a [QueryCountingDatabase] wrapping the [_delegate] database.
+  QueryCountingDatabase(this._delegate);
+
+  final Database _delegate;
+
+  /// Number of database operations delegated so far.
+  int queryCount = 0;
+
+  @override
+  DatabaseAnalyzer get analyzer => _delegate.analyzer;
+
+  @override
+  DatabaseDialect get dialect => _delegate.dialect;
+
+  @override
+  DatabaseSerializationManager get serializationManager =>
+      _delegate.serializationManager;
+
+  @override
+  Future<List<T>> find<T extends TableRow>({
+    Expression? where,
+    int? limit,
+    int? offset,
+    Column? orderBy,
+    List<Column>? orderByList,
+    bool orderDescending = false,
+    Transaction? transaction,
+    Include? include,
+    LockMode? lockMode,
+    LockBehavior? lockBehavior,
+  }) async {
+    queryCount++;
+    return _delegate.find<T>(
+      where: where,
+      limit: limit,
+      offset: offset,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      // Remove this once the deprecated member is removed.
+      // ignore: deprecated_member_use
+      orderDescending: orderDescending,
+      transaction: transaction,
+      include: include,
+      lockMode: lockMode,
+      lockBehavior: lockBehavior,
+    );
+  }
+
+  @override
+  Future<T?> findById<T extends TableRow>(
+    Object id, {
+    Transaction? transaction,
+    Include? include,
+    LockMode? lockMode,
+    LockBehavior? lockBehavior,
+  }) async {
+    queryCount++;
+    return _delegate.findById<T>(
+      id,
+      transaction: transaction,
+      include: include,
+      lockMode: lockMode,
+      lockBehavior: lockBehavior,
+    );
+  }
+
+  @override
+  Future<T?> findFirstRow<T extends TableRow>({
+    Expression? where,
+    int? offset,
+    Column? orderBy,
+    List<Column>? orderByList,
+    bool orderDescending = false,
+    Transaction? transaction,
+    Include? include,
+    LockMode? lockMode,
+    LockBehavior? lockBehavior,
+  }) async {
+    queryCount++;
+    return _delegate.findFirstRow<T>(
+      where: where,
+      offset: offset,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      // Remove this once the deprecated member is removed.
+      // ignore: deprecated_member_use
+      orderDescending: orderDescending,
+      transaction: transaction,
+      include: include,
+      lockMode: lockMode,
+      lockBehavior: lockBehavior,
+    );
+  }
+
+  @override
+  Future<List<T>> insert<T extends TableRow>(
+    List<T> rows, {
+    Transaction? transaction,
+    bool ignoreConflicts = false,
+  }) async {
+    queryCount++;
+    return _delegate.insert<T>(
+      rows,
+      transaction: transaction,
+      ignoreConflicts: ignoreConflicts,
+    );
+  }
+
+  @override
+  Future<T> insertRow<T extends TableRow>(
+    T row, {
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.insertRow<T>(row, transaction: transaction);
+  }
+
+  @override
+  Future<List<T>> upsert<T extends TableRow>(
+    List<T> rows, {
+    required List<Column> conflictColumns,
+    List<Column>? updateColumns,
+    Expression? updateWhere,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.upsert<T>(
+      rows,
+      conflictColumns: conflictColumns,
+      updateColumns: updateColumns,
+      updateWhere: updateWhere,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<T?> upsertRow<T extends TableRow>(
+    T row, {
+    required List<Column> conflictColumns,
+    List<Column>? updateColumns,
+    Expression? updateWhere,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.upsertRow<T>(
+      row,
+      conflictColumns: conflictColumns,
+      updateColumns: updateColumns,
+      updateWhere: updateWhere,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<List<T>> update<T extends TableRow>(
+    List<T> rows, {
+    List<Column>? columns,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.update<T>(rows, columns: columns, transaction: transaction);
+  }
+
+  @override
+  Future<T> updateRow<T extends TableRow>(
+    T row, {
+    List<Column>? columns,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.updateRow<T>(row, columns: columns, transaction: transaction);
+  }
+
+  @override
+  Future<T?> updateById<T extends TableRow>(
+    Object id, {
+    required List<ColumnValue> columnValues,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.updateById<T>(
+      id,
+      columnValues: columnValues,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<List<T>> updateWhere<T extends TableRow>({
+    required List<ColumnValue> columnValues,
+    required Expression where,
+    int? limit,
+    int? offset,
+    Column? orderBy,
+    List<Column>? orderByList,
+    bool orderDescending = false,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.updateWhere<T>(
+      columnValues: columnValues,
+      where: where,
+      limit: limit,
+      offset: offset,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      // Remove this once the deprecated member is removed.
+      // ignore: deprecated_member_use
+      orderDescending: orderDescending,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<List<T>> delete<T extends TableRow>(
+    List<T> rows, {
+    Column? orderBy,
+    List<Column>? orderByList,
+    bool orderDescending = false,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.delete<T>(
+      rows,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      // Remove this once the deprecated member is removed.
+      // ignore: deprecated_member_use
+      orderDescending: orderDescending,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<T> deleteRow<T extends TableRow>(
+    T row, {
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.deleteRow<T>(row, transaction: transaction);
+  }
+
+  @override
+  Future<List<T>> deleteWhere<T extends TableRow>({
+    required Expression where,
+    Column? orderBy,
+    List<Column>? orderByList,
+    bool orderDescending = false,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.deleteWhere<T>(
+      where: where,
+      orderBy: orderBy,
+      orderByList: orderByList,
+      // Remove this once the deprecated member is removed.
+      // ignore: deprecated_member_use
+      orderDescending: orderDescending,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<int> count<T extends TableRow>({
+    Expression? where,
+    int? limit,
+    bool useCache = true,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.count<T>(
+      where: where,
+      limit: limit,
+      useCache: useCache,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<void> lockRows<T extends TableRow>({
+    required Expression where,
+    required LockMode lockMode,
+    required Transaction transaction,
+    LockBehavior lockBehavior = LockBehavior.wait,
+  }) async {
+    queryCount++;
+    return _delegate.lockRows<T>(
+      where: where,
+      lockMode: lockMode,
+      transaction: transaction,
+      lockBehavior: lockBehavior,
+    );
+  }
+
+  @override
+  Future<DatabaseResult> unsafeQuery(
+    String query, {
+    int? timeoutInSeconds,
+    Transaction? transaction,
+    QueryParameters? parameters,
+  }) async {
+    queryCount++;
+    return _delegate.unsafeQuery(
+      query,
+      timeoutInSeconds: timeoutInSeconds,
+      transaction: transaction,
+      parameters: parameters,
+    );
+  }
+
+  @override
+  Future<int> unsafeExecute(
+    String query, {
+    int? timeoutInSeconds,
+    Transaction? transaction,
+    QueryParameters? parameters,
+  }) async {
+    queryCount++;
+    return _delegate.unsafeExecute(
+      query,
+      timeoutInSeconds: timeoutInSeconds,
+      transaction: transaction,
+      parameters: parameters,
+    );
+  }
+
+  @override
+  Future<DatabaseResult> unsafeSimpleQuery(
+    String query, {
+    int? timeoutInSeconds,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.unsafeSimpleQuery(
+      query,
+      timeoutInSeconds: timeoutInSeconds,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<int> unsafeSimpleExecute(
+    String query, {
+    int? timeoutInSeconds,
+    Transaction? transaction,
+  }) async {
+    queryCount++;
+    return _delegate.unsafeSimpleExecute(
+      query,
+      timeoutInSeconds: timeoutInSeconds,
+      transaction: transaction,
+    );
+  }
+
+  @override
+  Future<R> transaction<R>(
+    TransactionFunction<R> transactionFunction, {
+    TransactionSettings? settings,
+  }) async {
+    return _delegate.transaction(transactionFunction, settings: settings);
+  }
+
+  @override
+  Future<bool> testConnection() => _delegate.testConnection();
+}

benchmark/utils/results.dart

@@ -105,6 +105,32 @@ void printStorageImpact(
   if (runningInCI) print('```');
 }
 
+void printMergeImpact(
+  MergeBenchmarkResults results, {
+  bool runningInCI = false,
+}) {
+  final timePerChangeUs = results.average / results.changeCount;
+  final changesPerSecond = Duration.microsecondsPerSecond / timePerChangeUs;
+  final queriesPerChange = results.averageQueries / results.changeCount;
+
+  print(
+    '${runningInCI ? '```' : ''}'
+    '\n🔀 MERGE ${results.title} performance:',
+  );
+  final batchDescription = results.batchDescription;
+  if (batchDescription != null) {
+    print('  Batch: $batchDescription');
+  }
+  print('  Merge time: ${results.average.toFormattedDuration()}');
+  print('  Time per change: ${timePerChangeUs.toFormattedDuration()}');
+  print('  Throughput: ${formatter0.format(changesPerSecond)} changes/s');
+  print(
+    '  Queries: ${formatter0.format(results.averageQueries)} per batch '
+    '(${formatter2.format(queriesPerChange)} per change)',
+  );
+  if (runningInCI) print('```');
+}
+
 class BenchmarkResults {
   const BenchmarkResults({
     required this.operation,
@@ -117,6 +143,26 @@ class BenchmarkResults {
   final double crdt;
 }
 
+class MergeBenchmarkResults {
+  const MergeBenchmarkResults({
+    required this.title,
+    required this.batchDescription,
+    required this.average,
+    required this.averageQueries,
+    required this.changeCount,
+  });
+
+  final String title;
+  final String? batchDescription;
+
+  /// Average microseconds to merge one batch of [changeCount] changes.
+  final double average;
+
+  /// Average number of queries issued while merging one batch.
+  final double averageQueries;
+  final int changeCount;
+}
+
 class StorageBenchmarkResults {
   const StorageBenchmarkResults({
     required this.baseline,

benchmark/utils/tables.dart

@@ -8,11 +8,24 @@ List<Table> get benchmarkSyncTables => [
   Address.t,
   City.t,
   Company.t,
+  FkChainCascadeMiddle.t,
+  FkChainMiddleCascadeChild.t,
+  FkChainMiddleSetNullChild.t,
+  FkChainRestrictBlocker.t,
+  FkChainRoot.t,
+  FkChainSetNullCascadeChild.t,
+  FkChainSetNullMiddle.t,
+  FkChainSetNullRestrictChild.t,
+  FkChainSetNullSetNullChild.t,
   Organization.t,
   Person.t,
+  RequiredSetNullChild.t,
+  RestrictChild.t,
   Town.t,
   Types.t,
   Unique.t,
+  UniqueComposite.t,
+  UniqueSetNullChild.t,
   UniqueUuid.t,
 ];