lib/sql/src/diff.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const catalog_mod = @import("catalog.zig");
3 const key = @import("key.zig");
4 const relation_mod = @import("relation.zig");
5 const row = @import("row.zig");
6 const tree = @import("tree.zig");
7 const version = @import("version.zig");
8 const wal = @import("wal.zig");
9
10 const Allocator = std.mem.Allocator;
11
12 pub const Error = Allocator.Error || relation_mod.Error || key.Error || row.Error;
13
14 pub const ChangeKind = enum {
15 added,
16 removed,
17 modified,
18 schema,
19 };
20
21 pub const Change = struct {
22 kind: ChangeKind,
23 rowid: i64,
24 from: ?[]u8 = null,
25 to: ?[]u8 = null,
26
27 fn deinit(self: *Change, allocator: Allocator) void {
28 if (self.from) |bytes| allocator.free(bytes);
29 if (self.to) |bytes| allocator.free(bytes);
30 self.* = undefined;
31 }
32 };
33
34 pub const RelationSnapshot = struct {
35 root: *const version.RelationRoot,
36 rows: Rows,
37 };
38
39 pub const Row = struct {
40 rowid: i64,
41 bytes: []const u8,
42 };
43
44 pub const Rows = union(enum) {
45 live: *const catalog_mod.RelationHandle,
46 materialized: []const version.RelationRow,
47
48 /// Starts a scan of the rows from `start` up to `end` in `target`.
49 pub fn scan(
50 self: Rows,
51 target: *Scan,
52 allocator: Allocator,
53 start: ?i64,
54 end: ?i64,
55 ) Error!void {
56 switch (self) {
57 .live => |handle| {
58 target.* = .{ .live = undefined };
59 try handle.relation.scan(&target.live, allocator, start, end);
60 },
61 .materialized => |rows| target.* = .{
62 .materialized = MaterializedScan.init(rows, start, end),
63 },
64 }
65 }
66 };
67
68 pub const Scan = union(enum) {
69 live: relation_mod.Scan,
70 materialized: MaterializedScan,
71
72 pub fn deinit(self: *Scan) void {
73 switch (self.*) {
74 .live => |*scan| scan.deinit(),
75 .materialized => {},
76 }
77 self.* = undefined;
78 }
79
80 pub fn next(self: *Scan) Error!?Row {
81 return switch (self.*) {
82 .live => |*scan| if (try scan.next()) |entry| .{
83 .rowid = entry.rowid,
84 .bytes = entry.bytes,
85 } else null,
86 .materialized => |*scan| scan.next(),
87 };
88 }
89 };
90
91 const MaterializedScan = struct {
92 rows: []const version.RelationRow,
93 index: usize,
94 start: ?i64,
95 end: ?i64,
96
97 fn init(rows: []const version.RelationRow, start: ?i64, end: ?i64) MaterializedScan {
98 var index: usize = 0;
99 while (index < rows.len and beforeStart(rows[index].rowid, start)) : (index += 1) {}
100 return .{
101 .rows = rows,
102 .index = index,
103 .start = start,
104 .end = end,
105 };
106 }
107
108 fn next(self: *MaterializedScan) ?Row {
109 while (self.index < self.rows.len) : (self.index += 1) {
110 const row_value = self.rows[self.index];
111 if (beforeStart(row_value.rowid, self.start)) continue;
112 if (!beforeEnd(row_value.rowid, self.end)) return null;
113 self.index += 1;
114 return .{
115 .rowid = row_value.rowid,
116 .bytes = row_value.bytes,
117 };
118 }
119 return null;
120 }
121 };
122
123 pub const RelationDiff = struct {
124 allocator: Allocator,
125 schema_changed: bool,
126 skipped_ranges: usize = 0,
127 changes: []Change,
128
129 pub fn deinit(self: *RelationDiff) void {
130 for (self.changes) |*change| change.deinit(self.allocator);
131 if (self.changes.len != 0) self.allocator.free(self.changes);
132 self.* = undefined;
133 }
134 };
135
136 pub fn relation(allocator: Allocator, left: RelationSnapshot, right: RelationSnapshot) Error!RelationDiff {
137 const schema_changed = !version.same(left.root.schema, right.root.schema);
138 if (!schema_changed and version.same(left.root.hash, right.root.hash)) {
139 return .{
140 .allocator = allocator,
141 .schema_changed = false,
142 .skipped_ranges = 1,
143 .changes = &.{},
144 };
145 }
146
147 const skipped = if (schema_changed) &.{} else try equalTableRanges(allocator, &left.root.table, &right.root.table);
148 defer if (skipped.len != 0) allocator.free(skipped);
149
150 var changes: std.ArrayList(Change) = .empty;
151 errdefer {
152 for (changes.items) |*change| change.deinit(allocator);
153 changes.deinit(allocator);
154 }
155
156 var start: ?i64 = null;
157 for (skipped) |skip| {
158 if (start != null or skip.start != null) try appendRangeDiff(allocator, &changes, left, right, schema_changed, start, skip.start);
159 start = skip.end;
160 if (start == null) break;
161 }
162 if (start != null or skipped.len == 0) try appendRangeDiff(allocator, &changes, left, right, schema_changed, start, null);
163
164 return .{
165 .allocator = allocator,
166 .schema_changed = schema_changed,
167 .skipped_ranges = skipped.len,
168 .changes = try changes.toOwnedSlice(allocator),
169 };
170 }
171
172 const RowRange = struct {
173 start: ?i64,
174 end: ?i64,
175 };
176
177 fn appendRangeDiff(allocator: Allocator, changes: *std.ArrayList(Change), left: RelationSnapshot, right: RelationSnapshot, schema_changed: bool, start: ?i64, end: ?i64) Error!void {
178 if (!rangeCanContainRows(start, end)) return;
179
180 var left_scan: Scan = undefined;
181 try left.rows.scan(&left_scan, allocator, start, end);
182 defer left_scan.deinit();
183 var right_scan: Scan = undefined;
184 try right.rows.scan(&right_scan, allocator, start, end);
185 defer right_scan.deinit();
186
187 var left_entry = try left_scan.next();
188 var right_entry = try right_scan.next();
189
190 while (left_entry != null or right_entry != null) {
191 if (left_entry == null) {
192 try appendAdded(allocator, changes, right_entry.?);
193 right_entry = try right_scan.next();
194 continue;
195 }
196 if (right_entry == null) {
197 try appendRemoved(allocator, changes, left_entry.?);
198 left_entry = try left_scan.next();
199 continue;
200 }
201
202 const l = left_entry.?;
203 const r = right_entry.?;
204 if (l.rowid < r.rowid) {
205 try appendRemoved(allocator, changes, l);
206 left_entry = try left_scan.next();
207 continue;
208 }
209 if (l.rowid > r.rowid) {
210 try appendAdded(allocator, changes, r);
211 right_entry = try right_scan.next();
212 continue;
213 }
214
215 if (!std.mem.eql(u8, l.bytes, r.bytes)) {
216 try appendChanged(allocator, changes, .modified, l.rowid, l.bytes, r.bytes);
217 } else if (schema_changed) {
218 try appendChanged(allocator, changes, .schema, l.rowid, l.bytes, r.bytes);
219 }
220 left_entry = try left_scan.next();
221 right_entry = try right_scan.next();
222 }
223 }
224
225 fn equalTableRanges(allocator: Allocator, left: *const version.MapRoot, right: *const version.MapRoot) Error![]RowRange {
226 var ranges: std.ArrayList(RowRange) = .empty;
227 errdefer ranges.deinit(allocator);
228 if (left.nodes.len != 0 and right.nodes.len != 0) {
229 try appendEqualRanges(allocator, &ranges, left, left.rootNode(), right, right.rootNode());
230 }
231 return try ranges.toOwnedSlice(allocator);
232 }
233
234 fn appendEqualRanges(allocator: Allocator, ranges: *std.ArrayList(RowRange), left: *const version.MapRoot, left_node: *const tree.Node, right: *const version.MapRoot, right_node: *const tree.Node) Error!void {
235 if (sameSubtree(left_node, right_node)) {
236 const range = RowRange{
237 .start = try decodeLower(left_node.lower),
238 .end = try decodeUpper(left_node.upper),
239 };
240 if (rangeCanContainRows(range.start, range.end)) try ranges.append(allocator, range);
241 return;
242 }
243
244 if (left_node.kind != .branch or right_node.kind != .branch) return;
245 for (left.childIndexes(left_node)) |left_child_index| {
246 const left_child = &left.nodes[left_child_index];
247 const right_child = matchingChild(right, right_node, left_child) orelse continue;
248 try appendEqualRanges(allocator, ranges, left, left_child, right, right_child);
249 }
250 }
251
252 fn matchingChild(root: *const version.MapRoot, parent: *const tree.Node, target: *const tree.Node) ?*const tree.Node {
253 for (root.childIndexes(parent)) |child_index| {
254 const child = &root.nodes[child_index];
255 if (sameRange(child, target)) return child;
256 }
257 return null;
258 }
259
260 fn sameSubtree(left: *const tree.Node, right: *const tree.Node) bool {
261 return left.kind == right.kind and sameRange(left, right) and std.mem.eql(u8, left.hash[0..], right.hash[0..]);
262 }
263
264 fn sameRange(left: *const tree.Node, right: *const tree.Node) bool {
265 return std.mem.eql(u8, left.lower, right.lower) and sameUpper(left.upper, right.upper);
266 }
267
268 fn sameUpper(left: ?[]const u8, right: ?[]const u8) bool {
269 if (left) |left_bytes| {
270 const right_bytes = right orelse return false;
271 return std.mem.eql(u8, left_bytes, right_bytes);
272 }
273 return right == null;
274 }
275
276 fn decodeLower(bytes: []const u8) Error!?i64 {
277 if (bytes.len == 0) return null;
278 return try key.decodeRowId(bytes);
279 }
280
281 fn decodeUpper(bytes: ?[]const u8) Error!?i64 {
282 const bound = bytes orelse return null;
283 return try decodeLower(bound);
284 }
285
286 fn rangeCanContainRows(start: ?i64, end: ?i64) bool {
287 if (start) |lower| {
288 if (end) |upper| return lower < upper;
289 }
290 return true;
291 }
292
293 fn beforeStart(rowid: i64, start: ?i64) bool {
294 const lower = start orelse return false;
295 return rowid < lower;
296 }
297
298 fn beforeEnd(rowid: i64, end: ?i64) bool {
299 const upper = end orelse return true;
300 return rowid < upper;
301 }
302
303 fn appendAdded(allocator: Allocator, changes: *std.ArrayList(Change), entry: Row) Error!void {
304 const to = try allocator.dupe(u8, entry.bytes);
305 errdefer allocator.free(to);
306 try changes.append(allocator, .{
307 .kind = .added,
308 .rowid = entry.rowid,
309 .to = to,
310 });
311 }
312
313 fn appendRemoved(allocator: Allocator, changes: *std.ArrayList(Change), entry: Row) Error!void {
314 const from = try allocator.dupe(u8, entry.bytes);
315 errdefer allocator.free(from);
316 try changes.append(allocator, .{
317 .kind = .removed,
318 .rowid = entry.rowid,
319 .from = from,
320 });
321 }
322
323 fn appendChanged(allocator: Allocator, changes: *std.ArrayList(Change), kind: ChangeKind, rowid: i64, from_bytes: []const u8, to_bytes: []const u8) Error!void {
324 const from = try allocator.dupe(u8, from_bytes);
325 errdefer allocator.free(from);
326 const to = try allocator.dupe(u8, to_bytes);
327 errdefer allocator.free(to);
328 try changes.append(allocator, .{
329 .kind = kind,
330 .rowid = rowid,
331 .from = from,
332 .to = to,
333 });
334 }
335
336 test "relation diff reports added removed and modified rows" {
337 var tmp = std.testing.tmpDir(.{});
338 defer tmp.cleanup();
339
340 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
341 .paths = .{ .database = "diff.db", .wal = "diff.wal" },
342 .header = testingHeader(),
343 });
344 defer database.deinit();
345 try database.reserve(.{ .wal_frames = 512 });
346
347 var catalog = try catalog_mod.Catalog.open(&database, .{});
348 _ = try catalog.createRelation(std.testing.allocator, .{
349 .name = "left",
350 .columns = &.{.{ .name = "value" }},
351 }, .{ .durability = .buffered });
352 _ = try catalog.createRelation(std.testing.allocator, .{
353 .name = "right",
354 .columns = &.{.{ .name = "value" }},
355 }, .{ .durability = .buffered });
356
357 var left = try catalog.openRelation(std.testing.allocator, "left");
358 defer left.deinit();
359 var right = try catalog.openRelation(std.testing.allocator, "right");
360 defer right.deinit();
361 const schema = try catalog.schemaState(std.testing.allocator);
362
363 _ = try left.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
364 _ = try left.relation.put(std.testing.allocator, 2, &.{.{ .text = "old" }}, .{ .durability = .buffered });
365 _ = try left.relation.put(std.testing.allocator, 3, &.{.{ .text = "removed" }}, .{ .durability = .buffered });
366 _ = try right.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
367 _ = try right.relation.put(std.testing.allocator, 2, &.{.{ .text = "new" }}, .{ .durability = .buffered });
368 _ = try right.relation.put(std.testing.allocator, 4, &.{.{ .text = "added" }}, .{ .durability = .buffered });
369
370 var left_root = try version.relationRoot(std.testing.allocator, "left", schema, &left, null);
371 defer left_root.deinit();
372 var right_root = try version.relationRoot(std.testing.allocator, "right", schema, &right, null);
373 defer right_root.deinit();
374
375 var result = try relation(std.testing.allocator, .{
376 .root = &left_root,
377 .rows = .{ .live = &left },
378 }, .{
379 .root = &right_root,
380 .rows = .{ .live = &right },
381 });
382 defer result.deinit();
383
384 try std.testing.expect(!result.schema_changed);
385 try std.testing.expectEqual(@as(usize, 3), result.changes.len);
386 try std.testing.expectEqual(ChangeKind.modified, result.changes[0].kind);
387 try std.testing.expectEqual(@as(i64, 2), result.changes[0].rowid);
388 try std.testing.expectEqual(ChangeKind.removed, result.changes[1].kind);
389 try std.testing.expectEqual(@as(i64, 3), result.changes[1].rowid);
390 try std.testing.expectEqual(ChangeKind.added, result.changes[2].kind);
391 try std.testing.expectEqual(@as(i64, 4), result.changes[2].rowid);
392 }
393
394 test "relation diff marks rows when schema roots differ" {
395 var tmp = std.testing.tmpDir(.{});
396 defer tmp.cleanup();
397
398 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
399 .paths = .{ .database = "schema-diff.db", .wal = "schema-diff.wal" },
400 .header = testingHeader(),
401 });
402 defer database.deinit();
403 try database.reserve(.{ .wal_frames = 512 });
404
405 var catalog = try catalog_mod.Catalog.open(&database, .{});
406 _ = try catalog.createRelation(std.testing.allocator, .{
407 .name = "left",
408 .columns = &.{.{ .name = "value" }},
409 }, .{ .durability = .buffered });
410 _ = try catalog.createRelation(std.testing.allocator, .{
411 .name = "right",
412 .columns = &.{.{ .name = "value", .column = .{ .collation = .nocase } }},
413 }, .{ .durability = .buffered });
414
415 var left = try catalog.openRelation(std.testing.allocator, "left");
416 defer left.deinit();
417 var right = try catalog.openRelation(std.testing.allocator, "right");
418 defer right.deinit();
419 const schema = try catalog.schemaState(std.testing.allocator);
420
421 _ = try left.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
422 _ = try right.relation.put(std.testing.allocator, 1, &.{.{ .text = "same" }}, .{ .durability = .buffered });
423
424 var left_root = try version.relationRoot(std.testing.allocator, "left", schema, &left, null);
425 defer left_root.deinit();
426 var right_root = try version.relationRoot(std.testing.allocator, "right", schema, &right, null);
427 defer right_root.deinit();
428
429 var result = try relation(std.testing.allocator, .{
430 .root = &left_root,
431 .rows = .{ .live = &left },
432 }, .{
433 .root = &right_root,
434 .rows = .{ .live = &right },
435 });
436 defer result.deinit();
437
438 try std.testing.expect(result.schema_changed);
439 try std.testing.expectEqual(@as(usize, 1), result.changes.len);
440 try std.testing.expectEqual(ChangeKind.schema, result.changes[0].kind);
441 try std.testing.expectEqual(@as(i64, 1), result.changes[0].rowid);
442 }
443
444 test "relation diff skips equal table subtrees" {
445 var tmp = std.testing.tmpDir(.{});
446 defer tmp.cleanup();
447
448 var database = try @import("file.zig").Database.openForTesting(std.testing.allocator, tmp.dir, .{
449 .paths = .{ .database = "subtree-diff.db", .wal = "subtree-diff.wal" },
450 .header = testingHeader(),
451 });
452 defer database.deinit();
453 try database.reserve(.{ .wal_frames = 768 });
454
455 var catalog = try catalog_mod.Catalog.open(&database, .{});
456 _ = try catalog.createRelation(std.testing.allocator, .{
457 .name = "left",
458 .columns = &.{.{ .name = "value" }},
459 }, .{ .durability = .buffered });
460 _ = try catalog.createRelation(std.testing.allocator, .{
461 .name = "right",
462 .columns = &.{.{ .name = "value" }},
463 }, .{ .durability = .buffered });
464
465 var left = try catalog.openRelation(std.testing.allocator, "left");
466 defer left.deinit();
467 var right = try catalog.openRelation(std.testing.allocator, "right");
468 defer right.deinit();
469
470 var rowid: i64 = 0;
471 while (rowid < 260) : (rowid += 1) {
472 var value_buffer: [16]u8 = undefined;
473 const value = try std.fmt.bufPrint(&value_buffer, "v{d:0>8}", .{rowid});
474 _ = try left.relation.put(std.testing.allocator, rowid, &.{.{ .text = value }}, .{ .durability = .buffered });
475 _ = try right.relation.put(std.testing.allocator, rowid, &.{.{ .text = value }}, .{ .durability = .buffered });
476 }
477 _ = try right.relation.put(std.testing.allocator, 259, &.{.{ .text = "changed" }}, .{ .durability = .buffered });
478
479 const schema = try catalog.schemaState(std.testing.allocator);
480 var left_root = try version.relationRoot(std.testing.allocator, "left", schema, &left, null);
481 defer left_root.deinit();
482 var right_root = try version.relationRoot(std.testing.allocator, "right", schema, &right, null);
483 defer right_root.deinit();
484
485 var result = try relation(std.testing.allocator, .{
486 .root = &left_root,
487 .rows = .{ .live = &left },
488 }, .{
489 .root = &right_root,
490 .rows = .{ .live = &right },
491 });
492 defer result.deinit();
493
494 try std.testing.expect(!result.schema_changed);
495 try std.testing.expect(result.skipped_ranges > 0);
496 try std.testing.expectEqual(@as(usize, 1), result.changes.len);
497 try std.testing.expectEqual(ChangeKind.modified, result.changes[0].kind);
498 try std.testing.expectEqual(@as(i64, 259), result.changes[0].rowid);
499 }
500
501 test "relation diff recursively skips equal descendant table subtrees" {
502 const shared = testHash(7);
503 const left_specs = [_]TestNode{
504 .{ .kind = .branch, .lower = null, .upper = null, .depth = 0, .hash = testHash(1), .children_start = 0, .children_len = 2 },
505 .{ .kind = .branch, .lower = null, .upper = 100, .depth = 1, .hash = testHash(2), .children_start = 2, .children_len = 2 },
506 .{ .kind = .leaf, .lower = 100, .upper = null, .depth = 1, .hash = testHash(3) },
507 .{ .kind = .leaf, .lower = null, .upper = 50, .depth = 2, .hash = shared },
508 .{ .kind = .leaf, .lower = 50, .upper = 100, .depth = 2, .hash = testHash(4) },
509 };
510 const right_specs = [_]TestNode{
511 .{ .kind = .branch, .lower = null, .upper = null, .depth = 0, .hash = testHash(11), .children_start = 0, .children_len = 2 },
512 .{ .kind = .branch, .lower = null, .upper = 100, .depth = 1, .hash = testHash(12), .children_start = 2, .children_len = 2 },
513 .{ .kind = .leaf, .lower = 100, .upper = null, .depth = 1, .hash = testHash(13) },
514 .{ .kind = .leaf, .lower = null, .upper = 50, .depth = 2, .hash = shared },
515 .{ .kind = .leaf, .lower = 50, .upper = 100, .depth = 2, .hash = testHash(14) },
516 };
517 const edges = [_]usize{ 1, 2, 3, 4 };
518 var left = try syntheticRoot(std.testing.allocator, &left_specs, &edges);
519 defer left.deinit();
520 var right = try syntheticRoot(std.testing.allocator, &right_specs, &edges);
521 defer right.deinit();
522
523 const ranges = try equalTableRanges(std.testing.allocator, &left, &right);
524 defer std.testing.allocator.free(ranges);
525
526 try std.testing.expectEqual(@as(usize, 1), ranges.len);
527 try std.testing.expect(ranges[0].start == null);
528 try std.testing.expectEqual(@as(i64, 50), ranges[0].end.?);
529 }
530
531 fn testingHeader() wal.Header {
532 return .{
533 .sequence = 1901,
534 .salt = .{ .first = 0xfeed_0101, .second = 0xbeef_0202 },
535 };
536 }
537
538 const TestNode = struct {
539 kind: tree.NodeKind,
540 lower: ?i64,
541 upper: ?i64,
542 depth: usize,
543 hash: tree.Hash,
544 children_start: usize = 0,
545 children_len: usize = 0,
546 };
547
548 fn syntheticRoot(allocator: Allocator, specs: []const TestNode, edge_specs: []const usize) Error!version.MapRoot {
549 const nodes = try allocator.alloc(tree.Node, specs.len);
550 var node_count: usize = 0;
551 errdefer {
552 for (nodes[0..node_count]) |node| {
553 allocator.free(node.lower);
554 if (node.upper) |upper| allocator.free(upper);
555 }
556 allocator.free(nodes);
557 }
558
559 for (specs) |spec| {
560 const lower = try rowLower(allocator, spec.lower);
561 var lower_assigned = false;
562 errdefer if (!lower_assigned) allocator.free(lower);
563 const upper = try rowUpper(allocator, spec.upper);
564 var upper_assigned = upper == null;
565 errdefer if (!upper_assigned) allocator.free(upper.?);
566 nodes[node_count] = .{
567 .kind = spec.kind,
568 .lower = lower,
569 .upper = upper,
570 .depth = spec.depth,
571 .summary = .{},
572 .hash = spec.hash,
573 .children_start = spec.children_start,
574 .children_len = spec.children_len,
575 };
576 lower_assigned = true;
577 upper_assigned = true;
578 node_count += 1;
579 }
580
581 const edges = try allocator.dupe(usize, edge_specs);
582 errdefer allocator.free(edges);
583
584 return .{
585 .allocator = allocator,
586 .summary = .{},
587 .hash = specs[0].hash,
588 .subtree = specs[0].hash,
589 .nodes = nodes,
590 .edges = edges,
591 };
592 }
593
594 fn rowLower(allocator: Allocator, value: ?i64) Error![]u8 {
595 const rowid = value orelse return try allocator.dupe(u8, &.{});
596 const bytes = try allocator.alloc(u8, key.rowid_size);
597 errdefer allocator.free(bytes);
598 _ = try key.encodeRowId(bytes, rowid);
599 return bytes;
600 }
601
602 fn rowUpper(allocator: Allocator, value: ?i64) Error!?[]u8 {
603 const rowid = value orelse return null;
604 const bytes = try allocator.alloc(u8, key.rowid_size);
605 errdefer allocator.free(bytes);
606 _ = try key.encodeRowId(bytes, rowid);
607 return bytes;
608 }
609
610 fn testHash(seed: u8) tree.Hash {
611 var hash: tree.Hash = undefined;
612 @memset(hash[0..], seed);
613 return hash;
614 }