perf: Write a merge's new rows in chunks when the pass is left out
A merge wrote each inserted row one by one: a savepoint for the domain row, another for its tracker, two statements each, and then read every tracker back to record attempted values. A device's first push of a library of 8,000 cards (24,650 changes) took the server 34 to 37 s, nearly all of it in those round trips. A batch the projection pass leaves out (nothing in it touches a foreign key or a unique index, or its inserts are proven to change no decision) now writes its new rows many to a statement, up to 1,000 at a time, in one savepoint: each table's domain rows in one insert, parents' tables first, then their trackers. A row qualifies when the batch inserts it once and touches it no other way, no tracker exists for it, nothing earlier in the batch names it, and every row of the batch it names was written before it. Its values are planned in the batch's order, as the per-row path plans them. A chunk whose writes fail (a row there already, perhaps another space's) is rolled back, the presence it claimed for its rows is forgotten, and it and every row after it are left to the per-row path, which tells those cases apart. Attempted values read the trackers the merge holds instead of querying them. The proof for merged inserts also accepts a parent of the row's own table inserted earlier in the batch, so a library's nested studysets no longer send its first push through the pass; a chunk ends before a row whose parent of its own table it holds. A row naming itself, or a parent inserted after it, still takes the pass. Measured on PostgreSQL, the 24,650 changes merge in about 5 s instead of 34 to 37 s, and with the studysets nested instead of 41 s. A device on SQLite, whose adapter still inserts row by row, receives the library in 44 s instead of 60 s. The new tests cover a batch spanning chunks, a chunk failing on a row whose tracker was lost and falling back, and a tree of one table's rows inserted parents first (the shortcut, a chunk per level) and children first (the pass). The core, server, integration and simulation suites pass, but for two PostgreSQL tests that can't start their database while another embedded PostgreSQL runs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
012d94d0ca
commit
949a6645e0
GPG key ID:
07F6FCF405488B64
1 changed files with 581 additions and 34 deletions
|
|
@ -92,13 +92,28 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
session: _session,
|
||||
projector: _foreignKeyProjector,
|
||||
fields: context.fields,
|
||||
rows: context.rows,
|
||||
transaction: transaction,
|
||||
);
|
||||
final deferredDeletes = <CrdtMergeDelete>[];
|
||||
// Without the pass, new rows nothing else in the batch touches are
|
||||
// written many to a statement instead of one by one in savepoints.
|
||||
final writtenInChunks = mayAffectProjection
|
||||
? const <MergeRowKey>{}
|
||||
: await _applyNewRowInsertsInChunks(
|
||||
operations,
|
||||
nodesByUuid,
|
||||
context,
|
||||
writes,
|
||||
presence,
|
||||
transaction,
|
||||
);
|
||||
|
||||
await writes.apply(
|
||||
operations.where(
|
||||
(operation) => _context.isCrdtTrackedTableName(operation.tableName),
|
||||
(operation) =>
|
||||
_context.isCrdtTrackedTableName(operation.tableName) &&
|
||||
!writtenInChunks.contains((operation.tableName, operation.uuidRowId)),
|
||||
),
|
||||
(operation) async {
|
||||
switch (operation) {
|
||||
|
|
@ -472,6 +487,244 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
);
|
||||
}
|
||||
|
||||
/// Writes the batch's new rows that nothing else in it touches, many to a
|
||||
/// statement, and returns their keys for the per-row pass to skip.
|
||||
///
|
||||
/// Only for a batch the projection pass skips, where each merged insert
|
||||
/// writes its row as planned, one by one in two savepoints and two
|
||||
/// statements of its own: nearly all the time of a first sync of thousands
|
||||
/// of rows. A row qualifies when the batch inserts it once and touches it
|
||||
/// no other way, nothing earlier in the batch names it, and each row of
|
||||
/// the batch it names qualified before it. Its values are planned in the
|
||||
/// batch's order, as the per-row path plans them. A chunk whose writes fail
|
||||
/// (a row there already, perhaps another space's) is rolled back and left,
|
||||
/// with every row after it, to the per-row path, which tells those apart.
|
||||
Future<Set<MergeRowKey>> _applyNewRowInsertsInChunks(
|
||||
List<CrdtMergeChange> operations,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
MergeContext context,
|
||||
_MergeWriteBatch writes,
|
||||
CrdtForeignKeyPresenceCache? presence,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final touches = <MergeRowKey, int>{};
|
||||
for (final operation in operations) {
|
||||
if (!_context.isCrdtTrackedTableName(operation.tableName)) continue;
|
||||
final rowKey = (operation.tableName, operation.uuidRowId);
|
||||
touches[rowKey] = (touches[rowKey] ?? 0) + 1;
|
||||
}
|
||||
|
||||
final written = <MergeRowKey>{};
|
||||
// Rows an insert names, by foreign key or as its set-default fallback: one
|
||||
// named before it is written would find the row missing.
|
||||
final named = <MergeRowKey>{};
|
||||
final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
|
||||
var chunk = _NewRowChunk();
|
||||
Future<bool> writeChunk() async {
|
||||
if (!await _writeNewRowChunk(chunk, context, writes, transaction)) {
|
||||
// Marked visible for their children, but not written after all.
|
||||
if (presence != null) chunk.rowKeys.forEach(presence.forget);
|
||||
return false;
|
||||
}
|
||||
written.addAll(chunk.rowKeys);
|
||||
chunk = _NewRowChunk();
|
||||
return true;
|
||||
}
|
||||
|
||||
for (final operation in operations) {
|
||||
if (operation is! CrdtMergeInsert ||
|
||||
!_context.isCrdtTrackedTableName(operation.tableName)) {
|
||||
continue;
|
||||
}
|
||||
final rowKey = (operation.tableName, operation.uuidRowId);
|
||||
final data = _sanitizeMergeRowData(
|
||||
operation.tableName,
|
||||
operation.databaseColumns,
|
||||
);
|
||||
final parents = _rowsNamedByMergeInsert(operation.tableName, data);
|
||||
// A parent named by another column could be any row: leave it all.
|
||||
if (parents == null) break;
|
||||
final qualifies =
|
||||
operation.data is TableRow &&
|
||||
touches[rowKey] == 1 &&
|
||||
context.rows[rowKey] == null &&
|
||||
!named.contains(rowKey) &&
|
||||
parents.every(
|
||||
(parent) =>
|
||||
!touches.containsKey(parent) ||
|
||||
written.contains(parent) ||
|
||||
chunk.holds(parent),
|
||||
);
|
||||
named.addAll(parents);
|
||||
for (final edge
|
||||
in _foreignKeys.edgesByChildTable[operation.tableName] ??
|
||||
const <ForeignKeyEdge>[]) {
|
||||
if (edge.action != ForeignKeyAction.setDefault) continue;
|
||||
if (tryUuidValue(edge.defaultValue) case final fallback?) {
|
||||
named.add((edge.parentTableName, fallback));
|
||||
}
|
||||
}
|
||||
if (!qualifies) continue;
|
||||
// Parents go in before their children: a chunk writes each table after
|
||||
// the tables its rows name there, so a row naming one of its own table,
|
||||
// or one whose table has to follow its own, waits for the next chunk.
|
||||
if (!chunk.canTake(operation.tableName, parents) && !await writeChunk()) {
|
||||
return written;
|
||||
}
|
||||
|
||||
final planned = await _planMergeInsertValues(
|
||||
operation,
|
||||
data,
|
||||
null,
|
||||
presence,
|
||||
transaction,
|
||||
);
|
||||
final domain = _mergeInsertDomainRow(operation, planned.domain, mergingSpaceId);
|
||||
chunk.add(
|
||||
(
|
||||
tableName: operation.tableName,
|
||||
rowId: operation.uuidRowId,
|
||||
node: _requireRemoteNode(remoteNodes, operation.uuidNodeId),
|
||||
hlc: operation.hlc,
|
||||
domainRow: domain.row,
|
||||
explicitNulls: domain.explicitNulls,
|
||||
attempts: planned.attempts,
|
||||
),
|
||||
parents,
|
||||
);
|
||||
// Its children further on find it written, as one by one.
|
||||
presence?.markVisible(rowKey);
|
||||
if (chunk.length >= _NewRowChunk.maxRows && !await writeChunk()) {
|
||||
return written;
|
||||
}
|
||||
}
|
||||
await writeChunk();
|
||||
return written;
|
||||
}
|
||||
|
||||
/// Writes [chunk] in one savepoint, domain rows first and their trackers
|
||||
/// after them, as the per-row path does, and records what it would.
|
||||
/// Rolls the chunk back and returns false when a statement fails.
|
||||
Future<bool> _writeNewRowChunk(
|
||||
_NewRowChunk chunk,
|
||||
MergeContext context,
|
||||
_MergeWriteBatch writes,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
if (chunk.isEmpty) return true;
|
||||
final Map<MergeRowKey, CrdtDataRow> trackers;
|
||||
try {
|
||||
trackers = await DatabaseUtil.runInTransactionOrSavepoint(
|
||||
_db,
|
||||
transaction,
|
||||
(savepoint) async {
|
||||
for (final tableName in chunk.tablesInWriteOrder) {
|
||||
final rows = chunk.rowsByTable[tableName]!;
|
||||
await insertWithExplicitNulls<TableRow>(
|
||||
_db,
|
||||
[for (final row in rows) row.domainRow],
|
||||
explicitNulls: {
|
||||
for (final row in rows)
|
||||
if (row.explicitNulls.isNotEmpty) row.rowId: row.explicitNulls,
|
||||
},
|
||||
transaction: savepoint,
|
||||
noReturn: true,
|
||||
);
|
||||
}
|
||||
final inserted = <MergeRowKey, CrdtDataRow>{};
|
||||
for (final MapEntry(key: tableName, value: rows)
|
||||
in chunk.rowsByTable.entries) {
|
||||
final trackers = await CrdtDataRow.db.insert(
|
||||
_session,
|
||||
[
|
||||
for (final row in rows)
|
||||
_newMergeRow(tableName, row.rowId, row.node, row.hlc, savepoint),
|
||||
],
|
||||
transaction: savepoint,
|
||||
);
|
||||
for (final tracker in trackers) {
|
||||
inserted[(tableName, tracker.uuidRowId)] = tracker;
|
||||
}
|
||||
}
|
||||
return inserted;
|
||||
},
|
||||
);
|
||||
} on DatabaseQueryException {
|
||||
return false;
|
||||
}
|
||||
|
||||
final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
|
||||
for (final row in chunk.rows) {
|
||||
final rowKey = (row.tableName, row.rowId);
|
||||
context.rows[rowKey] = trackers[rowKey]!.copyWith(node: row.node);
|
||||
context.domainOwners[rowKey] = (exists: true, spaceId: mergingSpaceId);
|
||||
writes.recordInsertAttempts(
|
||||
tableName: row.tableName,
|
||||
rowId: row.rowId,
|
||||
node: row.node,
|
||||
attempts: row.attempts,
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/// The rows [data], an insert into [tableName], names by foreign key, or
|
||||
/// null when one names its row by another column or not by a UUID.
|
||||
Set<MergeRowKey>? _rowsNamedByMergeInsert(
|
||||
String tableName,
|
||||
Map<String, Object?> data,
|
||||
) {
|
||||
final named = <MergeRowKey>{};
|
||||
for (final edge
|
||||
in _foreignKeys.edgesByChildTable[tableName] ?? const <ForeignKeyEdge>[]) {
|
||||
final value = data[edge.childColumn];
|
||||
if (value == null) continue;
|
||||
if (edge.parentColumn != 'id') return null;
|
||||
final id = tryUuidValue(value);
|
||||
if (id == null) return null;
|
||||
named.add((edge.parentTableName, id));
|
||||
}
|
||||
return named;
|
||||
}
|
||||
|
||||
/// The domain row a merged insert writes with its planned values [data],
|
||||
/// owned by [spaceId], and the columns whose null it writes though they
|
||||
/// have a default.
|
||||
({TableRow row, Set<String> explicitNulls}) _mergeInsertDomainRow(
|
||||
CrdtMergeInsert insert,
|
||||
Map<String, Object?> data,
|
||||
int spaceId,
|
||||
) {
|
||||
final tableRow = _requireMergeInsertTableRow(insert);
|
||||
final row = _withPlannedDomainValues(tableRow, data).copyWithSpaceId(spaceId);
|
||||
return (
|
||||
row: row,
|
||||
explicitNulls: {
|
||||
for (final column in row.table.crdtSyncableColumns)
|
||||
if (column.hasDefault && data[column.columnName] == null) column.columnName,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/// The tracker of a row a merge inserts, not written yet.
|
||||
CrdtDataRow _newMergeRow(
|
||||
String tableName,
|
||||
UuidValue rowId,
|
||||
CrdtNode remoteNode,
|
||||
Hlc incomingHlc,
|
||||
Transaction transaction,
|
||||
) {
|
||||
final (tableId, _) = _context.schema[tableName]!;
|
||||
return CrdtDataRow(
|
||||
spaceId: _context.hlcManagerFor(transaction).normalizedSpaceId,
|
||||
tblId: tableId,
|
||||
uuidRowId: rowId,
|
||||
nodeId: remoteNode.id!,
|
||||
hlcDatetime: incomingHlc.datetime,
|
||||
hlcCounter: incomingHlc.counter,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _applyMergeInsert(
|
||||
CrdtMergeInsert insert,
|
||||
Map<UuidValue, CrdtNode> remoteNodes,
|
||||
|
|
@ -724,17 +977,9 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Transaction transaction,
|
||||
) async {
|
||||
if (currentRow == null) {
|
||||
final (tableId, _) = _context.schema[tableName]!;
|
||||
final insertedRow = await CrdtDataRow.db.insertRow(
|
||||
_session,
|
||||
CrdtDataRow(
|
||||
spaceId: _context.hlcManagerFor(transaction).normalizedSpaceId,
|
||||
tblId: tableId,
|
||||
uuidRowId: rowId,
|
||||
nodeId: remoteNode.id!,
|
||||
hlcDatetime: incomingHlc.datetime,
|
||||
hlcCounter: incomingHlc.counter,
|
||||
),
|
||||
_newMergeRow(tableName, rowId, remoteNode, incomingHlc, transaction),
|
||||
transaction: transaction,
|
||||
);
|
||||
|
||||
|
|
@ -878,10 +1123,12 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
Map<String, Object?> data,
|
||||
Transaction transaction,
|
||||
) async {
|
||||
final tableRow = _requireMergeInsertTableRow(insert);
|
||||
final patchedRow = _withPlannedDomainValues(tableRow, data);
|
||||
final mergingSpaceId = _context.hlcManagerFor(transaction).normalizedSpaceId;
|
||||
final spaceScopedRow = patchedRow.copyWithSpaceId(mergingSpaceId);
|
||||
final (row: spaceScopedRow, :explicitNulls) = _mergeInsertDomainRow(
|
||||
insert,
|
||||
data,
|
||||
mergingSpaceId,
|
||||
);
|
||||
try {
|
||||
// Domain insert first. A CRDT tracker row for this id, written before
|
||||
// the domain row, made SQLite reject a later null unique-FK insert.
|
||||
|
|
@ -889,11 +1136,7 @@ extension CrdtMergeRecorderExtension on CrdtMutationRecorder {
|
|||
_db,
|
||||
transaction,
|
||||
(savepoint) => _db.insertRow(
|
||||
withExplicitInsertNulls(spaceScopedRow, {
|
||||
for (final column in spaceScopedRow.table.crdtSyncableColumns)
|
||||
if (column.hasDefault && data[column.columnName] == null)
|
||||
column.columnName,
|
||||
}),
|
||||
withExplicitInsertNulls(spaceScopedRow, explicitNulls),
|
||||
transaction: savepoint,
|
||||
),
|
||||
);
|
||||
|
|
|
|||
Loading…
Reference in a new issue