lib/sql/src/history/pack.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const conflict_mod = @import("conflict.zig");
4 const record_mod = @import("record.zig");
5 const store_mod = @import("store.zig");
6 const materialize_mod = @import("materialize.zig");
7 const recover_mod = @import("recover.zig");
8 const chunk_mod = sql.chunk;
9 const row = sql.row;
10 const tree = sql.tree;
11 const version = sql.version;
12
13 const Allocator = std.mem.Allocator;
14 const Error = store_mod.Error;
15 const History = store_mod.History;
16
17 pub fn appendRowChunkPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), digest: version.Hash) Error!void {
18 const record = self.findRowChunk(digest) orelse return error.InvalidHistory;
19 const file = self.file orelse return error.InvalidHistory;
20 try target.ensureUnusedCapacity(allocator, record.payload.len);
21 const start = target.items.len;
22 target.items.len = start + record.payload.len;
23 const slice = target.items[start..][0..record.payload.len];
24 if (try file.readPositionalAll(self.io, slice, record.payload.offset) != record.payload.len) return error.InvalidHistory;
25 if (!version.same(record.payload.expected, record_mod.recordHash(@backingInt(record_mod.RecordKind.row_chunk), slice))) return error.InvalidHistory;
26 }
27
28 pub fn appendTreeNodePackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), key: version.Hash) Error!void {
29 const index = self.lookup.tree_nodes.get(key) orelse return error.InvalidHistory;
30 var node = try materialize_mod.readTreeNodeRecord(self, allocator, &self.tree_nodes.items[index]);
31 defer node.deinit(allocator);
32 try record_mod.appendU32(allocator, target, 1);
33 try record_mod.appendHash(allocator, target, key);
34 try record_mod.appendTreeNodeContent(allocator, target, &node.node, node.children);
35 }
36
37 pub fn appendChunkIndexPagePackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), digest: version.Hash) Error!void {
38 var chunks = (try self.indexPageChunks(allocator, digest)) orelse return error.InvalidHistory;
39 defer chunks.deinit();
40 if (chunks.items.len > std.math.maxInt(u32)) return error.InvalidHistory;
41 try record_mod.appendHash(allocator, target, digest);
42 try record_mod.appendU32(allocator, target, @intCast(chunks.items.len));
43 for (chunks.items) |chunk_digest| try record_mod.appendHash(allocator, target, chunk_digest);
44 }
45
46 pub fn appendRelationRowsPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), root: version.Hash) Error!void {
47 var pages = (try self.relationRowsPages(allocator, root)) orelse return error.InvalidHistory;
48 defer pages.deinit();
49 if (pages.items.len > std.math.maxInt(u32)) return error.InvalidHistory;
50 try record_mod.appendHash(allocator, target, root);
51 try record_mod.appendU32(allocator, target, @intCast(pages.items.len));
52 for (pages.items) |digest| try record_mod.appendHash(allocator, target, digest);
53 }
54
55 pub fn appendRelationRootPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), hash: version.Hash) Error!void {
56 const record = self.findRelationRoot(hash) orelse return error.InvalidHistory;
57 switch (record.storage) {
58 .materialized => |index| {
59 const materialized = &self.materialized_relation_roots.items[index];
60 try record_mod.appendRelationRootMerkle(allocator, target, materialized.root, materialized.table_key, materialized.index_keys);
61 },
62 .indexed => |location| {
63 const payload = try materialize_mod.readPayload(self, allocator, .relation_root, location);
64 defer allocator.free(payload);
65 try target.appendSlice(allocator, payload);
66 },
67 }
68 }
69
70 pub fn appendDatabaseRootPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), hash: version.Hash) Error!void {
71 const record = self.findDatabaseRoot(hash) orelse return error.InvalidHistory;
72 try record_mod.appendDatabaseRootValue(allocator, target, record.root);
73 }
74
75 pub fn appendConflictPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), hash: version.Hash) Error!void {
76 const record = self.findConflict(hash) orelse return error.InvalidHistory;
77 try conflict_mod.appendConflictArtifactRecordPayload(allocator, target, record.artifact);
78 }
79
80 pub fn appendConflictRootPackPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), hash: version.Hash) Error!void {
81 try self.validateConflictRoot(hash);
82 const record = self.findConflictRoot(hash) orelse return error.InvalidHistory;
83 try conflict_mod.appendConflictEntries(allocator, target, record.entries);
84 }
85
86 pub fn importRowChunkPayload(self: *History, payload: []const u8) Error!bool {
87 var reader = record_mod.PayloadReader.init(payload);
88 const digest = try reader.hash();
89 const rows = try record_mod.readRelationRows(self.allocator, &reader);
90 defer version.freeRelationRows(self.allocator, rows);
91 try reader.finish();
92 if (!version.same(digest, chunk_mod.digest(rows))) return error.InvalidHistory;
93 if (self.findRowChunk(digest) != null) return false;
94 const offset = self.bytes_written + record_mod.record_header_size;
95 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.row_chunk), payload);
96 try self.appendRecord(.row_chunk, payload);
97 try recover_mod.indexRowChunk(self, digest, expected, offset, payload.len);
98 return true;
99 }
100
101 pub fn importTreeNodePayload(self: *History, payload: []const u8) Error!bool {
102 var reader = record_mod.PayloadReader.init(payload);
103 if (try reader.readU32() != 1) return error.InvalidHistory;
104 const key = try reader.hash();
105 const content = payload[reader.cursor..];
106 if (!version.same(key, record_mod.treeNodeKey(content))) return error.InvalidHistory;
107 _ = try record_mod.nodeKind(try reader.readU8());
108 _ = try reader.readBytes();
109 _ = try reader.optionalBytes();
110 _ = try record_mod.readUsize(&reader);
111 _ = try record_mod.readSummary(&reader);
112 _ = try reader.hash();
113 const child_count = try reader.readU32();
114 var child_index: u32 = 0;
115 while (child_index < child_count) : (child_index += 1) {
116 const child = try reader.hash();
117 if (self.findTreeNode(child) == null) return error.InvalidHistory;
118 }
119 try reader.finish();
120 if (self.findTreeNode(key) != null) return false;
121 const offset = self.bytes_written + record_mod.record_header_size;
122 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.tree_nodes), payload);
123 try self.appendRecord(.tree_nodes, payload);
124 try recover_mod.indexTreeNodesPayload(self, expected, offset, payload.len, payload);
125 return true;
126 }
127
128 pub fn importChunkIndexPagePayload(self: *History, payload: []const u8) Error!bool {
129 var reader = record_mod.PayloadReader.init(payload);
130 const digest = try reader.hash();
131 const chunk_count = try reader.readU32();
132 const chunks = try self.allocator.alloc(version.Hash, chunk_count);
133 defer self.allocator.free(chunks);
134 for (chunks) |*chunk_digest| chunk_digest.* = try reader.hash();
135 try reader.finish();
136 if (!version.same(digest, chunk_mod.pageDigest(chunks))) return error.InvalidHistory;
137 for (chunks) |chunk_digest| {
138 if (self.findRowChunk(chunk_digest) == null) return error.InvalidHistory;
139 }
140 if (self.hasIndexPage(digest)) return false;
141 const offset = self.bytes_written + record_mod.record_header_size;
142 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.chunk_index_page), payload);
143 try self.appendRecord(.chunk_index_page, payload);
144 try recover_mod.indexChunkIndexPagePayload(self, expected, offset, payload.len, payload);
145 return true;
146 }
147
148 pub fn importRelationRowsPayload(self: *History, payload: []const u8) Error!bool {
149 var reader = record_mod.PayloadReader.init(payload);
150 const root = try reader.hash();
151 const page_count = try reader.readU32();
152 var page_index: u32 = 0;
153 while (page_index < page_count) : (page_index += 1) {
154 const digest = try reader.hash();
155 if (!self.hasIndexPage(digest)) return error.InvalidHistory;
156 }
157 try reader.finish();
158 if (self.hasRelationRows(root)) return false;
159 const offset = self.bytes_written + record_mod.record_header_size;
160 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.relation_rows), payload);
161 try self.appendRecord(.relation_rows, payload);
162 try recover_mod.indexRelationRowsPayload(self, expected, offset, payload.len, payload);
163 return true;
164 }
165
166 pub fn importRelationRootPayload(self: *History, payload: []const u8) Error!bool {
167 var reader = record_mod.PayloadReader.init(payload);
168 var decoded = try materialize_mod.readRelationRootShallow(self.allocator, &reader);
169 defer decoded.root.deinit();
170 defer self.allocator.free(decoded.index_keys);
171 try reader.finish();
172 if (decoded.table_key) |key| {
173 if (self.findTreeNode(key) == null) return error.InvalidHistory;
174 }
175 for (decoded.index_keys) |index_key| {
176 const key = index_key orelse continue;
177 if (self.findTreeNode(key) == null) return error.InvalidHistory;
178 }
179 if (self.findRelationRoot(decoded.root.hash) != null) return false;
180 try self.relation_roots.ensureUnusedCapacity(self.allocator, 1);
181 try self.lookup.relation_roots.ensureUnusedCapacity(self.allocator, 1);
182 const offset = self.bytes_written + record_mod.record_header_size;
183 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.relation_root), payload);
184 try self.appendRecord(.relation_root, payload);
185 recover_mod.indexRelationRootAssumeCapacity(self, decoded.root.hash, .{
186 .expected = expected,
187 .offset = offset,
188 .len = payload.len,
189 });
190 return true;
191 }
192
193 pub fn importDatabaseRootPayload(self: *History, payload: []const u8) Error!bool {
194 var reader = record_mod.PayloadReader.init(payload);
195 const conflicts = try reader.hash();
196 const entry_count = try reader.readU32();
197 const entries = try self.allocator.alloc(version.RelationEntry, entry_count);
198 defer self.allocator.free(entries);
199 for (entries) |*entry| {
200 const name = try reader.readBytes();
201 const hash = try reader.hash();
202 entry.* = .{
203 .name = name,
204 .hash = hash,
205 };
206 }
207 try reader.finish();
208 for (entries) |entry| {
209 if (self.findRelationRoot(entry.hash) == null) return error.InvalidHistory;
210 if (!self.hasRelationRows(entry.hash)) return error.InvalidHistory;
211 }
212 if (!version.same(conflicts, version.ConflictRoot.empty().hash) and self.findConflictRoot(conflicts) == null) return error.InvalidHistory;
213 var root = try version.DatabaseRoot.initSorted(self.allocator, entries, .{ .hash = conflicts });
214 const hash = root.hash;
215 root.deinit();
216 if (self.findDatabaseRoot(hash) != null) return false;
217 try self.appendRecord(.database_root, payload);
218 try recover_mod.applyDatabaseRootPayload(self, payload);
219 return true;
220 }
221
222 pub fn importConflictPayload(self: *History, payload: []const u8) Error!bool {
223 var reader = record_mod.PayloadReader.init(payload);
224 const decoded = try conflict_mod.decodeConflictArtifactPayload(&reader);
225 try reader.finish();
226 if (self.findConflict(decoded.hash) != null) return false;
227 try self.appendRecord(.conflict, payload);
228 try recover_mod.applyConflictPayload(self, payload);
229 return true;
230 }
231
232 pub fn importConflictRootPayload(self: *History, payload: []const u8) Error!bool {
233 var reader = record_mod.PayloadReader.init(payload);
234 const entries = try conflict_mod.readConflictEntries(self.allocator, &reader);
235 defer conflict_mod.deinitConflictEntries(self.allocator, entries);
236 try reader.finish();
237 self.validateConflictEntries(entries) catch return error.InvalidHistory;
238 const root = version.ConflictRoot.init(entries);
239 if (self.findConflictRoot(root.hash) != null) {
240 self.validateConflictRoot(root.hash) catch return error.InvalidHistory;
241 return false;
242 }
243 try self.appendRecord(.conflict_root, payload);
244 try recover_mod.applyConflictRootPayload(self, payload);
245 return true;
246 }