perf: Reduce redundant projection and metadata reads (#124)

packages/serverpod_offline_sync/lib/src/database/database.dart

@@ -201,6 +201,8 @@ class CrdtDatabase implements Database {
     Transaction? transaction, {
     required bool membershipWide,
   }) async {
+    if (include == null && !_recorder.isCrdtTracked<T>()) return where;
+
     final scopeIds = await _scopeIdsForQueries(
       transaction,
       membershipWide: membershipWide,
@@ -429,18 +431,18 @@ class CrdtDatabase implements Database {
       transaction,
       (tx) async {
         final prepared = _prepareRowsForInsert(rows, tx);
-        if (rows.isNotEmpty) {
-          final rowIds = {
-            for (final row in rows)
-              if (row.id case final UuidValue rowId) rowId,
-          };
-          await _recorder.projectCurrent(rows.first.table.tableName, rowIds, tx);
-        }
+        final values = _recorder.withForeignKeyInsertDefaults(prepared.rows);
+        final projectionUnchanged = await _recorder.prepareLocalUpsert(
+          values,
+          conflictColumns,
+          updateColumns,
+          tx,
+        );
 
         // CRDT metadata needs the affected rows even when the public call uses
         // noReturn, because inserted rows may have database-generated ids.
         final result = await _delegate.upsert<T>(
-          _recorder.withForeignKeyInsertDefaults(prepared.rows),
+          values,
           conflictColumns: conflictColumns,
           updateColumns: updateColumns,
           updateWhere: await _whereVisibleWithTombstone<T>(
@@ -469,7 +471,12 @@ class CrdtDatabase implements Database {
           for (final row in result)
             if (row.id is! UuidValue || !insertedRowIds.contains(row.id)) row,
         ];
-        await _recorder.afterUpdate(updatedRows, updateColumns, tx);
+        await _recorder.afterUpdate(
+          updatedRows,
+          updateColumns,
+          tx,
+          projectionUnchanged: projectionUnchanged,
+        );
         if (noReturn) return <T>[];
         _stripStampedRows(result, prepared);
         for (final row in reinsertedRows) {
@@ -519,7 +526,7 @@ class CrdtDatabase implements Database {
       transaction,
     );
     final reinsertedRows = await _delegate.update<T>(
-      plannedReinserts,
+      plannedReinserts.rows,
       transaction: transaction,
     );
     await _recorder.afterReinsert(reinsertedRows, transaction);
@@ -623,7 +630,7 @@ class CrdtDatabase implements Database {
       (tx) async {
         final plannedUpdates = await _recorder.planLocalUpdates(rows, columns, tx);
         final updatedRows = [
-          for (final row in plannedUpdates)
+          for (final row in plannedUpdates.rows)
             await _updateRowWithoutRecording(
               row,
               stripScopeId: _shouldStripReturnedScopeId(row, tx),
@@ -632,7 +639,12 @@ class CrdtDatabase implements Database {
             ),
         ];
 
-        await _recorder.afterUpdate(updatedRows, columns, tx);
+        await _recorder.afterUpdate(
+          updatedRows,
+          columns,
+          tx,
+          projectionUnchanged: plannedUpdates.projectionUnchanged,
+        );
         return noReturn ? <T>[] : updatedRows;
       },
     );
@@ -652,13 +664,18 @@ class CrdtDatabase implements Database {
         final plannedUpdates = await _recorder.planLocalUpdates([row], columns, tx);
         final stripScopeId = _shouldStripReturnedScopeId(row, tx);
         final updatedRow = await _updateRowWithoutRecording(
-          plannedUpdates.single,
+          plannedUpdates.rows.single,
           stripScopeId: stripScopeId,
           transaction: tx,
           columns: columns,
         );
 
-        await _recorder.afterUpdate([updatedRow], columns, tx);
+        await _recorder.afterUpdate(
+          [updatedRow],
+          columns,
+          tx,
+          projectionUnchanged: plannedUpdates.projectionUnchanged,
+        );
         return updatedRow;
       },
     );
@@ -980,9 +997,23 @@ class CrdtDatabase implements Database {
 
     // On the server this is authoritative membership; on a persistent client it
     // is the server-projected membership cache.
-    return _scopeIdsForUuids(
-      await CrdtScopeMembership.memberScopes(_delegate.session, userId),
-    );
+    final scopeGroups = await Future.wait<List<CrdtScope>>([
+      CrdtScope.db.find(
+        _delegate.session,
+        where: (t) => t.uuidScopeId.equals(userId),
+      ),
+      CrdtScopeMember.db
+          .find(
+            _delegate.session,
+            where: (t) => t.userUuid.equals(userId),
+            include: CrdtScopeMember.include(scope: CrdtScope.include()),
+          )
+          .then((memberships) => [for (final member in memberships) member.scope!]),
+    ]);
+    return {
+      for (final scopes in scopeGroups)
+        for (final scope in scopes) scope.id!,
+    }.toList();
   }
 
   List<int>? _actingScopeIdsForQueries(Transaction? transaction) {
@@ -998,18 +1029,6 @@ class CrdtDatabase implements Database {
     return _recorder.persistentUserId;
   }
 
-  Future<List<int>> _scopeIdsForUuids(List<UuidValue> scopeUuids) async {
-    if (scopeUuids.isEmpty) return [];
-    final scopes = await CrdtScope.db.find(
-      _delegate.session,
-      where: (t) => t.uuidScopeId.inSet(scopeUuids.toSet()),
-    );
-    return [
-      for (final scope in scopes)
-        if (scope.id != null) scope.id!,
-    ];
-  }
-
   Future<UuidValue> _requireUserId(UuidValue? userId) async {
     return userId ??
         _recorder.persistentUserId ??

packages/serverpod_offline_sync/lib/src/database/merge_utils/foreign_key_graph.dart

@@ -155,6 +155,13 @@ class CrdtForeignKeyGraph {
     return reached;
   }
 
+  /// Components whose projection can load a default even when every seed is absent.
+  late final Set<String> tablesWithDefaultDependencies = {
+    for (final edge in edges)
+      if (edge.action == ForeignKeyAction.setDefault && edge.defaultValue != null)
+        ...connectedTables([edge.childTableName]),
+  };
+
   /// The child column names of foreign keys outgoing from [tableName].
   ///
   /// When [columnNames] is provided, only edges whose child column is in the set

packages/serverpod_offline_sync/lib/src/database/merge_utils/foreign_key_projector.dart

@@ -636,6 +636,130 @@ class CrdtForeignKeyProjector {
     };
   }
 
+  /// Proves that an ordinary local write leaves the existing projection fixed.
+  ///
+  /// Only new, unreferenced rows and additions to null FKs qualify. All parents
+  /// must already be live, no unique claim may be contested, and no authored
+  /// projection override may be pending on the table. Removing an edge, changing
+  /// a claim, restoring a row or creating a default target needs the full pass.
+  /// The proof is checked again after the write; it is never a transaction cache.
+  Future<bool> canLeaveLocalProjectionUnchanged<T extends TableRow>(
+    List<T> candidates,
+    Transaction transaction, {
+    required bool inserting,
+    bool persisted = true,
+    List<Column>? columns,
+  }) async {
+    if (candidates.isEmpty) return true;
+    final tableName = candidates.first.table.tableName;
+    if (!_context.isCrdtTrackedTableName(tableName)) return true;
+    if (candidates.any((row) => row.id is! UuidValue)) return false;
+    final ids = candidates.uuidRowIds;
+    if (ids.length != candidates.length) return false;
+    final childEdges = _foreignKeys.edgesByChildTable[tableName] ?? <ForeignKeyEdge>[];
+    final parentEdges =
+        _foreignKeys.edgesByParentTable[tableName] ?? <ForeignKeyEdge>[];
+    if (childEdges.any((edge) => edge.parentColumn != 'id')) return false;
+    if (inserting && parentEdges.any((edge) => edge.parentColumn != 'id')) return false;
+
+    final uniqueColumns = _uniqueResolver.uniqueColumnNamesFor(tableName);
+    final relevantColumns = {
+      for (final edge in childEdges) edge.childColumn,
+      for (final edge in parentEdges)
+        if (edge.parentColumn != 'id') edge.parentColumn,
+      ...uniqueColumns,
+    };
+    final current = inserting
+        ? <UuidValue, Map<String, Object?>>{}
+        : await _context.readDomainColumnValues(
+            tableName,
+            ids,
+            relevantColumns.toList(),
+            transaction,
+          );
+    final writtenColumns = columns?.map((column) => column.columnName).toSet();
+    final values = <UuidValue, Map<String, Object?>>{};
+    for (final candidate in candidates) {
+      final id = candidate.id as UuidValue;
+      final supplied = candidate.toJsonForDatabase() as Map;
+      final previous = current[id] ?? const <String, Object?>{};
+      final next = {
+        for (final column in relevantColumns)
+          column: canonicalDomainValue(
+            inserting || writtenColumns == null || writtenColumns.contains(column)
+                ? supplied[column]
+                : previous[column],
+          ),
+      };
+      if (!inserting) {
+        if (relevantColumns.isNotEmpty && !current.containsKey(id)) return false;
+        for (final column in relevantColumns) {
+          if (projectionValuesEqual(previous[column], next[column])) continue;
+          if (previous[column] != null ||
+              uniqueColumns.contains(column) ||
+              !childEdges.any((edge) => edge.childColumn == column) ||
+              parentEdges.any((edge) => edge.parentColumn == column)) {
+            return false;
+          }
+        }
+      }
+      values[id] = next;
+    }
+
+    final parentsByTable = <String, Set<UuidValue>>{};
+    for (final edge in childEdges) {
+      for (final row in values.values) {
+        final value = row[edge.childColumn];
+        if (value == null) {
+          if (!edge.childNullable || (!persisted && edge.defaultValue != null)) {
+            return false;
+          }
+          continue;
+        }
+        final id = tryUuidValue(value);
+        if (id == null || (edge.parentTableName == tableName && ids.contains(id))) {
+          return false;
+        }
+        (parentsByTable[edge.parentTableName] ??= {}).add(id);
+      }
+    }
+    final referencingColumnsByTable = <String, Set<String>>{};
+    if (inserting) {
+      for (final edge in parentEdges) {
+        if (edge.action == ForeignKeyAction.setDefault &&
+            ids.contains(tryUuidValue(edge.defaultValue))) {
+          return false;
+        }
+        (referencingColumnsByTable[edge.childTableName] ??= {}).add(edge.childColumn);
+      }
+    }
+
+    final uniqueClaims = <Map<String, Set<Object?>>>[];
+    for (final index in _uniqueResolver.uniqueIndexesFor(tableName)) {
+      final tuples = <String>{};
+      final claims = <String, Set<Object?>>{};
+      for (final row in values.values) {
+        final tuple = [for (final column in index.indexedColumns) row[column]];
+        if (tuple.any((value) => value == null)) continue;
+        if (!tuples.add(tuple.map(canonicalProjectionValue).join('\x1f'))) return false;
+        for (final column in index.indexedColumns) {
+          (claims[column] ??= {}).add(row[column]);
+        }
+      }
+      if (claims.isEmpty) continue;
+      uniqueClaims.add(claims);
+    }
+    return _context.localProjectionDependenciesAreStable(
+      tableName: tableName,
+      rowIds: ids,
+      persisted: persisted,
+      parentsByTable: parentsByTable,
+      referencingColumnsByTable: referencingColumnsByTable,
+      uniqueClaims: uniqueClaims,
+      transaction: transaction,
+    );
+  }
+
   /// Whether a mutation of [tableName] may require FK or unique projection.
   bool needsProjection(String tableName, Set<String>? columnNames) {
     if (_foreignKeys.columnsMayAffectForeignKeys(tableName, columnNames)) {

packages/serverpod_offline_sync/lib/src/database/merge_utils/recorder_context.dart

@@ -148,6 +148,95 @@ class CrdtRecorderContext {
     return crdtDataRows;
   }
 
+  /// Checks local projection dependencies in one database snapshot.
+  Future<bool> localProjectionDependenciesAreStable({
+    required String tableName,
+    required Set<UuidValue> rowIds,
+    required bool persisted,
+    required Map<String, Set<UuidValue>> parentsByTable,
+    required Map<String, Set<String>> referencingColumnsByTable,
+    required List<Map<String, Set<Object?>>> uniqueClaims,
+    required Transaction transaction,
+  }) async {
+    final (tableId, _) = schema[tableName]!;
+    final scopeId = hlcManagerFor(transaction).normalizedScopeId;
+    final ids = rowIds.sqlLiteralList();
+    final rowPredicate =
+        'r."scopeId" = $scopeId AND r."tblId" = $tableId '
+        'AND r."uuidRowId" IN ($ids)';
+    final conditions = <String>[
+      if (persisted)
+        '''(SELECT COUNT(*) FROM "crdt_data_rows" r
+LEFT JOIN "crdt_data_tombstone" d ON d."rowId" = r."id"
+WHERE $rowPredicate AND ($_rowVisible)
+  AND (d."id" IS NULL OR d."clFlag" % 2 = 1)) = ${rowIds.length}'''
+      else
+        'NOT EXISTS (SELECT 1 FROM "crdt_data_rows" r WHERE $rowPredicate)',
+      // An attempted FK may own a unique fallback different from its authored value.
+      '''NOT EXISTS (
+SELECT 1 FROM "crdt_data_attempted_value" a
+JOIN "crdt_data_fields" f ON f."id" = a."fieldId"
+JOIN "crdt_data_rows" r ON r."id" = f."rowId"
+WHERE r."scopeId" = $scopeId AND r."tblId" = $tableId)''',
+    ];
+    for (final MapEntry(key: parentTable, value: parentIds) in parentsByTable.entries) {
+      final (parentTableId, _) = schema[parentTable]!;
+      conditions.add('''(SELECT COUNT(*)
+FROM "${parentTable.escapeIdentifier()}" p
+JOIN "crdt_data_rows" r ON r."uuidRowId" = p."id"
+  AND r."tblId" = $parentTableId AND r."scopeId" = $scopeId
+LEFT JOIN "crdt_data_tombstone" d ON d."rowId" = r."id"
+WHERE p."scopeId" = $scopeId AND p."id" IN (${parentIds.sqlLiteralList()})
+  AND ($_rowVisible) AND (d."id" IS NULL OR d."clFlag" % 2 = 1)
+) = ${parentIds.length}''');
+    }
+    final encodedIds = {
+      for (final id in rowIds)
+        ValueEncoder.instance.encodeColumnValue(
+          CrdtDataAttemptedValue.t.value,
+          Protocol().dynamicFieldToJson(id),
+        ),
+    };
+    for (final MapEntry(key: childTable, value: columnNames)
+        in referencingColumnsByTable.entries) {
+      final predicates = [
+        for (final column in columnNames) '"${column.escapeIdentifier()}" IN ($ids)',
+      ].join(' OR ');
+      conditions.add(
+        'NOT EXISTS (SELECT 1 FROM "${childTable.escapeIdentifier()}" '
+        'WHERE $predicates)',
+      );
+      final (childTableId, columns) = schema[childTable]!;
+      final columnIds = {
+        for (final name in columnNames) ?columns[name]?.id,
+      };
+      if (columnIds.isEmpty) continue;
+      conditions.add('''NOT EXISTS (
+SELECT 1 FROM "crdt_data_attempted_value" a
+JOIN "crdt_data_fields" f ON f."id" = a."fieldId"
+JOIN "crdt_data_rows" r ON r."id" = f."rowId"
+WHERE r."scopeId" = $scopeId AND r."tblId" = $childTableId
+  AND f."columnId" IN (${columnIds.join(', ')})
+  AND a."value" IN (${encodedIds.join(', ')}))''');
+    }
+    for (final claims in uniqueClaims) {
+      final predicates = [
+        for (final MapEntry(key: column, value: values) in claims.entries)
+          '("${column.escapeIdentifier()}" IN (${values.sqlLiteralList()}))',
+      ].join(' AND ');
+      conditions.add(
+        'NOT EXISTS (SELECT 1 FROM "${tableName.escapeIdentifier()}" '
+        'WHERE $predicates AND "id" NOT IN ($ids))',
+      );
+    }
+    final result = await database.unsafeQuery(
+      'SELECT ${conditions.map((condition) => "($condition)").join(" AND ")}',
+      transaction: transaction,
+    );
+    final stable = result.single.first;
+    return stable == true || stable == 1;
+  }
+
   Future<void> upsertCrdtFieldsForRows(
     String tableName,
     List<CrdtDataRow> crdtDataRows,
@@ -157,66 +246,34 @@ class CrdtRecorderContext {
   }) async {
     if (crdtDataRows.isEmpty || schemaColumns.isEmpty) return;
 
-    final rowPks = crdtDataRows.map((r) => r.id!).toSet();
-    final columnPks = schemaColumns.map((c) => c.id!).toSet();
-    final existingFields = await CrdtDataField.db.find(
-      databaseSession,
-      where: (t) => t.rowId.inSet(rowPks) & t.columnId.inSet(columnPks),
-      transaction: transaction,
-    );
-
-    final fieldByRowAndColumn = {
-      for (final f in existingFields) (f.rowId, f.columnId): f,
-    };
-
     final hlcManager = hlcManagerFor(transaction);
-    final toInsert = <CrdtDataField>[];
-    final toUpdate = <CrdtDataField>[];
-
+    final fields = <CrdtDataField>[];
     for (final row in crdtDataRows) {
       for (final schemaCol in schemaColumns) {
         if (skippedFields.contains((tableName, row.uuidRowId, schemaCol.name))) {
           continue;
         }
-
         final hlc = hlcManager.increment();
-        final existing = fieldByRowAndColumn[(row.id!, schemaCol.id!)];
-        if (existing == null) {
-          toInsert.add(
-            CrdtDataField(
-              rowId: row.id!,
-              columnId: schemaCol.id!,
-              nodeId: hlcManager.normalizedNodeId,
-              hlcDatetime: hlc.datetime,
-              hlcCounter: hlc.counter,
-            ),
-          );
-        } else {
-          toUpdate.add(
-            existing.copyWith(
-              nodeId: hlcManager.normalizedNodeId,
-              hlcDatetime: hlc.datetime,
-              hlcCounter: hlc.counter,
-            ),
-          );
-        }
+        fields.add(
+          CrdtDataField(
+            rowId: row.id!,
+            columnId: schemaCol.id!,
+            nodeId: hlcManager.normalizedNodeId,
+            hlcDatetime: hlc.datetime,
+            hlcCounter: hlc.counter,
+          ),
+        );
       }
     }
-
-    if (toInsert.isNotEmpty) {
-      await CrdtDataField.db.insert(
-        databaseSession,
-        toInsert,
-        transaction: transaction,
-      );
-    }
-    if (toUpdate.isNotEmpty) {
-      await CrdtDataField.db.update(
-        databaseSession,
-        toUpdate,
-        transaction: transaction,
-      );
-    }
+    if (fields.isEmpty) return;
+    await CrdtDataField.db.upsert(
+      databaseSession,
+      fields,
+      conflictColumns: (t) => [t.rowId, t.columnId],
+      updateColumns: (t) => [t.nodeId, t.hlcDatetime, t.hlcCounter],
+      transaction: transaction,
+      noReturn: true,
+    );
   }
 
   Future<void> recordFieldsUpdatedByTable(

packages/serverpod_offline_sync/lib/src/database/recorder.dart

@@ -372,6 +372,42 @@ class CrdtMutationRecorder {
     );
   }
 
+  /// Returns whether primary-key upserts preserve the existing projection.
+  Future<bool> prepareLocalUpsert<T extends TableRow>(
+    List<T> rows,
+    List<Column> conflictColumns,
+    List<Column>? updateColumns,
+    Transaction transaction,
+  ) async {
+    final tableName = rows.first.table.tableName;
+    final rowIds = {
+      for (final row in rows)
+        if (row.id case final UuidValue rowId) rowId,
+    };
+    if (conflictColumns.length == 1 &&
+        conflictColumns.single.columnName == 'id' &&
+        rowIds.length == rows.length) {
+      final stored = await _context.findCrdtRows(tableName, rowIds, transaction);
+      if (stored.isEmpty &&
+          !_foreignKeys.tablesWithDefaultDependencies.contains(tableName)) {
+        // The full pass has no persisted seed or default to load. Classification
+        // after the physical upsert still reads metadata again.
+        return false;
+      }
+      if (stored.length == rowIds.length &&
+          await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
+            rows,
+            transaction,
+            inserting: false,
+            columns: updateColumns,
+          )) {
+        return true;
+      }
+    }
+    await projectCurrent(tableName, rowIds, transaction);
+    return false;
+  }
+
   /// Plans FK/unique projection for rows that are about to be inserted.
   ///
   /// Releases hidden unique claims first and returns copies whose unique/FK
@@ -392,6 +428,15 @@ class CrdtMutationRecorder {
       return (rows: rows, attempts: unplanned);
     }
 
+    if (await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
+      rows,
+      transaction,
+      inserting: true,
+      persisted: false,
+    )) {
+      return (rows: rows, attempts: unplanned);
+    }
+
     final hlcManager = _context.hlcManagerFor(transaction);
     final pendingHlc = hlcManager.peekNext();
     final node = hlcManager.getNode();
@@ -447,17 +492,29 @@ class CrdtMutationRecorder {
   }
 
   /// Plans FK/unique projection for rows that are about to be updated.
-  Future<List<T>> planLocalUpdates<T extends TableRow>(
+  Future<({List<T> rows, bool projectionUnchanged})>
+  planLocalUpdates<T extends TableRow>(
     List<T> rows,
     List<Column>? columns,
     Transaction transaction,
   ) async {
-    if (rows.isEmpty) return rows;
+    if (rows.isEmpty) return (rows: rows, projectionUnchanged: true);
     final tableName = rows.first.table.tableName;
-    if (!_context.isCrdtTrackedTableName(tableName)) return rows;
+    if (!_context.isCrdtTrackedTableName(tableName)) {
+      return (rows: rows, projectionUnchanged: true);
+    }
     final columnNames = columns?.map((column) => column.columnName).toSet();
     if (!_foreignKeyProjector.needsProjection(tableName, columnNames)) {
-      return rows;
+      return (rows: rows, projectionUnchanged: true);
+    }
+
+    if (await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
+      rows,
+      transaction,
+      inserting: false,
+      columns: columns,
+    )) {
+      return (rows: rows, projectionUnchanged: true);
     }
 
     final authored = {
@@ -500,13 +557,16 @@ class CrdtMutationRecorder {
           if (row.id case final UuidValue rowId) (tableName, rowId),
       },
     );
-    return [
-      for (final row in rows)
-        _withPlannedDomainValues(
-          row,
-          planned.domain[(tableName, row.id as UuidValue)],
-        ),
-    ];
+    return (
+      projectionUnchanged: false,
+      rows: [
+        for (final row in rows)
+          _withPlannedDomainValues(
+            row,
+            planned.domain[(tableName, row.id as UuidValue)],
+          ),
+      ],
+    );
   }
 
   Map<String, Object?> _authoredValuesFromRow<T extends TableRow>(
@@ -538,7 +598,10 @@ class CrdtMutationRecorder {
         data[column.fieldName] = planned[column.columnName];
       }
     }
-    final className = data.remove('__className__') as String?;
+    final className =
+        _session.db.serializationManager.getClassNameForObject(row) ??
+        data['__className__'] as String?;
+    data.remove('__className__');
     if (className == null) {
       throw StateError(
         'Cannot apply planned domain values: ${row.runtimeType} has no class name.',
@@ -613,7 +676,14 @@ class CrdtMutationRecorder {
         transaction,
         attempts,
       );
-      await _maybeProject(tableName, rowIds, null, transaction);
+      if (_foreignKeyProjector.needsProjection(tableName, null) &&
+          !await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
+            insertedRows,
+            transaction,
+            inserting: true,
+          )) {
+        await _maybeProject(tableName, rowIds, null, transaction);
+      }
     });
   }
 
@@ -667,8 +737,9 @@ class CrdtMutationRecorder {
   Future<void> afterUpdate<T extends TableRow>(
     List<T> updatedRows,
     List<Column>? columns,
-    Transaction transaction,
-  ) async {
+    Transaction transaction, {
+    bool projectionUnchanged = false,
+  }) async {
     await _foreignKeyProjector.assertVisibleTargets(updatedRows, columns, transaction);
 
     await _forTrackedRows(updatedRows, transaction, (
@@ -696,6 +767,15 @@ class CrdtMutationRecorder {
         transaction,
         skippedFields: implicitForeignKeyRepairFields,
       );
+      if (projectionUnchanged &&
+          await _foreignKeyProjector.canLeaveLocalProjectionUnchanged(
+            updatedRows,
+            transaction,
+            inserting: false,
+            columns: columns,
+          )) {
+        return;
+      }
       final updatedColumnNames = columns?.map((column) => column.columnName).toSet();
       await _maybeProject(
         tableName,

test/serverpod_offline_sync_test_server/test/integration/sync/sync_local_projection_test.dart

@@ -0,0 +1,542 @@
+import 'package:serverpod_offline_sync_server/serverpod_offline_sync_server.dart';
+import 'package:serverpod_offline_sync_test_client/serverpod_offline_sync_test_client.dart';
+import 'package:test/test.dart';
+
+import '../test_tools/client_session.dart';
+import '../test_tools/crdt_probes.dart';
+import '../test_tools/sync_topology.dart';
+
+void main() {
+  initTestClientSession(createSessionPerTest: false);
+
+  group(
+    'Given people sharing an organization and an address attached to an unassigned person,',
+    () {
+      late SyncNode author;
+      late SyncNode observer;
+      late Organization organization;
+      late List<Person> siblings;
+      late Person newcomer;
+      late Address address;
+      late Map<UuidValue, Hlc> siblingHlcs;
+
+      setUpAll(() async {
+        author = await syncNode(await createAdditionalTestSession(), testSyncTables);
+        observer = await syncNode(
+          await createAdditionalTestSession(),
+          testSyncTables,
+        );
+        organization = Organization(id: const Uuid().v7obj(), name: 'organization');
+        siblings = [
+          for (var i = 0; i < 24; i++)
+            Person(
+              id: const Uuid().v7obj(),
+              name: 'person-$i',
+              organizationId: organization.id,
+            ),
+        ];
+        newcomer = Person(id: const Uuid().v7obj(), name: 'newcomer');
+        address = Address(
+          id: const Uuid().v7obj(),
+          street: 'street',
+          inhabitantId: newcomer.id,
+        );
+        await author.crdt.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Organization.db.insertRow(author.crdt, organization, transaction: tx);
+          await Person.db.insert(author.crdt, siblings, transaction: tx);
+          await Person.db.insertRow(author.crdt, newcomer, transaction: tx);
+          await Address.db.insertRow(author.crdt, address, transaction: tx);
+        });
+        siblingHlcs = {
+          for (final person in siblings)
+            person.id!: await rowHlc(person.id!, databaseSession: author.crdt),
+        };
+      });
+
+      group(
+        'when more people are inserted and the unassigned person joins the organization offline,',
+        () {
+          late List<Person> added;
+          late List<Person> authorPeople;
+          late List<Person> observerPeople;
+          late Address? authorAddress;
+          late Address? observerAddress;
+          late int authorAttemptedCount;
+          late int observerAttemptedCount;
+          late Map<UuidValue, Hlc> authorSiblingHlcs;
+          late Map<UuidValue, Hlc> observerSiblingHlcs;
+
+          setUpAll(() async {
+            added = [
+              for (var i = 24; i < 48; i++)
+                Person(
+                  id: const Uuid().v7obj(),
+                  name: 'person-$i',
+                  organizationId: organization.id,
+                ),
+            ];
+
+            await author.crdt.db.transactionForUser(testCrdtUserId, (tx) async {
+              await Person.db.insert(
+                author.crdt,
+                added,
+                transaction: tx,
+                noReturn: true,
+              );
+              await Person.db.updateRow(
+                author.crdt,
+                newcomer.copyWith(organizationId: organization.id),
+                columns: (t) => [t.organizationId],
+                transaction: tx,
+              );
+            });
+            await syncWithServer(author, observer);
+
+            authorPeople = await Person.db.find(author.crdt);
+            observerPeople = await Person.db.find(observer.crdt);
+            authorAddress = await Address.db.findById(author.crdt, address.id!);
+            observerAddress = await Address.db.findById(observer.crdt, address.id!);
+            authorAttemptedCount = await CrdtDataAttemptedValue.db.count(author.crdt);
+            observerAttemptedCount = await CrdtDataAttemptedValue.db.count(
+              observer.crdt,
+            );
+            authorSiblingHlcs = {
+              for (final person in siblings)
+                person.id!: await rowHlc(person.id!, databaseSession: author.crdt),
+            };
+            observerSiblingHlcs = {
+              for (final person in siblings)
+                person.id!: await rowHlc(person.id!, databaseSession: observer.crdt),
+            };
+          });
+
+          test('then both replicas see every person.', () {
+            final expectedIds = {
+              for (final person in [...siblings, ...added, newcomer]) person.id,
+            };
+            expect({for (final person in authorPeople) person.id}, expectedIds);
+            expect({for (final person in observerPeople) person.id}, expectedIds);
+          });
+
+          test('then every person on both replicas belongs to the organization.', () {
+            expect(
+              authorPeople.map((person) => person.organizationId),
+              everyElement(organization.id),
+            );
+            expect(
+              observerPeople.map((person) => person.organizationId),
+              everyElement(organization.id),
+            );
+          });
+
+          test(
+            'then the address on both replicas remains attached to the newcomer.',
+            () {
+              expect(authorAddress!.inhabitantId, newcomer.id);
+              expect(observerAddress!.inhabitantId, newcomer.id);
+            },
+          );
+
+          test('then neither replica holds a withheld authored value.', () {
+            expect(authorAttemptedCount, 0);
+            expect(observerAttemptedCount, 0);
+          });
+
+          test(
+            'then the original siblings keep their row clocks on both replicas.',
+            () {
+              expect(authorSiblingHlcs, siblingHlcs);
+              expect(observerSiblingHlcs, siblingHlcs);
+            },
+          );
+        },
+      );
+    },
+  );
+
+  group(
+    'Given two people in the same organization with an address attached to the first,',
+    () {
+      late SyncNode author;
+      late SyncNode observer;
+      late Organization organization;
+      late List<Person> people;
+      late Address address;
+      late Map<UuidValue, Hlc> originalClocks;
+
+      setUpAll(() async {
+        author = await syncNode(await createAdditionalTestSession(), testSyncTables);
+        observer = await syncNode(
+          await createAdditionalTestSession(),
+          testSyncTables,
+        );
+        organization = Organization(id: const Uuid().v7obj(), name: 'organization');
+        people = [
+          for (final name in ['first', 'second'])
+            Person(
+              id: const Uuid().v7obj(),
+              name: name,
+              organizationId: organization.id,
+            ),
+        ];
+        address = Address(
+          id: const Uuid().v7obj(),
+          street: 'street',
+          inhabitantId: people.first.id,
+        );
+        await author.crdt.db.transactionForUser(testCrdtUserId, (tx) async {
+          await Organization.db.insertRow(author.crdt, organization, transaction: tx);
+          await Person.db.insert(author.crdt, people, transaction: tx);
+          await Address.db.insertRow(author.crdt, address, transaction: tx);
+        });
+        originalClocks = {
+          for (final person in people)
+            person.id!: await rowHlc(person.id!, databaseSession: author.crdt),
+        };
+      });
+
+      group(
+        'when a primary-key upsert changes names only for the person matching its filter,',
+        () {
+          late List<Person> updated;
+          late List<Person> authorPeople;
+          late List<Person> observerPeople;
+          late Address? authorAddress;
+          late Address? observerAddress;
+          late Map<UuidValue, Hlc> authorClocks;
+          late Map<UuidValue, Hlc> observerClocks;
+          late List<CrdtDataField> authorFields;
+          late List<CrdtDataField> observerFields;
+          late int authorAttemptedCount;
+          late int observerAttemptedCount;
+
+          setUpAll(() async {
+            updated = await author.crdt.db.transactionForUser(
+              testCrdtUserId,
+              (tx) => Person.db.upsert(
+                author.crdt,
+                [for (final person in people) person.copyWith(name: 'updated')],
+                conflictColumns: (t) => [t.id],
+                updateColumns: (t) => [t.name],
+                updateWhere: (t) => t.name.equals('first'),
+                transaction: tx,
+              ),
+            );
+            await syncWithServer(author, observer);
+
+            authorPeople = await Person.db.find(author.crdt);
+            observerPeople = await Person.db.find(observer.crdt);
+            authorAddress = await Address.db.findById(author.crdt, address.id!);
+            observerAddress = await Address.db.findById(observer.crdt, address.id!);
+            authorClocks = {
+              for (final person in people)
+                person.id!: await rowHlc(person.id!, databaseSession: author.crdt),
+            };
+            observerClocks = {
+              for (final person in people)
+                person.id!: await rowHlc(person.id!, databaseSession: observer.crdt),
+            };
+            authorFields = await CrdtDataField.db.find(
+              author.crdt,
+              where: (t) =>
+                  t.row.uuidRowId.inSet({for (final person in people) person.id!}),
+              include: CrdtDataField.include(
+                row: CrdtDataRow.include(),
+                column: CrdtSchemaColumn.include(),
+                node: CrdtNode.include(),
+              ),
+            );
+            observerFields = await CrdtDataField.db.find(
+              observer.crdt,
+              where: (t) =>
+                  t.row.uuidRowId.inSet({for (final person in people) person.id!}),
+              include: CrdtDataField.include(
+                row: CrdtDataRow.include(),
+                column: CrdtSchemaColumn.include(),
+                node: CrdtNode.include(),
+              ),
+            );
+            authorAttemptedCount = await CrdtDataAttemptedValue.db.count(author.crdt);
+            observerAttemptedCount = await CrdtDataAttemptedValue.db.count(
+              observer.crdt,
+            );
+          });
+
+          test('then the upsert reports only the filtered person.', () {
+            expect(updated.map((person) => person.id).toSet(), {people.first.id});
+          });
+
+          test(
+            'then both replicas show the updated name only on the filtered person.',
+            () {
+              final expectedNames = {
+                people.first.id: 'updated',
+                people.last.id: 'second',
+              };
+              expect(
+                {for (final person in authorPeople) person.id: person.name},
+                expectedNames,
+              );
+              expect(
+                {for (final person in observerPeople) person.id: person.name},
+                expectedNames,
+              );
+            },
+          );
+
+          test(
+            'then every person on both replicas still belongs to the organization.',
+            () {
+              expect(
+                authorPeople.map((person) => person.organizationId),
+                everyElement(organization.id),
+              );
+              expect(
+                observerPeople.map((person) => person.organizationId),
+                everyElement(organization.id),
+              );
+            },
+          );
+
+          test(
+            'then the address on both replicas remains attached to the first person.',
+            () {
+              expect(authorAddress!.inhabitantId, people.first.id);
+              expect(observerAddress!.inhabitantId, people.first.id);
+            },
+          );
+
+          test('then both replicas keep the original row clocks.', () {
+            expect(authorClocks, originalClocks);
+            expect(observerClocks, originalClocks);
+          });
+
+          test(
+            'then both replicas author an organization for every person and a name only for the filtered person.',
+            () {
+              final expectedFields = {
+                for (final person in people) (person.id, 'organizationId'),
+                (people.first.id, 'name'),
+              };
+              expect(
+                {
+                  for (final field in authorFields)
+                    (field.row!.uuidRowId, field.column!.name),
+                },
+                expectedFields,
+              );
+              expect(
+                {
+                  for (final field in observerFields)
+                    (field.row!.uuidRowId, field.column!.name),
+                },
+                expectedFields,
+              );
+            },
+          );
+
+          test(
+            'then organization field clocks stay at the original row clocks on both replicas.',
+            () {
+              for (final fields in [authorFields, observerFields]) {
+                for (final field in fields.where(
+                  (field) => field.column!.name == 'organizationId',
+                )) {
+                  expect(field.hlc, originalClocks[field.row!.uuidRowId]);
+                }
+              }
+            },
+          );
+
+          test('then neither replica holds a withheld authored value.', () {
+            expect(authorAttemptedCount, 0);
+            expect(observerAttemptedCount, 0);
+          });
+        },
+      );
+    },
+  );
+
+  group('Given a merged town waiting for a missing mayor,', () {
+    late SyncNode node;
+    late Person mayor;
+    late Town town;
+    late Hlc mayorFieldHlc;
+
+    setUpAll(() async {
+      node = await syncNode(await createAdditionalTestSession(), testSyncTables);
+      mayor = Person(id: const Uuid().v7obj(), name: 'mayor');
+      town = Town(id: const Uuid().v7obj(), name: 'town', mayorId: mayor.id);
+      final hlc = Hlc(DateTime.now().toUtc(), 0, const Uuid().v7obj());
+      await node.crdt.db.mergeChanges([
+        CrdtMergeInsert(
+          uuidScopeId: testCrdtUserId,
+          tableName: Town.t.tableName,
+          uuidRowId: town.id!,
+          uuidNodeId: hlc.nodeId,
+          hlcDatetime: hlc.datetime,
+          hlcCounter: hlc.counter,
+          data: town,
+        ),
+      ], scopeId: testCrdtUserId);
+      expect((await Town.db.findById(node.crdt, town.id!))!.mayorId, isNull);
+      expect(
+        (await attemptedValue(
+          rowId: town.id!,
+          columnName: 'mayorId',
+          databaseSession: node.crdt,
+        ))!.value,
+        mayor.id,
+      );
+      mayorFieldHlc = await _fieldHlc(
+        town.id!,
+        'mayorId',
+        databaseSession: node.crdt,
+      );
+    });
+
+    group('when that person is inserted locally,', () {
+      late Town? recoveredTown;
+      late CrdtDataAttemptedValue? withheldMayor;
+      late Hlc recoveredMayorFieldHlc;
+
+      setUpAll(() async {
+        await node.crdt.db.transactionForUser(
+          testCrdtUserId,
+          (tx) => Person.db.insertRow(node.crdt, mayor, transaction: tx),
+        );
+
+        recoveredTown = await Town.db.findById(node.crdt, town.id!);
+        withheldMayor = await attemptedValue(
+          rowId: town.id!,
+          columnName: 'mayorId',
+          databaseSession: node.crdt,
+        );
+        recoveredMayorFieldHlc = await _fieldHlc(
+          town.id!,
+          'mayorId',
+          databaseSession: node.crdt,
+        );
+      });
+
+      test('then the town recovers its authored mayor.', () {
+        expect(recoveredTown!.mayorId, mayor.id);
+      });
+
+      test('then the town no longer holds a withheld mayor.', () {
+        expect(withheldMayor, isNull);
+      });
+
+      test('then the mayor field clock is unchanged.', () {
+        expect(recoveredMayorFieldHlc, mayorFieldHlc);
+      });
+    });
+  });
+
+  group(
+    'Given two merged unique children waiting for unavailable parents and a missing fallback,',
+    () {
+      const fallbackId = UuidValue.raw('550e8400-e29b-41d4-a716-446655440000');
+      late SyncNode node;
+      late List<UniqueSetDefaultChild> children;
+
+      setUpAll(() async {
+        node = await syncNode(await createAdditionalTestSession(), testSyncTables);
+        children = [
+          for (var i = 0; i < 2; i++)
+            UniqueSetDefaultChild(
+              id: const Uuid().v7obj(),
+              name: 'orphan-$i',
+              parentId: const Uuid().v7obj(),
+            ),
+        ];
+        final hlc = Hlc(DateTime.now().toUtc(), 0, const Uuid().v7obj());
+        await node.crdt.db.mergeChanges([
+          for (final child in children)
+            CrdtMergeInsert(
+              uuidScopeId: testCrdtUserId,
+              tableName: child.table.tableName,
+              uuidRowId: child.id!,
+              uuidNodeId: hlc.nodeId,
+              hlcDatetime: hlc.datetime,
+              hlcCounter: hlc.counter,
+              data: child,
+            ),
+        ], scopeId: testCrdtUserId);
+        expect(
+          (await UniqueSetDefaultChild.db.find(node.crdt)).map((row) => row.parentId),
+          everyElement(isNull),
+        );
+      });
+
+      group('when the fallback and another claimant are inserted locally,', () {
+        late UniqueSetDefaultChild claimant;
+        late List<UniqueSetDefaultChild> rows;
+        late Map<UuidValue?, UuidValue?> authoredParents;
+
+        setUpAll(() async {
+          claimant = UniqueSetDefaultChild(
+            id: const Uuid().v7obj(),
+            name: 'claimant',
+            parentId: fallbackId,
+          );
+          await node.crdt.db.transactionForUser(testCrdtUserId, (tx) async {
+            await Town.db.insertRow(
+              node.crdt,
+              Town(id: fallbackId, name: 'fallback'),
+              transaction: tx,
+            );
+            await UniqueSetDefaultChild.db.insertRow(
+              node.crdt,
+              claimant,
+              transaction: tx,
+            );
+          });
+
+          rows = await UniqueSetDefaultChild.db.find(node.crdt);
+          final facts = await node.sync
+              .collectPendingChanges(
+                node.raw,
+                checkpointsByScopeUuid: {testCrdtUserId: const []},
+              )
+              .toList();
+          authoredParents = {
+            for (final fact in facts.whereType<CrdtMergeInsert>())
+              if (fact.data case final UniqueSetDefaultChild row) row.id: row.parentId,
+          };
+        });
+
+        test('then three unique children are visible.', () {
+          expect(rows, hasLength(3));
+        });
+
+        test('then exactly one child owns the fallback.', () {
+          expect(rows.where((row) => row.parentId == fallbackId), hasLength(1));
+        });
+
+        test('then the other children have no visible parent.', () {
+          expect(rows.where((row) => row.parentId == null), hasLength(2));
+        });
+
+        test('then every exported insert retains its authored parent.', () {
+          expect(authoredParents, {
+            for (final row in [...children, claimant]) row.id: row.parentId,
+          });
+        });
+      });
+    },
+  );
+}
+
+Future<Hlc> _fieldHlc(
+  UuidValue rowId,
+  String column, {
+  required CrdtDatabaseSession databaseSession,
+}) async {
+  final field = await CrdtDataField.db.findFirstRow(
+    databaseSession,
+    where: (t) => t.row.uuidRowId.equals(rowId) & t.column.name.equals(column),
+    include: CrdtDataField.include(node: CrdtNode.include()),
+  );
+  return field!.hlc;
+}