lib/trace/src/store/format/database.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("sql");
3
4 pub const file_name = "events.sql";
5 pub const wal_file_name = "events.sql.wal";
6 pub const key_bytes: usize = 8;
7
8 const root_page: u32 = 2;
9 const identity_page: u32 = 3;
10 const checkpoint_target_frames: usize = 256;
11 const tree_frame_overhead: usize = 68;
12 const heap_reserve_bytes: usize = 64 * 1024;
13
14 const header = sql.wal.Header{
15 .sequence = 0x5452_4345,
16 .salt = .{ .first = 0x5452_4345, .second = 0x5351_4c31 },
17 };
18
19 pub const Capacity = struct {
20 wal_frames: usize,
21 wal_bytes: usize,
22 heap_bytes: usize,
23 };
24
25 pub fn writerCapacity(max_value_bytes: usize) error{CapacityOverflow}!Capacity {
26 const overflow_frames = divideRoundUp(max_value_bytes, sql.page.overflow_capacity);
27 const value_frames = std.math.add(usize, overflow_frames, tree_frame_overhead) catch
28 return error.CapacityOverflow;
29 const wal_frames = std.math.add(usize, checkpoint_target_frames, value_frames) catch
30 return error.CapacityOverflow;
31 return try capacity(wal_frames);
32 }
33
34 pub fn readerCapacity() error{CapacityOverflow}!Capacity {
35 return try capacity(1);
36 }
37
38 pub fn workspaceLimits(storage: Capacity) sql.FileDatabase.Workspace.Limits {
39 return .{
40 .header = header,
41 .max_wal_bytes = storage.wal_bytes,
42 .path_storage = .{
43 .database_bytes = file_name.len,
44 .wal_bytes = wal_file_name.len,
45 },
46 .read_cache_pages = 0,
47 };
48 }
49
50 pub fn openWriter(
51 allocator: std.mem.Allocator,
52 workspace: *sql.FileDatabase.Workspace,
53 root_path: []const u8,
54 storage: Capacity,
55 ) !sql.FileDatabase {
56 return try open(allocator, workspace, root_path, storage, true);
57 }
58
59 pub fn openReader(
60 allocator: std.mem.Allocator,
61 workspace: *sql.FileDatabase.Workspace,
62 root_path: []const u8,
63 storage: Capacity,
64 ) !sql.FileDatabase {
65 return try open(allocator, workspace, root_path, storage, false);
66 }
67
68 pub fn events(file: *sql.FileDatabase) !sql.Tree {
69 return try sql.Tree.open(file, .{
70 .root_page = root_page,
71 .identity_page = identity_page,
72 .reserved_page_max = identity_page,
73 });
74 }
75
76 pub fn checkpoint(file: *sql.FileDatabase) !void {
77 _ = try file.checkpoint(.{ .restart_header = header });
78 }
79
80 pub fn checkpointIfNeeded(file: *sql.FileDatabase) !void {
81 const bytes = sql.lifecycle.walSize(file);
82 std.debug.assert(bytes >= sql.wal.header_size);
83 const frames = (bytes - sql.wal.header_size) / sql.wal.frame_size;
84 if (frames >= checkpoint_target_frames) try checkpoint(file);
85 }
86
87 pub fn sequenceKey(buffer: *[key_bytes]u8, sequence: u64) []const u8 {
88 std.mem.writeInt(u64, buffer, sequence, .big);
89 return buffer;
90 }
91
92 pub fn updateChecksum(hasher: *std.hash.Wyhash, key: []const u8, value: []const u8) void {
93 std.debug.assert(key.len == key_bytes);
94 hasher.update(key);
95 hasher.update(value);
96 }
97
98 fn open(
99 allocator: std.mem.Allocator,
100 workspace: *sql.FileDatabase.Workspace,
101 root_path: []const u8,
102 storage: Capacity,
103 writable: bool,
104 ) !sql.FileDatabase {
105 var dir = try std.Io.Dir.openDirAbsolute(std.Options.debug_io, root_path, .{});
106 defer dir.close(std.Options.debug_io);
107 return try sql.FileDatabase.open(allocator, workspace, dir, .{
108 .paths = .{ .database = file_name, .wal = wal_file_name },
109 .header = header,
110 .max_wal_bytes = storage.wal_bytes,
111 .wal_capacity_bytes = storage.wal_bytes,
112 .read_cache_capacity = 0,
113 .write_capacity = if (writable)
114 .{ .wal_frames = storage.wal_frames, .wal_pages = storage.wal_frames }
115 else
116 .{},
117 });
118 }
119
120 fn capacity(wal_frames: usize) error{CapacityOverflow}!Capacity {
121 const wal_capacity = try sql.wal.Writer.Capacity.derive(.{
122 .header = header,
123 .frames = wal_frames,
124 });
125 const tripled = std.math.mul(usize, wal_capacity.storage_bytes, 3) catch return error.CapacityOverflow;
126 const heap_bytes = std.math.add(usize, tripled, heap_reserve_bytes) catch return error.CapacityOverflow;
127 return .{
128 .wal_frames = wal_frames,
129 .wal_bytes = wal_capacity.storage_bytes,
130 .heap_bytes = heap_bytes,
131 };
132 }
133
134 fn divideRoundUp(numerator: usize, denominator: usize) usize {
135 std.debug.assert(denominator != 0);
136 return numerator / denominator + @intFromBool(numerator % denominator != 0);
137 }
138
139 test "SQL trace capacity covers exact WAL and workspace storage" {
140 const max_value_bytes: usize = 4 * 1024 * 1024;
141 const derived = try writerCapacity(max_value_bytes);
142 const overflow_frames = divideRoundUp(max_value_bytes, sql.page.overflow_capacity);
143 const value_frames = overflow_frames + tree_frame_overhead;
144 try std.testing.expectEqual(checkpoint_target_frames + value_frames, derived.wal_frames);
145 const expected_wal_bytes = sql.wal.header_size + derived.wal_frames * sql.wal.frame_size;
146 try std.testing.expectEqual(expected_wal_bytes, derived.wal_bytes);
147 try std.testing.expectEqual(expected_wal_bytes * 3 + heap_reserve_bytes, derived.heap_bytes);
148
149 const writer_workspace = try sql.FileDatabase.Workspace.Capacity.derive(
150 workspaceLimits(derived),
151 );
152 try std.testing.expect(writer_workspace.requested_bytes <= derived.heap_bytes);
153
154 const reader = try readerCapacity();
155 const reader_limits = workspaceLimits(reader);
156 try std.testing.expectEqual(@as(usize, 0), reader_limits.read_cache_pages);
157 const reader_workspace = try sql.FileDatabase.Workspace.Capacity.derive(reader_limits);
158 try std.testing.expect(reader_workspace.requested_bytes <= reader.heap_bytes);
159 }
160
161 test "SQL trace sequence keys preserve order" {
162 var first_buffer: [key_bytes]u8 = undefined;
163 var second_buffer: [key_bytes]u8 = undefined;
164 const first = sequenceKey(&first_buffer, 255);
165 const second = sequenceKey(&second_buffer, 256);
166 try std.testing.expectEqual(std.math.Order.lt, std.mem.order(u8, first, second));
167 }
168
169 test "SQL trace header owns the trace format identity" {
170 try std.testing.expectEqual(@as(u32, 0x5452_4345), header.sequence);
171 try std.testing.expectEqual(@as(u32, 0x5452_4345), header.salt.first);
172 try std.testing.expectEqual(@as(u32, 0x5351_4c31), header.salt.second);
173 }