lib/sql/src/merge.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const simd = @import("simd");
3 const catalog_mod = @import("catalog.zig");
4 const diff_mod = @import("diff.zig");
5 const history_mod = @import("history/root.zig");
6 const key = @import("key.zig");
7 const relation_mod = @import("relation.zig");
8 const row = @import("row.zig");
9 const version = @import("version.zig");
10 const wal = @import("wal.zig");
11
12 const Bytes = simd.ScalableTag(u8);
13
14 const Allocator = std.mem.Allocator;
15
16 pub const Error = version.Error || relation_mod.Error || key.Error || row.Error || error{
17 DuplicateConflictSlot,
18 InvalidConflictSnapshot,
19 SchemaMismatch,
20 };
21
22 pub const RelationSnapshot = diff_mod.RelationSnapshot;
23
24 pub const ConflictSnapshot = struct {
25 root: version.ConflictRoot,
26 artifacts: []const version.ConflictArtifact,
27
28 pub fn empty() ConflictSnapshot {
29 return .{
30 .root = version.ConflictRoot.empty(),
31 .artifacts = &.{},
32 };
33 }
34 };
35
36 pub const DatabaseSnapshot = struct {
37 relations: []const RelationSnapshot,
38 conflicts: ConflictSnapshot,
39 };
40
41 pub const DatabaseSource = struct {
42 value: *const version.DatabaseValue,
43 conflicts: ConflictSnapshot,
44 };
45
46 pub const EditKind = enum {
47 added,
48 removed,
49 modified,
50 };
51
52 pub const Edit = struct {
53 kind: EditKind,
54 rowid: i64,
55 from: ?[]u8 = null,
56 to: ?[]u8 = null,
57
58 fn deinit(self: *Edit, allocator: Allocator) void {
59 if (self.from) |bytes| allocator.free(bytes);
60 if (self.to) |bytes| allocator.free(bytes);
61 self.* = undefined;
62 }
63 };
64
65 pub const Conflict = struct {
66 relation: []u8,
67 rowid: i64,
68 base: ?[]u8 = null,
69 ours: ?[]u8 = null,
70 theirs: ?[]u8 = null,
71 artifact: version.ConflictArtifact,
72
73 fn deinit(self: *Conflict, allocator: Allocator) void {
74 allocator.free(self.relation);
75 if (self.base) |bytes| allocator.free(bytes);
76 if (self.ours) |bytes| allocator.free(bytes);
77 if (self.theirs) |bytes| allocator.free(bytes);
78 self.* = undefined;
79 }
80 };
81
82 pub const RelationMerge = struct {
83 allocator: Allocator,
84 edits: []Edit,
85 conflicts: []Conflict,
86
87 pub fn hasConflicts(self: *const RelationMerge) bool {
88 return self.conflicts.len != 0;
89 }
90
91 pub fn deinit(self: *RelationMerge) void {
92 for (self.edits) |*edit| edit.deinit(self.allocator);
93 for (self.conflicts) |*conflict| conflict.deinit(self.allocator);
94 if (self.edits.len != 0) self.allocator.free(self.edits);
95 if (self.conflicts.len != 0) self.allocator.free(self.conflicts);
96 self.* = undefined;
97 }
98 };
99
100 pub const DatabaseConflict = struct {
101 kind: version.ConflictKind = .row,
102 relation: []const u8,
103 rowid: ?i64 = null,
104 base: ?[]const u8 = null,
105 ours: ?[]const u8 = null,
106 theirs: ?[]const u8 = null,
107 base_root: ?version.Hash = null,
108 ours_root: ?version.Hash = null,
109 theirs_root: ?version.Hash = null,
110 artifact: version.ConflictArtifact,
111 };
112
113 pub const RelationMergeMode = enum {
114 rows,
115 ours,
116 theirs,
117 agreed,
118 conflict,
119 };
120
121 pub const DatabaseRelationMerge = struct {
122 name: []u8,
123 root: version.Hash,
124 mode: RelationMergeMode,
125 result: RelationMerge,
126
127 fn deinit(self: *DatabaseRelationMerge, allocator: Allocator) void {
128 allocator.free(self.name);
129 self.result.deinit();
130 self.* = undefined;
131 }
132 };
133
134 pub const DatabaseMerge = struct {
135 allocator: Allocator,
136 value: version.DatabaseValue,
137 conflict_root: version.ConflictRoot,
138 relations: []DatabaseRelationMerge,
139 discovered: []DatabaseConflict,
140 artifacts: []version.ConflictArtifact,
141
142 pub fn hasConflicts(self: *const DatabaseMerge) bool {
143 return self.conflict_root.count != 0;
144 }
145
146 pub fn persistConflicts(self: *const DatabaseMerge, history: *history_mod.History) history_mod.Error!void {
147 for (self.artifacts) |artifact| try history.putConflict(artifact);
148 const entries = try self.allocator.alloc(version.ConflictEntry, self.artifacts.len);
149 defer if (entries.len != 0) self.allocator.free(entries);
150 for (self.artifacts, entries) |artifact, *entry| entry.* = artifact.entry();
151 const persisted = try history.putConflictRoot(entries);
152 if (!version.same(persisted.hash, self.conflict_root.hash) or persisted.count != self.conflict_root.count) return error.InvalidHistory;
153 }
154
155 pub fn deinit(self: *DatabaseMerge) void {
156 self.value.deinit();
157 for (self.relations) |*entry| entry.deinit(self.allocator);
158 for (self.artifacts) |*artifact| deinitConflictArtifact(self.allocator, artifact);
159 if (self.relations.len != 0) self.allocator.free(self.relations);
160 if (self.discovered.len != 0) self.allocator.free(self.discovered);
161 if (self.artifacts.len != 0) self.allocator.free(self.artifacts);
162 self.* = undefined;
163 }
164 };
165
166 pub fn relation(allocator: Allocator, base: RelationSnapshot, ours: RelationSnapshot, theirs: RelationSnapshot) Error!RelationMerge {
167 if (!schemasAgree(base.root, ours.root, theirs.root)) return error.SchemaMismatch;
168
169 var edits: std.ArrayList(Edit) = .empty;
170 errdefer {
171 for (edits.items) |*edit| edit.deinit(allocator);
172 edits.deinit(allocator);
173 }
174
175 var conflicts: std.ArrayList(Conflict) = .empty;
176 errdefer {
177 for (conflicts.items) |*conflict| conflict.deinit(allocator);
178 conflicts.deinit(allocator);
179 }
180
181 var base_scan: diff_mod.Scan = undefined;
182 try base.rows.scan(&base_scan, allocator, null, null);
183 defer base_scan.deinit();
184 var ours_scan: diff_mod.Scan = undefined;
185 try ours.rows.scan(&ours_scan, allocator, null, null);
186 defer ours_scan.deinit();
187 var theirs_scan: diff_mod.Scan = undefined;
188 try theirs.rows.scan(&theirs_scan, allocator, null, null);
189 defer theirs_scan.deinit();
190
191 var base_entry = try base_scan.next();
192 var ours_entry = try ours_scan.next();
193 var theirs_entry = try theirs_scan.next();
194
195 while (lowestRowid(base_entry, ours_entry, theirs_entry)) |rowid| {
196 const base_bytes = entryBytes(base_entry, rowid);
197 const ours_bytes = entryBytes(ours_entry, rowid);
198 const theirs_bytes = entryBytes(theirs_entry, rowid);
199
200 const ours_changed = !sameBytes(base_bytes, ours_bytes);
201 const theirs_changed = !sameBytes(base_bytes, theirs_bytes);
202
203 if (ours_changed or theirs_changed) {
204 if (!ours_changed) {
205 try appendEdit(allocator, &edits, rowid, base_bytes, theirs_bytes);
206 } else if (!theirs_changed) {
207 try appendEdit(allocator, &edits, rowid, base_bytes, ours_bytes);
208 } else if (sameBytes(ours_bytes, theirs_bytes)) {
209 try appendEdit(allocator, &edits, rowid, base_bytes, ours_bytes);
210 } else {
211 try appendConflict(allocator, &conflicts, ours.root.name, rowid, base_bytes, ours_bytes, theirs_bytes);
212 }
213 }
214
215 if (entryMatches(base_entry, rowid)) base_entry = try base_scan.next();
216 if (entryMatches(ours_entry, rowid)) ours_entry = try ours_scan.next();
217 if (entryMatches(theirs_entry, rowid)) theirs_entry = try theirs_scan.next();
218 }
219
220 return .{
221 .allocator = allocator,
222 .edits = try edits.toOwnedSlice(allocator),
223 .conflicts = try conflicts.toOwnedSlice(allocator),
224 };
225 }
226
227 pub fn mergeDatabase(allocator: Allocator, base: DatabaseSnapshot, ours: DatabaseSource, theirs: DatabaseSnapshot) Error!DatabaseMerge {
228 try validateConflictSnapshot(allocator, base.conflicts, null);
229 try validateConflictSnapshot(allocator, ours.conflicts, ours.value.root.conflicts);
230 try validateConflictSnapshot(allocator, theirs.conflicts, null);
231
232 var relation_results: std.ArrayList(DatabaseRelationMerge) = .empty;
233 errdefer {
234 for (relation_results.items) |*entry| entry.deinit(allocator);
235 relation_results.deinit(allocator);
236 }
237 var fresh_artifacts: std.ArrayList(version.ConflictArtifact) = .empty;
238 defer fresh_artifacts.deinit(allocator);
239 errdefer {
240 for (fresh_artifacts.items) |*artifact| deinitConflictArtifact(allocator, artifact);
241 }
242 var relation_values: std.ArrayList(version.RelationValue) = .empty;
243 errdefer {
244 for (relation_values.items) |*value| value.deinit(allocator);
245 relation_values.deinit(allocator);
246 }
247
248 var names: std.ArrayList([]const u8) = .empty;
249 defer names.deinit(allocator);
250 try appendSnapshotNames(allocator, &names, base.relations);
251 try appendValueNames(allocator, &names, ours.value.relations);
252 try appendSnapshotNames(allocator, &names, theirs.relations);
253 std.mem.sort([]const u8, names.items, {}, relationNameLessThan);
254
255 for (names.items) |name| {
256 const base_relation = findSnapshot(base.relations, name);
257 const ours_relation = ours.value.findRelation(name);
258 const theirs_relation = findSnapshot(theirs.relations, name);
259
260 if (base_relation == null) {
261 if (ours_relation) |ours_value| {
262 if (theirs_relation) |theirs_snapshot| {
263 if (!version.same(ours_value.root.hash, theirs_snapshot.root.hash)) {
264 try appendRelationConflict(allocator, &fresh_artifacts, name, null, ours_value.root.hash, theirs_snapshot.root.hash);
265 }
266 }
267 try appendClonedRelationValue(allocator, &relation_values, ours_value);
268 } else if (theirs_relation) |theirs_snapshot| {
269 try appendSnapshotRelationValue(allocator, &relation_values, theirs_snapshot);
270 }
271 continue;
272 }
273
274 const base_snapshot = base_relation.?;
275 if (ours_relation == null and theirs_relation == null) continue;
276 if (ours_relation == null) {
277 const theirs_snapshot = theirs_relation orelse unreachable;
278 if (!version.same(base_snapshot.root.hash, theirs_snapshot.root.hash)) {
279 try appendRelationConflict(allocator, &fresh_artifacts, name, base_snapshot.root.hash, null, theirs_snapshot.root.hash);
280 }
281 continue;
282 }
283 if (theirs_relation == null) {
284 const ours_value = ours_relation.?;
285 if (!version.same(base_snapshot.root.hash, ours_value.root.hash)) {
286 try appendRelationConflict(allocator, &fresh_artifacts, name, base_snapshot.root.hash, ours_value.root.hash, null);
287 try appendClonedRelationValue(allocator, &relation_values, ours_value);
288 }
289 continue;
290 }
291
292 var existing = try mergeExistingRelation(
293 allocator,
294 name,
295 base_snapshot,
296 ours_relation.?,
297 theirs_relation.?,
298 );
299 var entry_owned = true;
300 errdefer if (entry_owned) existing.entry.deinit(allocator);
301 var value_owned = true;
302 errdefer if (value_owned) existing.value.deinit(allocator);
303
304 if (existing.relation_conflict) {
305 try appendRelationConflict(
306 allocator,
307 &fresh_artifacts,
308 name,
309 base_snapshot.root.hash,
310 ours_relation.?.root.hash,
311 theirs_relation.?.root.hash,
312 );
313 }
314 for (existing.entry.result.conflicts) |conflict| {
315 try appendDatabaseConflict(allocator, &fresh_artifacts, conflict);
316 }
317 try relation_values.append(allocator, existing.value);
318 value_owned = false;
319 try relation_results.append(allocator, existing.entry);
320 entry_owned = false;
321 }
322
323 const artifact_slice = try reconcileConflictArtifacts(
324 allocator,
325 base.conflicts,
326 ours.conflicts,
327 theirs.conflicts,
328 &fresh_artifacts,
329 );
330 errdefer {
331 for (artifact_slice) |*artifact| deinitConflictArtifact(allocator, artifact);
332 if (artifact_slice.len != 0) allocator.free(artifact_slice);
333 }
334
335 const discovered_slice = try allocator.alloc(DatabaseConflict, fresh_artifacts.items.len);
336 errdefer if (discovered_slice.len != 0) allocator.free(discovered_slice);
337 for (fresh_artifacts.items, discovered_slice) |artifact, *conflict| {
338 const retained = findConflictArtifact(artifact_slice, artifact.entry()) orelse return error.InvalidConflictSnapshot;
339 conflict.* = databaseConflict(retained.*);
340 }
341 for (fresh_artifacts.items) |*artifact| deinitConflictArtifact(allocator, artifact);
342 fresh_artifacts.clearRetainingCapacity();
343
344 const conflict_entries = try allocator.alloc(version.ConflictEntry, artifact_slice.len);
345 defer if (conflict_entries.len != 0) allocator.free(conflict_entries);
346 for (artifact_slice, conflict_entries) |artifact, *entry| entry.* = artifact.entry();
347 const conflict_root = if (artifact_slice.len == 0) version.ConflictRoot.empty() else version.ConflictRoot.init(conflict_entries);
348
349 const relation_value_slice = try relation_values.toOwnedSlice(allocator);
350 var relation_value_slice_owned = true;
351 errdefer {
352 if (relation_value_slice_owned) {
353 for (relation_value_slice) |*value| value.deinit(allocator);
354 if (relation_value_slice.len != 0) allocator.free(relation_value_slice);
355 }
356 }
357 var value = try version.databaseValueFromOwnedRelations(allocator, relation_value_slice, conflict_root.hash);
358 relation_value_slice_owned = false;
359 errdefer value.deinit();
360
361 const relation_slice = try relation_results.toOwnedSlice(allocator);
362 errdefer {
363 for (relation_slice) |*entry| entry.deinit(allocator);
364 if (relation_slice.len != 0) allocator.free(relation_slice);
365 }
366
367 return .{
368 .allocator = allocator,
369 .value = value,
370 .conflict_root = conflict_root,
371 .relations = relation_slice,
372 .discovered = discovered_slice,
373 .artifacts = artifact_slice,
374 };
375 }
376
377 fn validateConflictSnapshot(allocator: Allocator, snapshot: ConflictSnapshot, expected: ?version.Hash) Error!void {
378 if (snapshot.root.count != snapshot.artifacts.len) return error.InvalidConflictSnapshot;
379 if (expected) |hash| {
380 if (!version.same(hash, snapshot.root.hash)) return error.InvalidConflictSnapshot;
381 }
382
383 const entries = try allocator.alloc(version.ConflictEntry, snapshot.artifacts.len);
384 defer if (entries.len != 0) allocator.free(entries);
385 for (snapshot.artifacts, entries) |artifact, *entry| {
386 try validateConflictArtifact(artifact);
387 entry.* = artifact.entry();
388 }
389 std.mem.sort(version.ConflictEntry, entries, {}, version.ConflictEntry.lessThan);
390 const derived = if (entries.len == 0) version.ConflictRoot.empty() else version.ConflictRoot.init(entries);
391 if (!version.same(derived.hash, snapshot.root.hash) or derived.count != snapshot.root.count) return error.InvalidConflictSnapshot;
392 }
393
394 fn validateConflictArtifact(artifact: version.ConflictArtifact) Error!void {
395 const canonical = switch (artifact.kind) {
396 .row => version.ConflictArtifact.init(
397 artifact.relation,
398 artifact.rowid,
399 try optionalRowValue(artifact.base),
400 try optionalRowValue(artifact.ours),
401 try optionalRowValue(artifact.theirs),
402 ),
403 .relation => blk: {
404 if (artifact.rowid != 0) return error.InvalidConflictSnapshot;
405 break :blk version.ConflictArtifact.initRelation(
406 artifact.relation,
407 try optionalRelationValue(artifact.base),
408 try optionalRelationValue(artifact.ours),
409 try optionalRelationValue(artifact.theirs),
410 );
411 },
412 };
413 if (!version.same(canonical.hash, artifact.hash)) return error.InvalidConflictSnapshot;
414 }
415
416 fn reconcileConflictArtifacts(
417 allocator: Allocator,
418 base: ConflictSnapshot,
419 ours: ConflictSnapshot,
420 theirs: ConflictSnapshot,
421 fresh: *const std.ArrayList(version.ConflictArtifact),
422 ) Error![]version.ConflictArtifact {
423 for (fresh.items, 0..) |artifact, index| {
424 try validateConflictArtifact(artifact);
425 for (fresh.items[0..index]) |previous| {
426 if (artifact.entry().sameSlot(previous.entry())) return error.DuplicateConflictSlot;
427 }
428 }
429
430 var capacity = std.math.add(usize, base.artifacts.len, ours.artifacts.len) catch return error.InvalidConflictSnapshot;
431 capacity = std.math.add(usize, capacity, theirs.artifacts.len) catch return error.InvalidConflictSnapshot;
432 capacity = std.math.add(usize, capacity, fresh.items.len) catch return error.InvalidConflictSnapshot;
433
434 var result: std.ArrayList(version.ConflictArtifact) = .empty;
435 errdefer {
436 for (result.items) |*artifact| deinitConflictArtifact(allocator, artifact);
437 result.deinit(allocator);
438 }
439 try result.ensureTotalCapacity(allocator, capacity);
440
441 for (base.artifacts) |artifact| try appendCarriedArtifact(allocator, &result, artifact, base, ours, theirs, fresh.items);
442 for (ours.artifacts) |artifact| try appendCarriedArtifact(allocator, &result, artifact, base, ours, theirs, fresh.items);
443 for (theirs.artifacts) |artifact| try appendCarriedArtifact(allocator, &result, artifact, base, ours, theirs, fresh.items);
444 for (fresh.items) |artifact| try appendUniqueConflictArtifact(allocator, &result, artifact);
445
446 std.mem.sort(version.ConflictArtifact, result.items, {}, conflictArtifactLessThan);
447 return try result.toOwnedSlice(allocator);
448 }
449
450 fn appendCarriedArtifact(
451 allocator: Allocator,
452 result: *std.ArrayList(version.ConflictArtifact),
453 artifact: version.ConflictArtifact,
454 base: ConflictSnapshot,
455 ours: ConflictSnapshot,
456 theirs: ConflictSnapshot,
457 fresh: []const version.ConflictArtifact,
458 ) Error!void {
459 const entry = artifact.entry();
460 const in_base = hasConflictArtifact(base.artifacts, entry);
461 const in_ours = hasConflictArtifact(ours.artifacts, entry);
462 const in_theirs = hasConflictArtifact(theirs.artifacts, entry);
463 const keep = if (in_base) in_ours and in_theirs else in_ours or in_theirs;
464 if (!keep or hasConflictSlot(fresh, entry)) return;
465 try appendUniqueConflictArtifact(allocator, result, artifact);
466 }
467
468 fn appendUniqueConflictArtifact(allocator: Allocator, result: *std.ArrayList(version.ConflictArtifact), artifact: version.ConflictArtifact) Error!void {
469 if (findConflictArtifact(result.items, artifact.entry()) != null) return;
470 result.appendAssumeCapacity(try cloneConflictArtifact(allocator, artifact));
471 }
472
473 fn hasConflictArtifact(artifacts: []const version.ConflictArtifact, entry: version.ConflictEntry) bool {
474 return findConflictArtifact(artifacts, entry) != null;
475 }
476
477 fn findConflictArtifact(artifacts: []const version.ConflictArtifact, entry: version.ConflictEntry) ?*const version.ConflictArtifact {
478 for (artifacts) |*artifact| {
479 if (artifact.entry().eql(entry)) return artifact;
480 }
481 return null;
482 }
483
484 fn hasConflictSlot(artifacts: []const version.ConflictArtifact, entry: version.ConflictEntry) bool {
485 for (artifacts) |artifact| {
486 if (artifact.entry().sameSlot(entry)) return true;
487 }
488 return false;
489 }
490
491 fn cloneConflictArtifact(allocator: Allocator, artifact: version.ConflictArtifact) Error!version.ConflictArtifact {
492 const relation_name = try allocator.dupe(u8, artifact.relation);
493 errdefer allocator.free(relation_name);
494 return switch (artifact.kind) {
495 .row => blk: {
496 const base = try copyOptional(allocator, try optionalRowValue(artifact.base));
497 errdefer if (base) |bytes| allocator.free(bytes);
498 const ours = try copyOptional(allocator, try optionalRowValue(artifact.ours));
499 errdefer if (ours) |bytes| allocator.free(bytes);
500 const theirs = try copyOptional(allocator, try optionalRowValue(artifact.theirs));
501 errdefer if (theirs) |bytes| allocator.free(bytes);
502 break :blk version.ConflictArtifact.init(relation_name, artifact.rowid, base, ours, theirs);
503 },
504 .relation => version.ConflictArtifact.initRelation(
505 relation_name,
506 try optionalRelationValue(artifact.base),
507 try optionalRelationValue(artifact.ours),
508 try optionalRelationValue(artifact.theirs),
509 ),
510 };
511 }
512
513 fn deinitConflictArtifact(allocator: Allocator, artifact: *version.ConflictArtifact) void {
514 allocator.free(artifact.relation);
515 if (artifact.kind == .row) {
516 if (artifact.base) |value| allocator.free(value.row);
517 if (artifact.ours) |value| allocator.free(value.row);
518 if (artifact.theirs) |value| allocator.free(value.row);
519 }
520 artifact.* = undefined;
521 }
522
523 fn optionalRowValue(value: ?version.ConflictValue) Error!?[]const u8 {
524 const present = value orelse return null;
525 return switch (present) {
526 .row => |bytes| bytes,
527 .relation => error.InvalidConflictSnapshot,
528 };
529 }
530
531 fn optionalRelationValue(value: ?version.ConflictValue) Error!?version.Hash {
532 const present = value orelse return null;
533 return switch (present) {
534 .row => error.InvalidConflictSnapshot,
535 .relation => |hash| hash,
536 };
537 }
538
539 fn conflictArtifactLessThan(_: void, left: version.ConflictArtifact, right: version.ConflictArtifact) bool {
540 return version.ConflictEntry.lessThan({}, left.entry(), right.entry());
541 }
542
543 fn databaseConflict(artifact: version.ConflictArtifact) DatabaseConflict {
544 return switch (artifact.kind) {
545 .row => .{
546 .kind = .row,
547 .relation = artifact.relation,
548 .rowid = artifact.rowid,
549 .base = optionalRowValue(artifact.base) catch unreachable,
550 .ours = optionalRowValue(artifact.ours) catch unreachable,
551 .theirs = optionalRowValue(artifact.theirs) catch unreachable,
552 .artifact = artifact,
553 },
554 .relation => .{
555 .kind = .relation,
556 .relation = artifact.relation,
557 .base_root = optionalRelationValue(artifact.base) catch unreachable,
558 .ours_root = optionalRelationValue(artifact.ours) catch unreachable,
559 .theirs_root = optionalRelationValue(artifact.theirs) catch unreachable,
560 .artifact = artifact,
561 },
562 };
563 }
564
565 fn appendSnapshotNames(allocator: Allocator, names: *std.ArrayList([]const u8), relations: []const RelationSnapshot) Allocator.Error!void {
566 for (relations) |entry| try appendRelationName(allocator, names, entry.root.name);
567 }
568
569 fn appendValueNames(allocator: Allocator, names: *std.ArrayList([]const u8), relations: []const version.RelationValue) Allocator.Error!void {
570 for (relations) |entry| try appendRelationName(allocator, names, entry.root.name);
571 }
572
573 fn appendRelationName(allocator: Allocator, names: *std.ArrayList([]const u8), name: []const u8) Allocator.Error!void {
574 for (names.items) |existing| {
575 if (std.mem.eql(u8, existing, name)) return;
576 }
577 try names.append(allocator, name);
578 }
579
580 fn relationNameLessThan(_: void, left: []const u8, right: []const u8) bool {
581 return simd.order(Bytes, left, right) == .lt;
582 }
583
584 fn findSnapshot(relations: []const RelationSnapshot, name: []const u8) ?*const RelationSnapshot {
585 for (relations) |*entry| {
586 if (std.mem.eql(u8, entry.root.name, name)) return entry;
587 }
588 return null;
589 }
590
591 const ExistingRelationMerge = struct {
592 value: version.RelationValue,
593 entry: DatabaseRelationMerge,
594 relation_conflict: bool,
595 };
596
597 fn mergeExistingRelation(
598 allocator: Allocator,
599 name: []const u8,
600 base: *const RelationSnapshot,
601 ours: *const version.RelationValue,
602 theirs: *const RelationSnapshot,
603 ) Error!ExistingRelationMerge {
604 if (schemasAgree(base.root, &ours.root, theirs.root)) {
605 const ours_snapshot = RelationSnapshot{
606 .root = &ours.root,
607 .rows = .{ .materialized = ours.rows },
608 };
609 var merged = try relation(allocator, base.*, ours_snapshot, theirs.*);
610 errdefer merged.deinit();
611 var value = try relationValueApplyingMerge(allocator, ours, merged.edits);
612 errdefer value.deinit(allocator);
613 return .{
614 .value = value,
615 .entry = .{
616 .name = try allocator.dupe(u8, name),
617 .root = value.root.hash,
618 .mode = .rows,
619 .result = merged,
620 },
621 .relation_conflict = false,
622 };
623 }
624
625 const mode = wholeRelationMode(base.root, &ours.root, theirs.root);
626 var value = try wholeRelationValue(allocator, mode, ours, theirs);
627 errdefer value.deinit(allocator);
628 return .{
629 .value = value,
630 .entry = .{
631 .name = try allocator.dupe(u8, name),
632 .root = value.root.hash,
633 .mode = mode,
634 .result = emptyRelationMerge(allocator),
635 },
636 .relation_conflict = mode == .conflict,
637 };
638 }
639
640 fn schemasAgree(
641 base: *const version.RelationRoot,
642 ours: *const version.RelationRoot,
643 theirs: *const version.RelationRoot,
644 ) bool {
645 return version.same(base.schema, ours.schema) and
646 version.same(ours.schema, theirs.schema);
647 }
648
649 fn wholeRelationMode(
650 base: *const version.RelationRoot,
651 ours: *const version.RelationRoot,
652 theirs: *const version.RelationRoot,
653 ) RelationMergeMode {
654 if (version.same(ours.hash, theirs.hash)) return .agreed;
655 if (version.same(ours.hash, base.hash)) return .theirs;
656 if (version.same(theirs.hash, base.hash)) return .ours;
657 return .conflict;
658 }
659
660 fn wholeRelationValue(
661 allocator: Allocator,
662 mode: RelationMergeMode,
663 ours: *const version.RelationValue,
664 theirs: *const RelationSnapshot,
665 ) Error!version.RelationValue {
666 return switch (mode) {
667 .theirs => try relationValueFromSnapshot(allocator, theirs),
668 .ours, .agreed, .conflict => try ours.clone(allocator),
669 .rows => unreachable,
670 };
671 }
672
673 fn emptyRelationMerge(allocator: Allocator) RelationMerge {
674 return .{
675 .allocator = allocator,
676 .edits = &.{},
677 .conflicts = &.{},
678 };
679 }
680
681 fn appendClonedRelationValue(allocator: Allocator, relation_values: *std.ArrayList(version.RelationValue), source: *const version.RelationValue) Allocator.Error!void {
682 var clone = try source.clone(allocator);
683 var clone_owned = true;
684 errdefer if (clone_owned) clone.deinit(allocator);
685 try relation_values.append(allocator, clone);
686 clone_owned = false;
687 }
688
689 fn appendSnapshotRelationValue(allocator: Allocator, relation_values: *std.ArrayList(version.RelationValue), source: *const RelationSnapshot) Error!void {
690 var value = try relationValueFromSnapshot(allocator, source);
691 var value_owned = true;
692 errdefer if (value_owned) value.deinit(allocator);
693 try relation_values.append(allocator, value);
694 value_owned = false;
695 }
696
697 fn relationValueFromSnapshot(allocator: Allocator, snapshot: *const RelationSnapshot) Error!version.RelationValue {
698 var root = try snapshot.root.clone(allocator);
699 errdefer root.deinit();
700 var rows: std.ArrayList(version.RelationRow) = .empty;
701 errdefer {
702 for (rows.items) |row_value| allocator.free(row_value.bytes);
703 rows.deinit(allocator);
704 }
705
706 var scan: diff_mod.Scan = undefined;
707 try snapshot.rows.scan(&scan, allocator, null, null);
708 defer scan.deinit();
709 while (try scan.next()) |entry| {
710 const bytes = try allocator.dupe(u8, entry.bytes);
711 rows.append(allocator, .{
712 .rowid = entry.rowid,
713 .bytes = bytes,
714 }) catch |err| {
715 allocator.free(bytes);
716 return err;
717 };
718 }
719
720 return .{
721 .root = root,
722 .rows = try rows.toOwnedSlice(allocator),
723 };
724 }
725
726 fn appendEdit(allocator: Allocator, edits: *std.ArrayList(Edit), rowid: i64, from_bytes: ?[]const u8, to_bytes: ?[]const u8) Error!void {
727 if (sameBytes(from_bytes, to_bytes)) return;
728
729 const from = try copyOptional(allocator, from_bytes);
730 errdefer if (from) |bytes| allocator.free(bytes);
731 const to = try copyOptional(allocator, to_bytes);
732 errdefer if (to) |bytes| allocator.free(bytes);
733
734 try edits.append(allocator, .{
735 .kind = editKind(from_bytes, to_bytes),
736 .rowid = rowid,
737 .from = from,
738 .to = to,
739 });
740 }
741
742 fn appendConflict(allocator: Allocator, conflicts: *std.ArrayList(Conflict), relation_name: []const u8, rowid: i64, base_bytes: ?[]const u8, ours_bytes: ?[]const u8, theirs_bytes: ?[]const u8) Error!void {
743 const owned_relation = try allocator.dupe(u8, relation_name);
744 errdefer allocator.free(owned_relation);
745 const base = try copyOptional(allocator, base_bytes);
746 errdefer if (base) |bytes| allocator.free(bytes);
747 const ours = try copyOptional(allocator, ours_bytes);
748 errdefer if (ours) |bytes| allocator.free(bytes);
749 const theirs = try copyOptional(allocator, theirs_bytes);
750 errdefer if (theirs) |bytes| allocator.free(bytes);
751 const artifact = version.ConflictArtifact.init(owned_relation, rowid, base, ours, theirs);
752
753 try conflicts.append(allocator, .{
754 .relation = owned_relation,
755 .rowid = rowid,
756 .base = base,
757 .ours = ours,
758 .theirs = theirs,
759 .artifact = artifact,
760 });
761 }
762
763 fn appendDatabaseConflict(allocator: Allocator, conflicts: *std.ArrayList(version.ConflictArtifact), conflict: Conflict) Error!void {
764 const relation_name = try allocator.dupe(u8, conflict.artifact.relation);
765 errdefer allocator.free(relation_name);
766 const base = try copyOptional(allocator, conflict.base);
767 errdefer if (base) |bytes| allocator.free(bytes);
768 const ours = try copyOptional(allocator, conflict.ours);
769 errdefer if (ours) |bytes| allocator.free(bytes);
770 const theirs = try copyOptional(allocator, conflict.theirs);
771 errdefer if (theirs) |bytes| allocator.free(bytes);
772 const artifact = version.ConflictArtifact.init(relation_name, conflict.artifact.rowid, base, ours, theirs);
773 try conflicts.append(allocator, artifact);
774 }
775
776 fn appendRelationConflict(allocator: Allocator, conflicts: *std.ArrayList(version.ConflictArtifact), relation_name: []const u8, base: ?version.Hash, ours: ?version.Hash, theirs: ?version.Hash) Error!void {
777 const owned_relation = try allocator.dupe(u8, relation_name);
778 errdefer allocator.free(owned_relation);
779 const artifact = version.ConflictArtifact.initRelation(owned_relation, base, ours, theirs);
780 try conflicts.append(allocator, artifact);
781 }
782
783 fn relationValueApplyingMerge(allocator: Allocator, ours: *const version.RelationValue, edits: []const Edit) Error!version.RelationValue {
784 var relation_edits: std.ArrayList(relation_mod.Edit) = .empty;
785 defer relation_edits.deinit(allocator);
786 try relation_edits.ensureTotalCapacity(allocator, edits.len);
787 var ours_index: usize = 0;
788 var previous_rowid: ?i64 = null;
789 for (edits) |edit| {
790 if (previous_rowid) |rowid| std.debug.assert(rowid < edit.rowid);
791 while (ours_index < ours.rows.len and ours.rows[ours_index].rowid < edit.rowid) {
792 ours_index += 1;
793 }
794 const ours_bytes = if (ours_index < ours.rows.len and
795 ours.rows[ours_index].rowid == edit.rowid)
796 ours.rows[ours_index].bytes
797 else
798 null;
799 if (!sameBytes(ours_bytes, edit.to)) {
800 relation_edits.appendAssumeCapacity(relationEdit(edit));
801 }
802 previous_rowid = edit.rowid;
803 }
804 return try version.relationValueApplyingMaterializedEdits(allocator, ours, relation_edits.items);
805 }
806
807 fn relationEdit(edit: Edit) relation_mod.Edit {
808 if (edit.to) |bytes| {
809 return .{ .put = .{
810 .rowid = edit.rowid,
811 .bytes = bytes,
812 } };
813 }
814 return .{ .delete = edit.rowid };
815 }
816
817 fn editKind(from_bytes: ?[]const u8, to_bytes: ?[]const u8) EditKind {
818 if (from_bytes == null) return .added;
819 if (to_bytes == null) return .removed;
820 return .modified;
821 }
822
823 fn copyOptional(allocator: Allocator, bytes: ?[]const u8) Allocator.Error!?[]u8 {
824 return if (bytes) |value| try allocator.dupe(u8, value) else null;
825 }
826
827 fn sameBytes(left: ?[]const u8, right: ?[]const u8) bool {
828 if (left == null and right == null) return true;
829 if (left == null or right == null) return false;
830 return std.mem.eql(u8, left.?, right.?);
831 }
832
833 fn entryBytes(entry: ?diff_mod.Row, rowid: i64) ?[]const u8 {
834 if (entry) |value| {
835 if (value.rowid == rowid) return value.bytes;
836 }
837 return null;
838 }
839
840 fn entryMatches(entry: ?diff_mod.Row, rowid: i64) bool {
841 return if (entry) |value| value.rowid == rowid else false;
842 }
843
844 fn lowestRowid(base: ?diff_mod.Row, ours: ?diff_mod.Row, theirs: ?diff_mod.Row) ?i64 {
845 var found: ?i64 = null;
846 if (base) |entry| found = minRowid(found, entry.rowid);
847 if (ours) |entry| found = minRowid(found, entry.rowid);
848 if (theirs) |entry| found = minRowid(found, entry.rowid);
849 return found;
850 }
851
852 fn minRowid(current: ?i64, rowid: i64) i64 {
853 return if (current) |value| @min(value, rowid) else rowid;
854 }
855
856 test "relation merge resolves nonoverlapping row edits" {
857 var tmp = std.testing.tmpDir(.{});
858 defer tmp.cleanup();
859
860 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
861 .paths = .{ .database = "merge.db", .wal = "merge.wal" },
862 .header = testingHeader(),
863 });
864 defer database.deinit();
865 try database.reserve(.{ .wal_frames = 768 });
866
867 var catalog = try catalog_mod.Catalog.open(&database, .{});
868 _ = try catalog.createRelation(std.testing.allocator, .{
869 .name = "base",
870 .columns = &.{.{ .name = "value" }},
871 }, .{ .durability = .buffered });
872 _ = try catalog.createRelation(std.testing.allocator, .{
873 .name = "ours",
874 .columns = &.{.{ .name = "value" }},
875 }, .{ .durability = .buffered });
876 _ = try catalog.createRelation(std.testing.allocator, .{
877 .name = "theirs",
878 .columns = &.{.{ .name = "value" }},
879 }, .{ .durability = .buffered });
880
881 var base = try catalog.openRelation(std.testing.allocator, "base");
882 defer base.deinit();
883 var ours = try catalog.openRelation(std.testing.allocator, "ours");
884 defer ours.deinit();
885 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
886 defer theirs.deinit();
887
888 _ = try base.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
889 _ = try base.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
890 _ = try base.relation.put(std.testing.allocator, 3, &.{.{ .text = "delete" }}, .{ .durability = .buffered });
891
892 _ = try ours.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
893 _ = try ours.relation.put(std.testing.allocator, 2, &.{.{ .text = "ours" }}, .{ .durability = .buffered });
894 _ = try ours.relation.put(std.testing.allocator, 3, &.{.{ .text = "delete" }}, .{ .durability = .buffered });
895 _ = try ours.relation.put(std.testing.allocator, 5, &.{.{ .text = "ours-add" }}, .{ .durability = .buffered });
896
897 _ = try theirs.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
898 _ = try theirs.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
899 _ = try theirs.relation.put(std.testing.allocator, 6, &.{.{ .text = "theirs-add" }}, .{ .durability = .buffered });
900
901 const schema = try catalog.schemaState(std.testing.allocator);
902 var base_root = try version.relationRoot(std.testing.allocator, "base", schema, &base, null);
903 defer base_root.deinit();
904 var ours_root = try version.relationRoot(std.testing.allocator, "ours", schema, &ours, null);
905 defer ours_root.deinit();
906 var theirs_root = try version.relationRoot(std.testing.allocator, "theirs", schema, &theirs, null);
907 defer theirs_root.deinit();
908
909 var result = try relation(std.testing.allocator, .{
910 .root = &base_root,
911 .rows = .{ .live = &base },
912 }, .{
913 .root = &ours_root,
914 .rows = .{ .live = &ours },
915 }, .{
916 .root = &theirs_root,
917 .rows = .{ .live = &theirs },
918 });
919 defer result.deinit();
920
921 try std.testing.expect(!result.hasConflicts());
922 try std.testing.expectEqual(@as(usize, 4), result.edits.len);
923 try std.testing.expectEqual(EditKind.modified, result.edits[0].kind);
924 try std.testing.expectEqual(@as(i64, 2), result.edits[0].rowid);
925 try std.testing.expectEqual(EditKind.removed, result.edits[1].kind);
926 try std.testing.expectEqual(@as(i64, 3), result.edits[1].rowid);
927 try std.testing.expectEqual(EditKind.added, result.edits[2].kind);
928 try std.testing.expectEqual(@as(i64, 5), result.edits[2].rowid);
929 try std.testing.expectEqual(EditKind.added, result.edits[3].kind);
930 try std.testing.expectEqual(@as(i64, 6), result.edits[3].rowid);
931 }
932
933 test "relation merge records divergent same row conflicts" {
934 var tmp = std.testing.tmpDir(.{});
935 defer tmp.cleanup();
936
937 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
938 .paths = .{ .database = "merge-conflict.db", .wal = "merge-conflict.wal" },
939 .header = testingHeader(),
940 });
941 defer database.deinit();
942 try database.reserve(.{ .wal_frames = 768 });
943
944 var catalog = try catalog_mod.Catalog.open(&database, .{});
945 _ = try catalog.createRelation(std.testing.allocator, .{
946 .name = "base",
947 .columns = &.{.{ .name = "value" }},
948 }, .{ .durability = .buffered });
949 _ = try catalog.createRelation(std.testing.allocator, .{
950 .name = "ours",
951 .columns = &.{.{ .name = "value" }},
952 }, .{ .durability = .buffered });
953 _ = try catalog.createRelation(std.testing.allocator, .{
954 .name = "theirs",
955 .columns = &.{.{ .name = "value" }},
956 }, .{ .durability = .buffered });
957
958 var base = try catalog.openRelation(std.testing.allocator, "base");
959 defer base.deinit();
960 var ours = try catalog.openRelation(std.testing.allocator, "ours");
961 defer ours.deinit();
962 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
963 defer theirs.deinit();
964
965 _ = try base.relation.put(std.testing.allocator, 1, &.{.{ .text = "base" }}, .{ .durability = .buffered });
966 _ = try ours.relation.put(std.testing.allocator, 1, &.{.{ .text = "ours" }}, .{ .durability = .buffered });
967 _ = try theirs.relation.put(std.testing.allocator, 1, &.{.{ .text = "theirs" }}, .{ .durability = .buffered });
968
969 const schema = try catalog.schemaState(std.testing.allocator);
970 var base_root = try version.relationRoot(std.testing.allocator, "base", schema, &base, null);
971 defer base_root.deinit();
972 var ours_root = try version.relationRoot(std.testing.allocator, "ours", schema, &ours, null);
973 defer ours_root.deinit();
974 var theirs_root = try version.relationRoot(std.testing.allocator, "theirs", schema, &theirs, null);
975 defer theirs_root.deinit();
976
977 var result = try relation(std.testing.allocator, .{
978 .root = &base_root,
979 .rows = .{ .live = &base },
980 }, .{
981 .root = &ours_root,
982 .rows = .{ .live = &ours },
983 }, .{
984 .root = &theirs_root,
985 .rows = .{ .live = &theirs },
986 });
987 defer result.deinit();
988
989 try std.testing.expect(result.hasConflicts());
990 try std.testing.expectEqual(@as(usize, 0), result.edits.len);
991 try std.testing.expectEqual(@as(usize, 1), result.conflicts.len);
992 try std.testing.expectEqualStrings("ours", result.conflicts[0].relation);
993 try std.testing.expectEqual(@as(i64, 1), result.conflicts[0].rowid);
994 const expected = version.ConflictArtifact.init("ours", 1, result.conflicts[0].base, result.conflicts[0].ours, result.conflicts[0].theirs);
995 try std.testing.expect(version.same(expected.hash, result.conflicts[0].artifact.hash));
996 }
997
998 test "relation merge coalesces identical concurrent edits" {
999 var tmp = std.testing.tmpDir(.{});
1000 defer tmp.cleanup();
1001
1002 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1003 .paths = .{ .database = "merge-same.db", .wal = "merge-same.wal" },
1004 .header = testingHeader(),
1005 });
1006 defer database.deinit();
1007 try database.reserve(.{ .wal_frames = 768 });
1008
1009 var catalog = try catalog_mod.Catalog.open(&database, .{});
1010 _ = try catalog.createRelation(std.testing.allocator, .{
1011 .name = "base",
1012 .columns = &.{.{ .name = "value" }},
1013 }, .{ .durability = .buffered });
1014 _ = try catalog.createRelation(std.testing.allocator, .{
1015 .name = "ours",
1016 .columns = &.{.{ .name = "value" }},
1017 }, .{ .durability = .buffered });
1018 _ = try catalog.createRelation(std.testing.allocator, .{
1019 .name = "theirs",
1020 .columns = &.{.{ .name = "value" }},
1021 }, .{ .durability = .buffered });
1022
1023 var base = try catalog.openRelation(std.testing.allocator, "base");
1024 defer base.deinit();
1025 var ours = try catalog.openRelation(std.testing.allocator, "ours");
1026 defer ours.deinit();
1027 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
1028 defer theirs.deinit();
1029
1030 _ = try base.relation.put(std.testing.allocator, 1, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1031 _ = try ours.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1032 _ = try theirs.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1033
1034 const schema = try catalog.schemaState(std.testing.allocator);
1035 var base_root = try version.relationRoot(std.testing.allocator, "base", schema, &base, null);
1036 defer base_root.deinit();
1037 var ours_root = try version.relationRoot(std.testing.allocator, "ours", schema, &ours, null);
1038 defer ours_root.deinit();
1039 var theirs_root = try version.relationRoot(std.testing.allocator, "theirs", schema, &theirs, null);
1040 defer theirs_root.deinit();
1041
1042 var result = try relation(std.testing.allocator, .{
1043 .root = &base_root,
1044 .rows = .{ .live = &base },
1045 }, .{
1046 .root = &ours_root,
1047 .rows = .{ .live = &ours },
1048 }, .{
1049 .root = &theirs_root,
1050 .rows = .{ .live = &theirs },
1051 });
1052 defer result.deinit();
1053
1054 try std.testing.expect(!result.hasConflicts());
1055 try std.testing.expectEqual(@as(usize, 1), result.edits.len);
1056 try std.testing.expectEqual(EditKind.modified, result.edits[0].kind);
1057 try std.testing.expectEqual(@as(i64, 1), result.edits[0].rowid);
1058 }
1059
1060 test "database merge applies clean relation edits and derives root" {
1061 var tmp = std.testing.tmpDir(.{});
1062 defer tmp.cleanup();
1063
1064 var backing = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1065 .paths = .{ .database = "database-merge.db", .wal = "database-merge.wal" },
1066 .header = testingHeader(),
1067 });
1068 defer backing.deinit();
1069 try backing.reserve(.{ .wal_frames = 1024 });
1070
1071 var catalog = try catalog_mod.Catalog.open(&backing, .{});
1072 _ = try catalog.createRelation(std.testing.allocator, .{
1073 .name = "base",
1074 .columns = &.{.{ .name = "value" }},
1075 }, .{ .durability = .buffered });
1076 _ = try catalog.createRelation(std.testing.allocator, .{
1077 .name = "ours",
1078 .columns = &.{.{ .name = "value" }},
1079 }, .{ .durability = .buffered });
1080 _ = try catalog.createRelation(std.testing.allocator, .{
1081 .name = "theirs",
1082 .columns = &.{.{ .name = "value" }},
1083 }, .{ .durability = .buffered });
1084
1085 var base = try catalog.openRelation(std.testing.allocator, "base");
1086 defer base.deinit();
1087 var ours = try catalog.openRelation(std.testing.allocator, "ours");
1088 defer ours.deinit();
1089 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
1090 defer theirs.deinit();
1091
1092 _ = try base.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1093 _ = try base.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1094 _ = try base.relation.put(std.testing.allocator, 3, &.{.{ .text = "delete" }}, .{ .durability = .buffered });
1095
1096 _ = try ours.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1097 _ = try ours.relation.put(std.testing.allocator, 2, &.{.{ .text = "ours" }}, .{ .durability = .buffered });
1098 _ = try ours.relation.put(std.testing.allocator, 3, &.{.{ .text = "delete" }}, .{ .durability = .buffered });
1099 _ = try ours.relation.put(std.testing.allocator, 5, &.{.{ .text = "ours-add" }}, .{ .durability = .buffered });
1100
1101 _ = try theirs.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1102 _ = try theirs.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1103 _ = try theirs.relation.put(std.testing.allocator, 6, &.{.{ .text = "theirs-add" }}, .{ .durability = .buffered });
1104
1105 const schema = try catalog.schemaState(std.testing.allocator);
1106 var base_root = try version.relationRoot(std.testing.allocator, "items", schema, &base, null);
1107 defer base_root.deinit();
1108 var theirs_root = try version.relationRoot(std.testing.allocator, "items", schema, &theirs, null);
1109 defer theirs_root.deinit();
1110 var ours_value = try version.relationValue(std.testing.allocator, "items", schema, &ours, null);
1111 defer ours_value.deinit(std.testing.allocator);
1112 var ours_database = try databaseValueForRelation(std.testing.allocator, &ours_value);
1113 defer ours_database.deinit();
1114
1115 var result = try mergeDatabase(std.testing.allocator, .{
1116 .relations = &.{.{ .root = &base_root, .rows = .{ .live = &base } }},
1117 .conflicts = .empty(),
1118 }, .{ .value = &ours_database, .conflicts = .empty() }, .{
1119 .relations = &.{.{ .root = &theirs_root, .rows = .{ .live = &theirs } }},
1120 .conflicts = .empty(),
1121 });
1122 defer result.deinit();
1123
1124 try std.testing.expect(!result.hasConflicts());
1125 try std.testing.expectEqual(@as(usize, 1), result.relations.len);
1126 try std.testing.expectEqual(@as(usize, 4), result.relations[0].result.edits.len);
1127 try std.testing.expectEqual(@as(usize, 0), result.discovered.len);
1128 try std.testing.expectEqual(@as(usize, 0), result.conflict_root.count);
1129 try std.testing.expect(version.same(result.value.root.conflicts, result.conflict_root.hash));
1130 try std.testing.expectEqual(@as(usize, 1), result.value.relations.len);
1131 try std.testing.expectEqual(@as(usize, 4), result.value.relations[0].rows.len);
1132
1133 const live_added = try ours.relation.get(std.testing.allocator, 6);
1134 if (live_added) |bytes| std.testing.allocator.free(bytes);
1135 try std.testing.expect(live_added == null);
1136 const live_removed = (try ours.relation.get(std.testing.allocator, 3)).?;
1137 std.testing.allocator.free(live_removed);
1138 const added = relationRowBytes(result.value.relations[0].rows, 6).?;
1139 const added_view = try row.View.init(added);
1140 try std.testing.expectEqualStrings("theirs-add", (try added_view.column(0)).text);
1141 try std.testing.expect(relationRowBytes(result.value.relations[0].rows, 3) == null);
1142
1143 var expected = try version.DatabaseRoot.initSorted(std.testing.allocator, &.{.{
1144 .name = "items",
1145 .hash = result.relations[0].root,
1146 }}, version.ConflictRoot.empty());
1147 defer expected.deinit();
1148 try std.testing.expect(version.same(expected.hash, result.value.root.hash));
1149 }
1150
1151 test "database merge supersedes stale slots and persists complete conflict root" {
1152 var tmp = std.testing.tmpDir(.{});
1153 defer tmp.cleanup();
1154
1155 var backing = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1156 .paths = .{ .database = "database-conflict.db", .wal = "database-conflict.wal" },
1157 .header = testingHeader(),
1158 });
1159 defer backing.deinit();
1160 try backing.reserve(.{ .wal_frames = 1024 });
1161
1162 var catalog = try catalog_mod.Catalog.open(&backing, .{});
1163 _ = try catalog.createRelation(std.testing.allocator, .{
1164 .name = "base",
1165 .columns = &.{.{ .name = "value" }},
1166 }, .{ .durability = .buffered });
1167 _ = try catalog.createRelation(std.testing.allocator, .{
1168 .name = "ours",
1169 .columns = &.{.{ .name = "value" }},
1170 }, .{ .durability = .buffered });
1171 _ = try catalog.createRelation(std.testing.allocator, .{
1172 .name = "theirs",
1173 .columns = &.{.{ .name = "value" }},
1174 }, .{ .durability = .buffered });
1175
1176 var base = try catalog.openRelation(std.testing.allocator, "base");
1177 defer base.deinit();
1178 var ours = try catalog.openRelation(std.testing.allocator, "ours");
1179 defer ours.deinit();
1180 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
1181 defer theirs.deinit();
1182
1183 _ = try base.relation.put(std.testing.allocator, 1, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1184 _ = try ours.relation.put(std.testing.allocator, 1, &.{.{ .text = "ours" }}, .{ .durability = .buffered });
1185 _ = try theirs.relation.put(std.testing.allocator, 1, &.{.{ .text = "theirs" }}, .{ .durability = .buffered });
1186
1187 const schema = try catalog.schemaState(std.testing.allocator);
1188 var base_root = try version.relationRoot(std.testing.allocator, "items", schema, &base, null);
1189 defer base_root.deinit();
1190 var theirs_root = try version.relationRoot(std.testing.allocator, "items", schema, &theirs, null);
1191 defer theirs_root.deinit();
1192 var ours_value = try version.relationValue(std.testing.allocator, "items", schema, &ours, null);
1193 defer ours_value.deinit(std.testing.allocator);
1194 var ours_database = try databaseValueForRelation(std.testing.allocator, &ours_value);
1195 defer ours_database.deinit();
1196
1197 const carried_artifacts = [_]version.ConflictArtifact{
1198 version.ConflictArtifact.init("items", 1, "old-base", "old-ours", "old-theirs"),
1199 version.ConflictArtifact.init("items", 2, "base-two", "ours-two", "theirs-two"),
1200 };
1201 const carried_entries = [_]version.ConflictEntry{
1202 carried_artifacts[0].entry(),
1203 carried_artifacts[1].entry(),
1204 };
1205 const carried_root = version.ConflictRoot.init(&carried_entries);
1206 const adjusted_ours_root = try version.DatabaseRoot.initSorted(std.testing.allocator, ours_database.root.entries, carried_root);
1207 ours_database.root.deinit();
1208 ours_database.root = adjusted_ours_root;
1209 const carried = ConflictSnapshot{
1210 .root = carried_root,
1211 .artifacts = &carried_artifacts,
1212 };
1213
1214 var result = try mergeDatabase(std.testing.allocator, .{
1215 .relations = &.{.{ .root = &base_root, .rows = .{ .live = &base } }},
1216 .conflicts = carried,
1217 }, .{ .value = &ours_database, .conflicts = carried }, .{
1218 .relations = &.{.{ .root = &theirs_root, .rows = .{ .live = &theirs } }},
1219 .conflicts = carried,
1220 });
1221 defer result.deinit();
1222
1223 try std.testing.expect(result.hasConflicts());
1224 try std.testing.expectEqual(@as(usize, 1), result.discovered.len);
1225 try std.testing.expectEqual(@as(usize, 2), result.conflict_root.count);
1226 try std.testing.expectEqual(@as(usize, 2), result.artifacts.len);
1227 try std.testing.expect(version.same(result.value.root.conflicts, result.conflict_root.hash));
1228 try std.testing.expectEqualStrings("items", result.discovered[0].artifact.relation);
1229 try std.testing.expect(!hasConflictArtifact(result.artifacts, carried_artifacts[0].entry()));
1230 try std.testing.expect(hasConflictArtifact(result.artifacts, carried_artifacts[1].entry()));
1231 try std.testing.expect(hasConflictArtifact(result.artifacts, result.discovered[0].artifact.entry()));
1232
1233 var history = try history_mod.History.open(std.testing.allocator, tmp.dir, .{ .path = "database-conflict.history", .recovery = .reject });
1234 defer history.deinit();
1235 try result.persistConflicts(&history);
1236
1237 const recovered = history.conflict(result.discovered[0].artifact.hash).?;
1238 try std.testing.expectEqualStrings("items", recovered.relation);
1239 const recovered_ours = try row.View.init(recovered.ours.?.row);
1240 try std.testing.expectEqualStrings("ours", (try recovered_ours.column(0)).text);
1241 var root_entries = try history.conflictEntries(std.testing.allocator, result.conflict_root.hash);
1242 defer root_entries.deinit();
1243 try std.testing.expect(version.same(result.conflict_root.hash, root_entries.root.hash));
1244 try std.testing.expectEqual(@as(usize, 2), root_entries.entries.len);
1245 try std.testing.expect(version.same(result.discovered[0].artifact.hash, root_entries.entries[0].hash));
1246 try std.testing.expect(version.same(carried_artifacts[1].hash, root_entries.entries[1].hash));
1247 }
1248
1249 test "database merge matches relation names and preserves independent relation set changes" {
1250 var tmp = std.testing.tmpDir(.{});
1251 defer tmp.cleanup();
1252
1253 var backing = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1254 .paths = .{ .database = "database-relation-set.db", .wal = "database-relation-set.wal" },
1255 .header = testingHeader(),
1256 });
1257 defer backing.deinit();
1258 try backing.reserve(.{ .wal_frames = 1600 });
1259
1260 var catalog = try catalog_mod.Catalog.open(&backing, .{});
1261 const names = [_][]const u8{
1262 "base_items",
1263 "ours_items",
1264 "theirs_items",
1265 "base_dropped",
1266 "base_drop_on_ours",
1267 "theirs_drop_on_ours",
1268 "base_drop_on_theirs",
1269 "ours_drop_on_theirs",
1270 "ours_only",
1271 "theirs_only",
1272 };
1273 for (names) |name| {
1274 _ = try catalog.createRelation(std.testing.allocator, .{
1275 .name = name,
1276 .columns = &.{.{ .name = "value" }},
1277 }, .{ .durability = .buffered });
1278 }
1279
1280 var base_items = try catalog.openRelation(std.testing.allocator, "base_items");
1281 defer base_items.deinit();
1282 var ours_items = try catalog.openRelation(std.testing.allocator, "ours_items");
1283 defer ours_items.deinit();
1284 var theirs_items = try catalog.openRelation(std.testing.allocator, "theirs_items");
1285 defer theirs_items.deinit();
1286 var base_dropped = try catalog.openRelation(std.testing.allocator, "base_dropped");
1287 defer base_dropped.deinit();
1288 var base_drop_on_ours = try catalog.openRelation(std.testing.allocator, "base_drop_on_ours");
1289 defer base_drop_on_ours.deinit();
1290 var theirs_drop_on_ours = try catalog.openRelation(std.testing.allocator, "theirs_drop_on_ours");
1291 defer theirs_drop_on_ours.deinit();
1292 var base_drop_on_theirs = try catalog.openRelation(std.testing.allocator, "base_drop_on_theirs");
1293 defer base_drop_on_theirs.deinit();
1294 var ours_drop_on_theirs = try catalog.openRelation(std.testing.allocator, "ours_drop_on_theirs");
1295 defer ours_drop_on_theirs.deinit();
1296 var ours_only = try catalog.openRelation(std.testing.allocator, "ours_only");
1297 defer ours_only.deinit();
1298 var theirs_only = try catalog.openRelation(std.testing.allocator, "theirs_only");
1299 defer theirs_only.deinit();
1300
1301 _ = try base_items.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1302 _ = try base_items.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1303 _ = try ours_items.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1304 _ = try ours_items.relation.put(std.testing.allocator, 2, &.{.{ .text = "ours" }}, .{ .durability = .buffered });
1305 _ = try theirs_items.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
1306 _ = try theirs_items.relation.put(std.testing.allocator, 2, &.{.{ .text = "base" }}, .{ .durability = .buffered });
1307 _ = try theirs_items.relation.put(std.testing.allocator, 3, &.{.{ .text = "theirs" }}, .{ .durability = .buffered });
1308 _ = try base_dropped.relation.put(std.testing.allocator, 9, &.{.{ .text = "dropped" }}, .{ .durability = .buffered });
1309 _ = try base_drop_on_ours.relation.put(std.testing.allocator, 10, &.{.{ .text = "drop-on-ours" }}, .{ .durability = .buffered });
1310 _ = try theirs_drop_on_ours.relation.put(std.testing.allocator, 10, &.{.{ .text = "drop-on-ours" }}, .{ .durability = .buffered });
1311 _ = try base_drop_on_theirs.relation.put(std.testing.allocator, 11, &.{.{ .text = "drop-on-theirs" }}, .{ .durability = .buffered });
1312 _ = try ours_drop_on_theirs.relation.put(std.testing.allocator, 11, &.{.{ .text = "drop-on-theirs" }}, .{ .durability = .buffered });
1313 _ = try ours_only.relation.put(std.testing.allocator, 7, &.{.{ .text = "ours-only" }}, .{ .durability = .buffered });
1314 _ = try theirs_only.relation.put(std.testing.allocator, 8, &.{.{ .text = "theirs-only" }}, .{ .durability = .buffered });
1315
1316 const schema = try catalog.schemaState(std.testing.allocator);
1317 var base_items_root = try version.relationRoot(std.testing.allocator, "items", schema, &base_items, null);
1318 defer base_items_root.deinit();
1319 var base_dropped_root = try version.relationRoot(std.testing.allocator, "dropped", schema, &base_dropped, null);
1320 defer base_dropped_root.deinit();
1321 var base_drop_on_ours_root = try version.relationRoot(std.testing.allocator, "drop_on_ours", schema, &base_drop_on_ours, null);
1322 defer base_drop_on_ours_root.deinit();
1323 var theirs_drop_on_ours_root = try version.relationRoot(std.testing.allocator, "drop_on_ours", schema, &theirs_drop_on_ours, null);
1324 defer theirs_drop_on_ours_root.deinit();
1325 var base_drop_on_theirs_root = try version.relationRoot(std.testing.allocator, "drop_on_theirs", schema, &base_drop_on_theirs, null);
1326 defer base_drop_on_theirs_root.deinit();
1327 var theirs_items_root = try version.relationRoot(std.testing.allocator, "items", schema, &theirs_items, null);
1328 defer theirs_items_root.deinit();
1329 var theirs_only_root = try version.relationRoot(std.testing.allocator, "theirs_only", schema, &theirs_only, null);
1330 defer theirs_only_root.deinit();
1331 var ours_items_value = try version.relationValue(std.testing.allocator, "items", schema, &ours_items, null);
1332 defer ours_items_value.deinit(std.testing.allocator);
1333 var ours_only_value = try version.relationValue(std.testing.allocator, "ours_only", schema, &ours_only, null);
1334 defer ours_only_value.deinit(std.testing.allocator);
1335 var ours_drop_on_theirs_value = try version.relationValue(std.testing.allocator, "drop_on_theirs", schema, &ours_drop_on_theirs, null);
1336 defer ours_drop_on_theirs_value.deinit(std.testing.allocator);
1337 const ours_values = [_]*const version.RelationValue{ &ours_items_value, &ours_drop_on_theirs_value, &ours_only_value };
1338 var ours_database = try databaseValueForRelations(std.testing.allocator, &ours_values);
1339 defer ours_database.deinit();
1340
1341 var result = try mergeDatabase(std.testing.allocator, .{
1342 .relations = &.{
1343 .{ .root = &base_items_root, .rows = .{ .live = &base_items } },
1344 .{ .root = &base_dropped_root, .rows = .{ .live = &base_dropped } },
1345 .{ .root = &base_drop_on_ours_root, .rows = .{ .live = &base_drop_on_ours } },
1346 .{ .root = &base_drop_on_theirs_root, .rows = .{ .live = &base_drop_on_theirs } },
1347 },
1348 .conflicts = .empty(),
1349 }, .{ .value = &ours_database, .conflicts = .empty() }, .{
1350 .relations = &.{
1351 .{ .root = &theirs_only_root, .rows = .{ .live = &theirs_only } },
1352 .{ .root = &theirs_drop_on_ours_root, .rows = .{ .live = &theirs_drop_on_ours } },
1353 .{ .root = &theirs_items_root, .rows = .{ .live = &theirs_items } },
1354 },
1355 .conflicts = .empty(),
1356 });
1357 defer result.deinit();
1358
1359 try std.testing.expect(!result.hasConflicts());
1360 try std.testing.expectEqual(@as(usize, 3), result.value.relations.len);
1361 const items = result.value.findRelation("items").?;
1362 const ours_added = result.value.findRelation("ours_only").?;
1363 const theirs_added = result.value.findRelation("theirs_only").?;
1364 try std.testing.expect(result.value.findRelation("dropped") == null);
1365 try std.testing.expect(result.value.findRelation("drop_on_ours") == null);
1366 try std.testing.expect(result.value.findRelation("drop_on_theirs") == null);
1367 try std.testing.expectEqual(@as(usize, 3), items.rows.len);
1368 const row_two = try row.View.init(relationRowBytes(items.rows, 2).?);
1369 try std.testing.expectEqualStrings("ours", (try row_two.column(0)).text);
1370 const row_three = try row.View.init(relationRowBytes(items.rows, 3).?);
1371 try std.testing.expectEqualStrings("theirs", (try row_three.column(0)).text);
1372 const ours_added_view = try row.View.init(relationRowBytes(ours_added.rows, 7).?);
1373 try std.testing.expectEqualStrings("ours-only", (try ours_added_view.column(0)).text);
1374 const theirs_added_view = try row.View.init(relationRowBytes(theirs_added.rows, 8).?);
1375 try std.testing.expectEqualStrings("theirs-only", (try theirs_added_view.column(0)).text);
1376
1377 var expected = try version.DatabaseRoot.initSorted(std.testing.allocator, &.{
1378 .{ .name = "items", .hash = items.root.hash },
1379 .{ .name = "ours_only", .hash = ours_added.root.hash },
1380 .{ .name = "theirs_only", .hash = theirs_added.root.hash },
1381 }, version.ConflictRoot.empty());
1382 defer expected.deinit();
1383 try std.testing.expect(version.same(expected.hash, result.value.root.hash));
1384 }
1385
1386 test "database merge records relation topology conflicts under database root" {
1387 var tmp = std.testing.tmpDir(.{});
1388 defer tmp.cleanup();
1389
1390 var backing = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1391 .paths = .{ .database = "database-relation-conflicts.db", .wal = "database-relation-conflicts.wal" },
1392 .header = testingHeader(),
1393 });
1394 defer backing.deinit();
1395 try backing.reserve(.{ .wal_frames = 1600 });
1396
1397 var catalog = try catalog_mod.Catalog.open(&backing, .{});
1398 const names = [_][]const u8{
1399 "base_changed_deleted",
1400 "theirs_changed_deleted",
1401 "base_deleted_changed",
1402 "ours_deleted_changed",
1403 "ours_added",
1404 "theirs_added",
1405 };
1406 for (names) |name| {
1407 _ = try catalog.createRelation(std.testing.allocator, .{
1408 .name = name,
1409 .columns = &.{.{ .name = "value" }},
1410 }, .{ .durability = .buffered });
1411 }
1412
1413 var base_changed_deleted = try catalog.openRelation(std.testing.allocator, "base_changed_deleted");
1414 defer base_changed_deleted.deinit();
1415 var theirs_changed_deleted = try catalog.openRelation(std.testing.allocator, "theirs_changed_deleted");
1416 defer theirs_changed_deleted.deinit();
1417 var base_deleted_changed = try catalog.openRelation(std.testing.allocator, "base_deleted_changed");
1418 defer base_deleted_changed.deinit();
1419 var ours_deleted_changed = try catalog.openRelation(std.testing.allocator, "ours_deleted_changed");
1420 defer ours_deleted_changed.deinit();
1421 var ours_added = try catalog.openRelation(std.testing.allocator, "ours_added");
1422 defer ours_added.deinit();
1423 var theirs_added = try catalog.openRelation(std.testing.allocator, "theirs_added");
1424 defer theirs_added.deinit();
1425
1426 _ = try base_changed_deleted.relation.put(std.testing.allocator, 1, &.{.{ .text = "base-changed-deleted" }}, .{ .durability = .buffered });
1427 _ = try theirs_changed_deleted.relation.put(std.testing.allocator, 1, &.{.{ .text = "theirs-changed-deleted" }}, .{ .durability = .buffered });
1428 _ = try base_deleted_changed.relation.put(std.testing.allocator, 2, &.{.{ .text = "base-deleted-changed" }}, .{ .durability = .buffered });
1429 _ = try ours_deleted_changed.relation.put(std.testing.allocator, 2, &.{.{ .text = "ours-deleted-changed" }}, .{ .durability = .buffered });
1430 _ = try ours_added.relation.put(std.testing.allocator, 3, &.{.{ .text = "ours-added" }}, .{ .durability = .buffered });
1431 _ = try theirs_added.relation.put(std.testing.allocator, 3, &.{.{ .text = "theirs-added" }}, .{ .durability = .buffered });
1432
1433 const schema = try catalog.schemaState(std.testing.allocator);
1434 var base_changed_deleted_root = try version.relationRoot(std.testing.allocator, "changed_deleted", schema, &base_changed_deleted, null);
1435 defer base_changed_deleted_root.deinit();
1436 var theirs_changed_deleted_root = try version.relationRoot(std.testing.allocator, "changed_deleted", schema, &theirs_changed_deleted, null);
1437 defer theirs_changed_deleted_root.deinit();
1438 var base_deleted_changed_root = try version.relationRoot(std.testing.allocator, "deleted_changed", schema, &base_deleted_changed, null);
1439 defer base_deleted_changed_root.deinit();
1440 var theirs_added_root = try version.relationRoot(std.testing.allocator, "added", schema, &theirs_added, null);
1441 defer theirs_added_root.deinit();
1442 var ours_deleted_changed_value = try version.relationValue(std.testing.allocator, "deleted_changed", schema, &ours_deleted_changed, null);
1443 defer ours_deleted_changed_value.deinit(std.testing.allocator);
1444 var ours_added_value = try version.relationValue(std.testing.allocator, "added", schema, &ours_added, null);
1445 defer ours_added_value.deinit(std.testing.allocator);
1446 const ours_values = [_]*const version.RelationValue{ &ours_deleted_changed_value, &ours_added_value };
1447 var ours_database = try databaseValueForRelations(std.testing.allocator, &ours_values);
1448 defer ours_database.deinit();
1449
1450 var result = try mergeDatabase(std.testing.allocator, .{
1451 .relations = &.{
1452 .{ .root = &base_changed_deleted_root, .rows = .{ .live = &base_changed_deleted } },
1453 .{ .root = &base_deleted_changed_root, .rows = .{ .live = &base_deleted_changed } },
1454 },
1455 .conflicts = .empty(),
1456 }, .{ .value = &ours_database, .conflicts = .empty() }, .{
1457 .relations = &.{
1458 .{ .root = &theirs_changed_deleted_root, .rows = .{ .live = &theirs_changed_deleted } },
1459 .{ .root = &theirs_added_root, .rows = .{ .live = &theirs_added } },
1460 },
1461 .conflicts = .empty(),
1462 });
1463 defer result.deinit();
1464
1465 try std.testing.expect(result.hasConflicts());
1466 try std.testing.expectEqual(@as(usize, 0), result.relations.len);
1467 try std.testing.expectEqual(@as(usize, 3), result.discovered.len);
1468 try std.testing.expectEqual(@as(usize, 3), result.conflict_root.count);
1469 try std.testing.expect(version.same(result.value.root.conflicts, result.conflict_root.hash));
1470 try std.testing.expectEqual(@as(usize, 2), result.value.relations.len);
1471 try std.testing.expect(result.value.findRelation("changed_deleted") == null);
1472
1473 const kept_deleted_changed = result.value.findRelation("deleted_changed").?;
1474 const kept_deleted_changed_view = try row.View.init(relationRowBytes(kept_deleted_changed.rows, 2).?);
1475 try std.testing.expectEqualStrings("ours-deleted-changed", (try kept_deleted_changed_view.column(0)).text);
1476 const kept_added = result.value.findRelation("added").?;
1477 const kept_added_view = try row.View.init(relationRowBytes(kept_added.rows, 3).?);
1478 try std.testing.expectEqualStrings("ours-added", (try kept_added_view.column(0)).text);
1479
1480 const added_conflict = findDatabaseConflict(result.discovered, "added").?;
1481 try std.testing.expectEqual(version.ConflictKind.relation, added_conflict.kind);
1482 try std.testing.expect(added_conflict.rowid == null);
1483 try std.testing.expect(added_conflict.base_root == null);
1484 try expectOptionalHash(ours_added_value.root.hash, added_conflict.ours_root);
1485 try expectOptionalHash(theirs_added_root.hash, added_conflict.theirs_root);
1486 try std.testing.expect(added_conflict.artifact.base == null);
1487 try std.testing.expect(version.same(ours_added_value.root.hash, added_conflict.artifact.ours.?.relation));
1488 try std.testing.expect(version.same(theirs_added_root.hash, added_conflict.artifact.theirs.?.relation));
1489
1490 const changed_deleted_conflict = findDatabaseConflict(result.discovered, "changed_deleted").?;
1491 try std.testing.expectEqual(version.ConflictKind.relation, changed_deleted_conflict.kind);
1492 try expectOptionalHash(base_changed_deleted_root.hash, changed_deleted_conflict.base_root);
1493 try std.testing.expect(changed_deleted_conflict.ours_root == null);
1494 try expectOptionalHash(theirs_changed_deleted_root.hash, changed_deleted_conflict.theirs_root);
1495
1496 const deleted_changed_conflict = findDatabaseConflict(result.discovered, "deleted_changed").?;
1497 try std.testing.expectEqual(version.ConflictKind.relation, deleted_changed_conflict.kind);
1498 try expectOptionalHash(base_deleted_changed_root.hash, deleted_changed_conflict.base_root);
1499 try expectOptionalHash(ours_deleted_changed_value.root.hash, deleted_changed_conflict.ours_root);
1500 try std.testing.expect(deleted_changed_conflict.theirs_root == null);
1501
1502 var expected = try version.DatabaseRoot.initSorted(std.testing.allocator, &.{
1503 .{ .name = "added", .hash = kept_added.root.hash },
1504 .{ .name = "deleted_changed", .hash = kept_deleted_changed.root.hash },
1505 }, result.conflict_root);
1506 defer expected.deinit();
1507 try std.testing.expect(version.same(expected.hash, result.value.root.hash));
1508 }
1509
1510 test "database merge preserves a theirs-only index schema" {
1511 var tmp = std.testing.tmpDir(.{});
1512 defer tmp.cleanup();
1513
1514 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1515 .paths = .{ .database = "schema-select.db", .wal = "schema-select.wal" },
1516 .header = testingHeader(),
1517 });
1518 defer database.deinit();
1519 try database.reserve(.{ .wal_frames = 768 });
1520
1521 var catalog = try catalog_mod.Catalog.open(&database, .{});
1522 _ = try catalog.createRelation(std.testing.allocator, .{
1523 .name = "base",
1524 .columns = &.{.{ .name = "value" }},
1525 }, .{ .durability = .buffered });
1526 _ = try catalog.createRelation(std.testing.allocator, .{
1527 .name = "theirs",
1528 .columns = &.{.{ .name = "value" }},
1529 }, .{ .durability = .buffered });
1530 _ = try catalog.createIndex(std.testing.allocator, "theirs", .{
1531 .name = "theirs_value",
1532 .fields = &.{0},
1533 }, .{ .durability = .buffered });
1534
1535 var base = try catalog.openRelation(std.testing.allocator, "base");
1536 defer base.deinit();
1537 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
1538 defer theirs.deinit();
1539 const schema = try catalog.schemaState(std.testing.allocator);
1540 var base_value = try version.relationValue(std.testing.allocator, "items", schema, &base, null);
1541 defer base_value.deinit(std.testing.allocator);
1542 var theirs_value = try version.relationValue(std.testing.allocator, "items", schema, &theirs, null);
1543 defer theirs_value.deinit(std.testing.allocator);
1544 var ours_database = try databaseValueForRelation(std.testing.allocator, &base_value);
1545 defer ours_database.deinit();
1546
1547 var result = try mergeDatabase(std.testing.allocator, .{
1548 .relations = &.{.{ .root = &base_value.root, .rows = .{ .materialized = base_value.rows } }},
1549 .conflicts = .empty(),
1550 }, .{ .value = &ours_database, .conflicts = .empty() }, .{
1551 .relations = &.{.{ .root = &theirs_value.root, .rows = .{ .materialized = theirs_value.rows } }},
1552 .conflicts = .empty(),
1553 });
1554 defer result.deinit();
1555
1556 try std.testing.expect(!result.hasConflicts());
1557 try std.testing.expectEqual(RelationMergeMode.theirs, result.relations[0].mode);
1558 try std.testing.expect(version.same(theirs_value.root.hash, result.relations[0].root));
1559 try std.testing.expect(version.same(theirs_value.root.schema, result.value.relations[0].root.schema));
1560 try std.testing.expectEqual(@as(usize, 1), result.value.relations[0].root.indexes.len);
1561 }
1562
1563 test "database merge contains cross-schema row divergence" {
1564 var tmp = std.testing.tmpDir(.{});
1565 defer tmp.cleanup();
1566
1567 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
1568 .paths = .{ .database = "schema-conflict.db", .wal = "schema-conflict.wal" },
1569 .header = testingHeader(),
1570 });
1571 defer database.deinit();
1572 try database.reserve(.{ .wal_frames = 768 });
1573
1574 var catalog = try catalog_mod.Catalog.open(&database, .{});
1575 try createMergeTestRelation(&catalog, "base", false, "base");
1576 try createMergeTestRelation(&catalog, "ours", false, "ours");
1577 try createMergeTestRelation(&catalog, "theirs", true, "theirs");
1578
1579 var base = try catalog.openRelation(std.testing.allocator, "base");
1580 defer base.deinit();
1581 var ours = try catalog.openRelation(std.testing.allocator, "ours");
1582 defer ours.deinit();
1583 var theirs = try catalog.openRelation(std.testing.allocator, "theirs");
1584 defer theirs.deinit();
1585 const schema = try catalog.schemaState(std.testing.allocator);
1586 var base_value = try version.relationValue(std.testing.allocator, "items", schema, &base, null);
1587 defer base_value.deinit(std.testing.allocator);
1588 var ours_value = try version.relationValue(std.testing.allocator, "items", schema, &ours, null);
1589 defer ours_value.deinit(std.testing.allocator);
1590 var theirs_value = try version.relationValue(std.testing.allocator, "items", schema, &theirs, null);
1591 defer theirs_value.deinit(std.testing.allocator);
1592 var ours_database = try databaseValueForRelation(std.testing.allocator, &ours_value);
1593 defer ours_database.deinit();
1594
1595 const base_snapshot = RelationSnapshot{
1596 .root = &base_value.root,
1597 .rows = .{ .materialized = base_value.rows },
1598 };
1599 const theirs_snapshot = RelationSnapshot{
1600 .root = &theirs_value.root,
1601 .rows = .{ .materialized = theirs_value.rows },
1602 };
1603 try std.testing.expectError(error.SchemaMismatch, relation(
1604 std.testing.allocator,
1605 base_snapshot,
1606 .{ .root = &ours_value.root, .rows = .{ .materialized = ours_value.rows } },
1607 theirs_snapshot,
1608 ));
1609
1610 var result = try mergeDatabase(std.testing.allocator, .{
1611 .relations = &.{base_snapshot},
1612 .conflicts = .empty(),
1613 }, .{ .value = &ours_database, .conflicts = .empty() }, .{
1614 .relations = &.{theirs_snapshot},
1615 .conflicts = .empty(),
1616 });
1617 defer result.deinit();
1618
1619 try std.testing.expect(result.hasConflicts());
1620 try std.testing.expectEqual(RelationMergeMode.conflict, result.relations[0].mode);
1621 try std.testing.expect(version.same(ours_value.root.hash, result.relations[0].root));
1622 try std.testing.expectEqual(@as(usize, 0), result.relations[0].result.edits.len);
1623 try std.testing.expectEqual(@as(usize, 1), result.discovered.len);
1624 try std.testing.expectEqual(version.ConflictKind.relation, result.discovered[0].kind);
1625 try std.testing.expect(version.same(base_value.root.hash, result.discovered[0].base_root.?));
1626 try std.testing.expect(version.same(ours_value.root.hash, result.discovered[0].ours_root.?));
1627 try std.testing.expect(version.same(theirs_value.root.hash, result.discovered[0].theirs_root.?));
1628 }
1629
1630 fn relationRowBytes(rows: []const version.RelationRow, rowid: i64) ?[]const u8 {
1631 for (rows) |row_value| {
1632 if (row_value.rowid == rowid) return row_value.bytes;
1633 }
1634 return null;
1635 }
1636
1637 fn createMergeTestRelation(
1638 catalog: *const catalog_mod.Catalog,
1639 name: []const u8,
1640 two_columns: bool,
1641 value: []const u8,
1642 ) !void {
1643 if (two_columns) {
1644 _ = try catalog.createRelation(std.testing.allocator, .{
1645 .name = name,
1646 .columns = &.{ .{ .name = "value" }, .{ .name = "extra" } },
1647 }, .{ .durability = .buffered });
1648 } else {
1649 _ = try catalog.createRelation(std.testing.allocator, .{
1650 .name = name,
1651 .columns = &.{.{ .name = "value" }},
1652 }, .{ .durability = .buffered });
1653 }
1654 var handle = try catalog.openRelation(std.testing.allocator, name);
1655 defer handle.deinit();
1656 if (two_columns) {
1657 _ = try handle.relation.put(std.testing.allocator, 1, &.{
1658 .{ .text = value },
1659 .{ .integer = 7 },
1660 }, .{ .durability = .buffered });
1661 } else {
1662 _ = try handle.relation.put(std.testing.allocator, 1, &.{
1663 .{ .text = value },
1664 }, .{ .durability = .buffered });
1665 }
1666 }
1667
1668 fn findDatabaseConflict(conflicts: []const DatabaseConflict, relation_name: []const u8) ?*const DatabaseConflict {
1669 for (conflicts) |*conflict| {
1670 if (std.mem.eql(u8, conflict.relation, relation_name)) return conflict;
1671 }
1672 return null;
1673 }
1674
1675 fn expectOptionalHash(expected: version.Hash, actual: ?version.Hash) !void {
1676 try std.testing.expect(actual != null);
1677 try std.testing.expect(version.same(expected, actual.?));
1678 }
1679
1680 fn databaseValueForRelation(allocator: Allocator, relation_value: *const version.RelationValue) !version.DatabaseValue {
1681 return try databaseValueForRelations(allocator, &.{relation_value});
1682 }
1683
1684 fn databaseValueForRelations(allocator: Allocator, relation_values: []const *const version.RelationValue) !version.DatabaseValue {
1685 const relations = try allocator.alloc(version.RelationValue, relation_values.len);
1686 var relation_count: usize = 0;
1687 errdefer {
1688 for (relations[0..relation_count]) |*entry| entry.deinit(allocator);
1689 allocator.free(relations);
1690 }
1691 for (relation_values, relations) |relation_value, *target| {
1692 target.* = try relation_value.clone(allocator);
1693 relation_count += 1;
1694 }
1695 return try version.databaseValueFromOwnedRelations(allocator, relations, version.ConflictRoot.empty().hash);
1696 }
1697
1698 fn testingHeader() wal.Header {
1699 return .{
1700 .sequence = 2001,
1701 .salt = .{ .first = 0xaaaa_0101, .second = 0xbbbb_0202 },
1702 };
1703 }