feat: Implement unique conflict handling to preserve invariants (#64)
This commit is contained in:
parent
60cca9cc85
commit
e74caa1400
3 changed files with 1347 additions and 66 deletions
|
|
@ -2,9 +2,18 @@ part of 'recorder.dart';
|
|||
|
||||
typedef _MergeRowKey = (String, UuidValue);
|
||||
typedef _MergeFieldKey = (String, UuidValue, String);
|
||||
typedef _MergeContext = ({
|
||||
Map<_MergeRowKey, CrdtDataRow> rows,
|
||||
Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
Map<_MergeFieldKey, Hlc> incomingFieldHlcs,
|
||||
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
});
|
||||
|
||||
/// Adds merge-specific behavior to [CrdtMutationRecorder].
|
||||
extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
||||
CrdtUniqueConflictResolver get _uniqueConflictResolver =>
|
||||
CrdtUniqueConflictResolver(this);
|
||||
|
||||
/// Locks the current user row so merges can serialize with other work.
|
||||
Future<void> lockCurrentUser(Transaction transaction) async {
|
||||
final user = _getEffectiveUser(transaction);
|
||||
|
|
@ -33,7 +42,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
{for (final change in operations) change.uuidNodeId},
|
||||
transaction,
|
||||
);
|
||||
final metadata = await _loadMergeMetadata(mergeSet, transaction);
|
||||
final context = await _loadMergeContext(mergeSet, transaction);
|
||||
|
||||
for (final operation in operations) {
|
||||
if (!_isCrdtTrackedTableName(operation.tableName)) {
|
||||
|
|
@ -45,25 +54,21 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
await _applyMergeInsert(
|
||||
insert,
|
||||
remoteNodes,
|
||||
metadata.rows,
|
||||
metadata.fields,
|
||||
metadata.tombstones,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
case final CrdtMergeUpdate update:
|
||||
await _applyMergeUpdate(
|
||||
update,
|
||||
remoteNodes,
|
||||
metadata.rows,
|
||||
metadata.fields,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
case final CrdtMergeDelete delete:
|
||||
await _applyMergeDelete(
|
||||
delete,
|
||||
remoteNodes,
|
||||
metadata.rows,
|
||||
metadata.tombstones,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
|
|
@ -171,14 +176,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
};
|
||||
}
|
||||
|
||||
Future<
|
||||
({
|
||||
Map<_MergeRowKey, CrdtDataRow> rows,
|
||||
Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
})
|
||||
>
|
||||
_loadMergeMetadata(
|
||||
Future<_MergeContext> _loadMergeContext(
|
||||
CrdtMergeSet mergeSet,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
|
|
@ -188,6 +186,10 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
|
||||
final rows = <_MergeRowKey, CrdtDataRow>{};
|
||||
final fields = <_MergeFieldKey, CrdtDataField>{};
|
||||
final incomingFieldHlcs = {
|
||||
for (final update in mergeSet.whereType<CrdtMergeUpdate>())
|
||||
(update.tableName, update.uuidRowId, update.columnName): update.hlc,
|
||||
};
|
||||
final tombstones = <_MergeRowKey, CrdtDataDeleted>{};
|
||||
|
||||
for (final MapEntry(key: tableName, value: rowIds) in rowIdsByTable.entries) {
|
||||
|
|
@ -245,6 +247,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
return (
|
||||
rows: rows,
|
||||
fields: fields,
|
||||
incomingFieldHlcs: incomingFieldHlcs,
|
||||
tombstones: tombstones,
|
||||
);
|
||||
}
|
||||
|
|
@ -252,16 +255,14 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeInsert(
|
||||
CrdtMergeInsert insert,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
Map<_MergeRowKey, CrdtDataRow> rows,
|
||||
Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (insert.tableName, insert.uuidRowId);
|
||||
final incomingHlc = insert.hlc;
|
||||
final remoteNode = _requireRemoteNode(remoteNodes, insert.uuidNodeId);
|
||||
|
||||
final currentRow = rows[rowKey];
|
||||
final currentRow = context.rows[rowKey];
|
||||
if (currentRow != null && incomingHlc <= currentRow.hlc) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -269,11 +270,18 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
final data = _sanitizeMergeRowData(insert.tableName, insert.databaseColumns);
|
||||
|
||||
if (currentRow == null) {
|
||||
rows[rowKey] = await _applyMergeInsertForMissingRow(
|
||||
final resolvedInsert = await _uniqueConflictResolver._resolveForIncomingInsert(
|
||||
insert,
|
||||
data,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
|
||||
context.rows[rowKey] = await _applyMergeInsertForMissingRow(
|
||||
insert,
|
||||
remoteNode,
|
||||
incomingHlc,
|
||||
data,
|
||||
resolvedInsert.data,
|
||||
transaction,
|
||||
);
|
||||
} else {
|
||||
|
|
@ -283,8 +291,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
remoteNode,
|
||||
incomingHlc,
|
||||
data,
|
||||
fields,
|
||||
tombstones,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
|
|
@ -293,12 +300,11 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeUpdate(
|
||||
CrdtMergeUpdate update,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
Map<_MergeRowKey, CrdtDataRow> rows,
|
||||
Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (update.tableName, update.uuidRowId);
|
||||
final row = rows[rowKey];
|
||||
final row = context.rows[rowKey];
|
||||
if (row == null) return;
|
||||
|
||||
final shouldApply = await _shouldMergeFieldMetadataIfNewer(
|
||||
|
|
@ -308,15 +314,22 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
row: row,
|
||||
remoteNode: _requireRemoteNode(remoteNodes, update.uuidNodeId),
|
||||
incomingHlc: update.hlc,
|
||||
fields: fields,
|
||||
fields: context.fields,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
if (!shouldApply) return;
|
||||
|
||||
final resolvedUpdates = await _uniqueConflictResolver._resolveForIncomingUpdate(
|
||||
update,
|
||||
row,
|
||||
context,
|
||||
transaction,
|
||||
);
|
||||
await _updateDomainRows(
|
||||
update.tableName,
|
||||
{update.uuidRowId},
|
||||
{update.columnName: update.value},
|
||||
resolvedUpdates,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
|
|
@ -324,21 +337,20 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Future<void> _applyMergeDelete(
|
||||
CrdtMergeDelete delete,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
Map<_MergeRowKey, CrdtDataRow> rows,
|
||||
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final rowKey = (delete.tableName, delete.uuidRowId);
|
||||
final row = rows[rowKey];
|
||||
final row = context.rows[rowKey];
|
||||
if (row == null) return;
|
||||
|
||||
final currentTombstone = tombstones[rowKey];
|
||||
final currentTombstone = context.tombstones[rowKey];
|
||||
final currentClFlag = currentTombstone?.clFlag ?? 1;
|
||||
final currentHlc = currentTombstone?.hlc ?? row.hlc;
|
||||
if (delete.clFlag < currentClFlag) return;
|
||||
if (delete.clFlag == currentClFlag && delete.hlc <= currentHlc) return;
|
||||
|
||||
tombstones[rowKey] = await _upsertMergeTombstone(
|
||||
context.tombstones[rowKey] = await _upsertMergeTombstone(
|
||||
row,
|
||||
_requireRemoteNode(remoteNodes, delete.uuidNodeId),
|
||||
delete.hlc,
|
||||
|
|
@ -503,17 +515,18 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
transaction,
|
||||
);
|
||||
|
||||
final row = _requireMergeInsertTableRow(insert);
|
||||
final tableRow = _requireMergeInsertTableRow(insert);
|
||||
final patchedRow = _db.serializationManager.patchTableRow(tableRow, data);
|
||||
try {
|
||||
await _db.updateRow(
|
||||
row,
|
||||
columns: row.table.managedColumns
|
||||
patchedRow,
|
||||
columns: patchedRow.table.managedColumns
|
||||
.where((c) => data.containsKey(c.columnName))
|
||||
.toList(),
|
||||
transaction: transaction,
|
||||
);
|
||||
} on DatabaseUpdateRowException {
|
||||
await _db.insertRow(row, transaction: transaction);
|
||||
await _db.insertRow(patchedRow, transaction: transaction);
|
||||
}
|
||||
|
||||
return insertedCrdtRow;
|
||||
|
|
@ -525,8 +538,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
CrdtNode remoteNode,
|
||||
Hlc incomingHlc,
|
||||
Map<String, Object?> data,
|
||||
Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
Map<_MergeRowKey, CrdtDataDeleted> tombstones,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final (_, columnsByName) = _schema[insert.tableName]!;
|
||||
|
|
@ -542,7 +554,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
row: currentRow,
|
||||
remoteNode: remoteNode,
|
||||
incomingHlc: incomingHlc,
|
||||
fields: fields,
|
||||
fields: context.fields,
|
||||
transaction: transaction,
|
||||
schemaColumn: schemaColumn,
|
||||
);
|
||||
|
|
@ -551,20 +563,31 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
updatedValues[columnName] = data[columnName];
|
||||
}
|
||||
|
||||
if (updatedValues.isNotEmpty) {
|
||||
var updatesToApply = updatedValues;
|
||||
if (updatesToApply.isNotEmpty) {
|
||||
updatesToApply = await _uniqueConflictResolver._resolveForIncomingUpdates(
|
||||
tableName: insert.tableName,
|
||||
row: currentRow,
|
||||
updates: updatesToApply,
|
||||
context: context,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
if (updatesToApply.isNotEmpty) {
|
||||
await _updateDomainRows(
|
||||
insert.tableName,
|
||||
{insert.uuidRowId},
|
||||
updatedValues,
|
||||
updatesToApply,
|
||||
transaction,
|
||||
);
|
||||
}
|
||||
|
||||
final rowKey = (insert.tableName, insert.uuidRowId);
|
||||
final currentTombstone = tombstones[rowKey];
|
||||
final currentTombstone = context.tombstones[rowKey];
|
||||
if ((currentTombstone?.clFlag ?? 1) == 1 &&
|
||||
incomingHlc > (currentTombstone?.hlc ?? currentRow.hlc)) {
|
||||
tombstones[rowKey] = await _upsertMergeTombstone(
|
||||
context.tombstones[rowKey] = await _upsertMergeTombstone(
|
||||
currentRow,
|
||||
remoteNode,
|
||||
incomingHlc,
|
||||
|
|
@ -640,3 +663,16 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
};
|
||||
}
|
||||
}
|
||||
|
||||
extension on DatabaseSerializationManager {
|
||||
/// Patches a [TableRow] with the given data.
|
||||
T patchTableRow<T extends TableRow>(T row, Map<String, Object?> data) {
|
||||
final rowJson = Map<String, dynamic>.from(row.toJson() as Map);
|
||||
for (final column in row.table.managedColumns) {
|
||||
if (column.columnName != 'id' && data.containsKey(column.columnName)) {
|
||||
rowJson[column.fieldName] = data[column.columnName];
|
||||
}
|
||||
}
|
||||
return deserialize<T>(rowJson);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,379 @@
|
|||
part of '../recorder.dart';
|
||||
|
||||
typedef _UniqueConflict = ({
|
||||
CrdtDataRow row,
|
||||
_UniqueIndexConflictRelease uniqueIndex,
|
||||
});
|
||||
|
||||
/// Resolves unique constraint conflicts that appear while merging CRDT changes.
|
||||
class CrdtUniqueConflictResolver {
|
||||
/// Creates a resolver for the merge recorder.
|
||||
CrdtUniqueConflictResolver(this._recorder);
|
||||
|
||||
final CrdtMutationRecorder _recorder;
|
||||
|
||||
/// Resolves unique conflicts for an incoming insert.
|
||||
Future<({Map<String, Object?> data, bool changed})> _resolveForIncomingInsert(
|
||||
CrdtMergeInsert insert,
|
||||
Map<String, Object?> data,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final resolvedData = Map<String, Object?>.from(data);
|
||||
var changed = false;
|
||||
final conflicts = await _findVisibleUniqueConflicts(
|
||||
tableName: insert.tableName,
|
||||
rowId: insert.uuidRowId,
|
||||
values: data,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
for (final conflict in conflicts) {
|
||||
final incomingClaimHlc = _incomingInsertUniqueClaim(
|
||||
insert: insert,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
context: context,
|
||||
);
|
||||
final conflictClaimHlc = await _uniqueIndexHlc(
|
||||
tableName: insert.tableName,
|
||||
row: conflict.row,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
fields: context.fields,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
if (incomingClaimHlc > conflictClaimHlc) {
|
||||
resolvedData.addAll(
|
||||
_uniqueConflictFreeValuesForData(
|
||||
insert.tableName,
|
||||
insert.uuidRowId,
|
||||
resolvedData,
|
||||
conflict.uniqueIndex,
|
||||
),
|
||||
);
|
||||
changed = true;
|
||||
} else {
|
||||
await _releaseUniqueConflictForRow(
|
||||
tableName: insert.tableName,
|
||||
rowId: conflict.row.uuidRowId,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return (data: resolvedData, changed: changed);
|
||||
}
|
||||
|
||||
Hlc _incomingInsertUniqueClaim({
|
||||
required CrdtMergeInsert insert,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required _MergeContext context,
|
||||
}) {
|
||||
var uniqueHlc = insert.hlc;
|
||||
for (final column in uniqueIndex.columns) {
|
||||
uniqueHlc = uniqueHlc.maxBetween(
|
||||
context.incomingFieldHlcs[(
|
||||
insert.tableName,
|
||||
insert.uuidRowId,
|
||||
column.columnName,
|
||||
)],
|
||||
);
|
||||
}
|
||||
return uniqueHlc;
|
||||
}
|
||||
|
||||
/// Resolves unique conflicts for a single incoming field update.
|
||||
Future<Map<String, Object?>> _resolveForIncomingUpdate(
|
||||
CrdtMergeUpdate update,
|
||||
CrdtDataRow row,
|
||||
_MergeContext context,
|
||||
Transaction transaction,
|
||||
) {
|
||||
return _resolveForIncomingUpdates(
|
||||
tableName: update.tableName,
|
||||
row: row,
|
||||
updates: {update.columnName: update.value},
|
||||
context: context,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
/// Resolves unique conflicts for one or more incoming field updates.
|
||||
Future<Map<String, Object?>> _resolveForIncomingUpdates({
|
||||
required String tableName,
|
||||
required CrdtDataRow row,
|
||||
required Map<String, Object?> updates,
|
||||
required _MergeContext context,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
if (updates.keys.every(
|
||||
(columnName) => !_isUniqueIndexedColumn(tableName, columnName),
|
||||
)) {
|
||||
return updates;
|
||||
}
|
||||
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) {
|
||||
return updates;
|
||||
}
|
||||
|
||||
final uniqueColumnNamesToRead = {
|
||||
for (final uniqueIndex in _recorder._uniqueIndexesForTable(tableDefinition))
|
||||
for (final column in uniqueIndex.columns) column.columnName,
|
||||
};
|
||||
if (uniqueColumnNamesToRead.isEmpty) {
|
||||
return updates;
|
||||
}
|
||||
|
||||
final currentValues = await _recorder._readDomainColumnValues(
|
||||
tableName,
|
||||
{row.uuidRowId},
|
||||
uniqueColumnNamesToRead.toList(),
|
||||
transaction,
|
||||
);
|
||||
final values = {
|
||||
...?currentValues[row.uuidRowId],
|
||||
...updates,
|
||||
};
|
||||
|
||||
final conflicts = await _findVisibleUniqueConflicts(
|
||||
tableName: tableName,
|
||||
rowId: row.uuidRowId,
|
||||
values: values,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
final resolvedUpdates = Map<String, Object?>.from(updates);
|
||||
for (final conflict in conflicts) {
|
||||
final rowClaimHlc = await _uniqueIndexHlc(
|
||||
tableName: tableName,
|
||||
row: row,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
fields: context.fields,
|
||||
transaction: transaction,
|
||||
);
|
||||
final conflictClaimHlc = await _uniqueIndexHlc(
|
||||
tableName: tableName,
|
||||
row: conflict.row,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
fields: context.fields,
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
if (rowClaimHlc > conflictClaimHlc) {
|
||||
final releasedValues = _uniqueConflictFreeValuesForData(
|
||||
tableName,
|
||||
row.uuidRowId,
|
||||
values,
|
||||
conflict.uniqueIndex,
|
||||
);
|
||||
for (final columnName in conflict.uniqueIndex.columnNames) {
|
||||
if (releasedValues.containsKey(columnName)) {
|
||||
resolvedUpdates[columnName] = releasedValues[columnName];
|
||||
values[columnName] = releasedValues[columnName];
|
||||
}
|
||||
}
|
||||
} else {
|
||||
await _releaseUniqueConflictForRow(
|
||||
tableName: tableName,
|
||||
rowId: conflict.row.uuidRowId,
|
||||
uniqueIndex: conflict.uniqueIndex,
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return resolvedUpdates;
|
||||
}
|
||||
|
||||
Map<String, Object?> _uniqueConflictFreeValuesForData(
|
||||
String tableName,
|
||||
UuidValue rowId,
|
||||
Map<String, Object?> data,
|
||||
_UniqueIndexConflictRelease uniqueIndex,
|
||||
) {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return data;
|
||||
|
||||
final released = Map<String, Object?>.from(data);
|
||||
final values = {
|
||||
for (final column in uniqueIndex.columns)
|
||||
column.columnName: released[column.columnName],
|
||||
};
|
||||
if (values.values.any((value) => value == null)) return released;
|
||||
|
||||
for (final column in uniqueIndex.columns) {
|
||||
released[column.columnName] = _recorder._conflictFreeValue(
|
||||
column,
|
||||
values[column.columnName],
|
||||
tableName,
|
||||
rowId,
|
||||
'conflict',
|
||||
);
|
||||
}
|
||||
return released;
|
||||
}
|
||||
|
||||
Future<void> _releaseUniqueConflictForRow({
|
||||
required String tableName,
|
||||
required UuidValue rowId,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final columnNames = uniqueIndex.columnNames.toList();
|
||||
final valuesByRowId = await _recorder._readDomainColumnValues(
|
||||
tableName,
|
||||
{rowId},
|
||||
columnNames,
|
||||
transaction,
|
||||
);
|
||||
final values = valuesByRowId[rowId];
|
||||
if (values == null || values.values.any((value) => value == null)) return;
|
||||
|
||||
final releasedValues = _uniqueConflictFreeValuesForData(
|
||||
tableName,
|
||||
rowId,
|
||||
values,
|
||||
uniqueIndex,
|
||||
);
|
||||
await _recorder._updateDomainRows(tableName, {rowId}, releasedValues, transaction);
|
||||
}
|
||||
|
||||
Future<List<_UniqueConflict>> _findVisibleUniqueConflicts({
|
||||
required String tableName,
|
||||
required UuidValue rowId,
|
||||
required Map<String, Object?> values,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return const [];
|
||||
|
||||
final uniqueIndexes = _recorder._uniqueIndexesForTable(tableDefinition);
|
||||
if (uniqueIndexes.isEmpty) return const [];
|
||||
|
||||
final uniqueIndexesByConflictId = <UuidValue, Set<_UniqueIndexConflictRelease>>{};
|
||||
for (final uniqueIndex in uniqueIndexes) {
|
||||
final columnValues = {
|
||||
for (final column in uniqueIndex.columns)
|
||||
if (values.containsKey(column.columnName))
|
||||
column.columnName: values[column.columnName],
|
||||
};
|
||||
if (columnValues.length != uniqueIndex.columns.length) continue;
|
||||
if (columnValues.values.any((value) => value == null)) continue;
|
||||
|
||||
final uniqueConflictIds = await _findVisibleRowsByUniqueValues(
|
||||
tableName: tableName,
|
||||
rowId: rowId,
|
||||
values: columnValues,
|
||||
transaction: transaction,
|
||||
);
|
||||
for (final conflictId in uniqueConflictIds) {
|
||||
uniqueIndexesByConflictId.putIfAbsent(conflictId, () => {}).add(uniqueIndex);
|
||||
}
|
||||
}
|
||||
|
||||
if (uniqueIndexesByConflictId.isEmpty) return const [];
|
||||
final conflictRows = await _recorder._findCrdtRows(
|
||||
tableName,
|
||||
uniqueIndexesByConflictId.keys.toSet(),
|
||||
transaction,
|
||||
include: CrdtDataRow.include(
|
||||
node: CrdtNode.include(),
|
||||
deleted: CrdtDataDeleted.include(node: CrdtNode.include()),
|
||||
),
|
||||
);
|
||||
return [
|
||||
for (final conflictRow in conflictRows)
|
||||
for (final uniqueIndex in uniqueIndexesByConflictId[conflictRow.uuidRowId]!)
|
||||
(
|
||||
row: conflictRow,
|
||||
uniqueIndex: uniqueIndex,
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
/// Returns the HLC for this row's current claim on [uniqueIndex].
|
||||
///
|
||||
/// Composite indexes use the newest field HLC among their columns; rows with
|
||||
/// no field metadata fall back to the row insert HLC. Missing field metadata
|
||||
/// is loaded lazily because conflicting rows may not be part of the preloaded
|
||||
/// merge context.
|
||||
Future<Hlc> _uniqueIndexHlc({
|
||||
required String tableName,
|
||||
required CrdtDataRow row,
|
||||
required _UniqueIndexConflictRelease uniqueIndex,
|
||||
required Map<_MergeFieldKey, CrdtDataField> fields,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final (_, columnsByName) = _recorder._schema[tableName]!;
|
||||
final uniqueColumnNames = {
|
||||
for (final column in uniqueIndex.columns) column.columnName,
|
||||
};
|
||||
final missingColumnIds = <int>{};
|
||||
final maxFieldHlc = <Hlc>[];
|
||||
for (final columnName in uniqueColumnNames) {
|
||||
final field = fields[(tableName, row.uuidRowId, columnName)];
|
||||
if (field == null) {
|
||||
final columnId = columnsByName[columnName]?.id;
|
||||
if (columnId != null) missingColumnIds.add(columnId);
|
||||
continue;
|
||||
}
|
||||
|
||||
maxFieldHlc.add(field.hlc);
|
||||
}
|
||||
|
||||
if (missingColumnIds.isNotEmpty && row.id != null) {
|
||||
final loadedFields = await CrdtDataField.db.find(
|
||||
_recorder._session,
|
||||
where: (t) => t.rowId.equals(row.id) & t.columnId.inSet(missingColumnIds),
|
||||
include: CrdtDataField.include(
|
||||
column: CrdtSchemaColumn.include(),
|
||||
node: CrdtNode.include(),
|
||||
),
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
for (final field in loadedFields) {
|
||||
final columnName = field.column?.name;
|
||||
if (columnName == null) continue;
|
||||
fields[(tableName, row.uuidRowId, columnName)] = field;
|
||||
maxFieldHlc.add(field.hlc);
|
||||
}
|
||||
}
|
||||
|
||||
if (maxFieldHlc.isEmpty) return row.hlc;
|
||||
return maxFieldHlc.reduce((left, right) => left.maxBetween(right));
|
||||
}
|
||||
|
||||
Future<Set<UuidValue>> _findVisibleRowsByUniqueValues({
|
||||
required String tableName,
|
||||
required UuidValue rowId,
|
||||
required Map<String, Object?> values,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
return _recorder._findVisibleDomainRowIdsWhere(
|
||||
tableName: tableName,
|
||||
predicates: [
|
||||
for (final MapEntry(key: columnName, value: value) in values.entries)
|
||||
'd."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}',
|
||||
'd."id" <> ${_sqlLiteral(rowId)}',
|
||||
],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
bool _isUniqueIndexedColumn(String tableName, String columnName) {
|
||||
final tableDefinition = _recorder._tableDefinitionsByName[tableName];
|
||||
if (tableDefinition == null) return false;
|
||||
return _recorder
|
||||
._uniqueIndexesForTable(tableDefinition)
|
||||
.any((i) => i.columns.any((c) => c.columnName == columnName));
|
||||
}
|
||||
}
|
||||
|
||||
extension on _UniqueIndexConflictRelease {
|
||||
Set<String> get columnNames => {
|
||||
for (final column in columns) column.columnName,
|
||||
};
|
||||
}
|
||||
|
|
@ -12,6 +12,7 @@ import 'schema.dart';
|
|||
import 'session.dart';
|
||||
|
||||
part 'merge.dart';
|
||||
part 'merge_utils/unique_resolver.dart';
|
||||
|
||||
typedef _ReferencingForeignKey = ({
|
||||
String childTableName,
|
||||
|
|
@ -893,6 +894,7 @@ WHERE c."id" IN ($whereRowIds)
|
|||
values[column.columnName],
|
||||
tableDefinition.name,
|
||||
rowId,
|
||||
'deleted',
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -925,6 +927,18 @@ WHERE c."id" IN ($whereRowIds)
|
|||
required String columnName,
|
||||
required Object? value,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
return _findVisibleDomainRowIdsWhere(
|
||||
tableName: tableName,
|
||||
predicates: ['d."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}'],
|
||||
transaction: transaction,
|
||||
);
|
||||
}
|
||||
|
||||
Future<Set<UuidValue>> _findVisibleDomainRowIdsWhere({
|
||||
required String tableName,
|
||||
required List<String> predicates,
|
||||
required Transaction transaction,
|
||||
}) async {
|
||||
final (tableId, _) = _schema[tableName]!;
|
||||
final userId = _getHlcManager(transaction).normalizedUserId;
|
||||
|
|
@ -936,7 +950,7 @@ LEFT JOIN "crdt_data_rows" r
|
|||
ON r."userId" = $userId AND r."tblId" = $tableId AND r."uuidRowId" = d."id"
|
||||
LEFT JOIN "crdt_data_tombstone" tomb
|
||||
ON tomb."rowId" = r."id"
|
||||
WHERE d."${_escapeIdentifier(columnName)}" = ${_sqlLiteral(value)}
|
||||
WHERE (${predicates.join(') AND (')})
|
||||
AND (tomb."id" IS NULL OR tomb."clFlag" % 2 = 1)
|
||||
''',
|
||||
transaction: transaction,
|
||||
|
|
@ -1017,21 +1031,20 @@ WHERE "id" IN (${_sqlLiteralList(rowIds)})
|
|||
Object? value,
|
||||
String tableName,
|
||||
UuidValue conflictingId,
|
||||
String releaseSuffix,
|
||||
) {
|
||||
switch (column.kind) {
|
||||
case _UniqueConflictReleaseKind.setNull:
|
||||
return null;
|
||||
case _UniqueConflictReleaseKind.textSuffix:
|
||||
if (value is String) {
|
||||
return '${value}__deleted__${conflictingId.uuid}';
|
||||
return '${value}__${releaseSuffix}__${conflictingId.uuid}';
|
||||
}
|
||||
case _UniqueConflictReleaseKind.syntheticUuid:
|
||||
if (value != null) {
|
||||
return _syntheticDeletedUuid(
|
||||
tableName,
|
||||
column.columnName,
|
||||
UuidValueJsonExtension.fromJson(value),
|
||||
conflictingId,
|
||||
return const Uuid().v5obj(
|
||||
Namespace.oid.value,
|
||||
'$tableName.${column.columnName}:${value}__${releaseSuffix}__$conflictingId',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -1140,20 +1153,6 @@ bool _isForeignKeyColumn(TableDefinition table, String columnName) {
|
|||
);
|
||||
}
|
||||
|
||||
UuidValue _syntheticDeletedUuid(
|
||||
String tableName,
|
||||
String columnName,
|
||||
UuidValue value,
|
||||
UuidValue conflictingId,
|
||||
) {
|
||||
return UuidValue.withValidation(
|
||||
const Uuid().v5(
|
||||
Namespace.oid.value,
|
||||
'$tableName.$columnName:${value.uuid}:${conflictingId.uuid}',
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
String _sqlLiteral(Object? value) => ValueEncoder.instance.convert(value);
|
||||
|
||||
String _sqlLiteralList(Iterable<Object?> values) =>
|
||||
|
|
|
|||
Loading…
Reference in a new issue