lib/sql/src/chunk.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const version = @import("version.zig");
3
4 const Allocator = std.mem.Allocator;
5
6 pub const min_rows: usize = 8;
7 pub const max_rows: usize = 128;
8 pub const max_chunk_bytes: usize = 64 * 1024;
9 const boundary_mask: u64 = 31;
10 const boundary_seed: u64 = 0x74696e795f73716c;
11
12 pub const Bounds = struct {
13 start: usize,
14 end: usize,
15 };
16
17 pub fn boundaries(allocator: Allocator, rows: []const version.RelationRow) Allocator.Error![]Bounds {
18 var list: std.ArrayList(Bounds) = .empty;
19 defer list.deinit(allocator);
20 var start: usize = 0;
21 var bytes: usize = 0;
22 for (rows, 0..) |row_value, index| {
23 bytes += row_value.bytes.len;
24 if (!shouldSplit(index + 1 - start, bytes, row_value)) continue;
25 try list.append(allocator, .{ .start = start, .end = index + 1 });
26 start = index + 1;
27 bytes = 0;
28 }
29 if (start < rows.len) try list.append(allocator, .{ .start = start, .end = rows.len });
30 return try list.toOwnedSlice(allocator);
31 }
32
33 fn shouldSplit(length: usize, bytes: usize, appended: version.RelationRow) bool {
34 return length >= max_rows or bytes >= max_chunk_bytes or (length >= min_rows and isBoundary(appended));
35 }
36
37 pub const Chunker = struct {
38 rowids: [max_rows]i64 = undefined,
39 offsets: [max_rows]usize = undefined,
40 lengths: [max_rows]usize = undefined,
41 row_count: usize = 0,
42 bytes: std.ArrayList(u8) = .empty,
43
44 pub fn deinit(self: *Chunker, allocator: Allocator) void {
45 self.bytes.deinit(allocator);
46 self.* = undefined;
47 }
48
49 pub fn append(self: *Chunker, allocator: Allocator, rowid: i64, row_bytes: []const u8) Allocator.Error!bool {
50 std.debug.assert(self.row_count < max_rows);
51 const offset = self.bytes.items.len;
52 try self.bytes.appendSlice(allocator, row_bytes);
53 self.rowids[self.row_count] = rowid;
54 self.offsets[self.row_count] = offset;
55 self.lengths[self.row_count] = row_bytes.len;
56 self.row_count += 1;
57 const appended = version.RelationRow{
58 .rowid = rowid,
59 .bytes = self.bytes.items[offset..][0..row_bytes.len],
60 };
61 return shouldSplit(self.row_count, self.bytes.items.len, appended);
62 }
63
64 pub fn pending(self: *const Chunker) usize {
65 return self.row_count;
66 }
67
68 pub fn view(self: *const Chunker, rows_scratch: *[max_rows]version.RelationRow) []version.RelationRow {
69 std.debug.assert(self.row_count > 0);
70 std.debug.assert(self.row_count <= max_rows);
71 for (
72 rows_scratch[0..self.row_count],
73 self.rowids[0..self.row_count],
74 self.offsets[0..self.row_count],
75 self.lengths[0..self.row_count],
76 ) |*slot, rowid, offset, length| {
77 slot.* = .{ .rowid = rowid, .bytes = self.bytes.items[offset..][0..length] };
78 }
79 return rows_scratch[0..self.row_count];
80 }
81
82 pub fn reset(self: *Chunker) void {
83 std.debug.assert(self.row_count > 0);
84 self.row_count = 0;
85 self.bytes.clearRetainingCapacity();
86 }
87 };
88
89 pub fn digest(rows: []const version.RelationRow) version.Hash {
90 var hasher = std.crypto.hash.sha2.Sha256.init(.{});
91 hasher.update("tiny.sql.history.row_chunk");
92 update64(&hasher, @intCast(rows.len));
93 for (rows) |row_value| {
94 update64(&hasher, @bitCast(row_value.rowid));
95 update64(&hasher, @intCast(row_value.bytes.len));
96 hasher.update(row_value.bytes);
97 }
98 var value: version.Hash = undefined;
99 hasher.final(&value);
100 return value;
101 }
102
103 pub const page_min_hashes: usize = 16;
104 pub const page_max_hashes: usize = 512;
105 const page_boundary_mask: u64 = 63;
106
107 pub fn pageBoundaries(allocator: Allocator, hashes: []const version.Hash) Allocator.Error![]Bounds {
108 var list: std.ArrayList(Bounds) = .empty;
109 defer list.deinit(allocator);
110 var start: usize = 0;
111 for (hashes, 0..) |hash, index| {
112 const length = index + 1 - start;
113 const split = length >= page_max_hashes or (length >= page_min_hashes and isPageBoundary(hash));
114 if (!split) continue;
115 try list.append(allocator, .{ .start = start, .end = index + 1 });
116 start = index + 1;
117 }
118 if (start < hashes.len) try list.append(allocator, .{ .start = start, .end = hashes.len });
119 return try list.toOwnedSlice(allocator);
120 }
121
122 pub fn pageDigest(hashes: []const version.Hash) version.Hash {
123 var hasher = std.crypto.hash.sha2.Sha256.init(.{});
124 hasher.update("tiny.sql.history.chunk_index_page");
125 update64(&hasher, @intCast(hashes.len));
126 for (hashes) |hash| hasher.update(hash[0..]);
127 var value: version.Hash = undefined;
128 hasher.final(&value);
129 return value;
130 }
131
132 fn isBoundary(row_value: version.RelationRow) bool {
133 var hasher = std.hash.Wyhash.init(boundary_seed);
134 var encoded: [8]u8 = undefined;
135 std.mem.writeInt(u64, encoded[0..], @bitCast(row_value.rowid), .big);
136 hasher.update(encoded[0..]);
137 hasher.update(row_value.bytes);
138 return (hasher.final() & boundary_mask) == 0;
139 }
140
141 fn isPageBoundary(hash: version.Hash) bool {
142 return (std.mem.readInt(u64, hash[0..8], .big) & page_boundary_mask) == 0;
143 }
144
145 fn update64(hasher: *std.crypto.hash.sha2.Sha256, value: u64) void {
146 var encoded: [8]u8 = undefined;
147 std.mem.writeInt(u64, encoded[0..], value, .big);
148 hasher.update(encoded[0..]);
149 }
150
151 fn buildRows(allocator: Allocator, count: usize, seed: u64, stride: i64) ![]version.RelationRow {
152 var prng = std.Random.DefaultPrng.init(seed);
153 const random = prng.random();
154 const rows = try allocator.alloc(version.RelationRow, count);
155 var built: usize = 0;
156 errdefer version.freeRelationRows(allocator, rows[0..built]);
157 for (rows, 0..) |*row_value, index| {
158 const length = 16 + random.uintLessThan(usize, 48);
159 const bytes = try allocator.alloc(u8, length);
160 random.bytes(bytes);
161 row_value.* = .{
162 .rowid = stride * @as(i64, @intCast(index + 1)),
163 .bytes = bytes,
164 };
165 built += 1;
166 }
167 return rows;
168 }
169
170 fn containsDigest(digests: []const version.Hash, needle: version.Hash) bool {
171 for (digests) |candidate| {
172 if (version.same(candidate, needle)) return true;
173 }
174 return false;
175 }
176
177 fn chunkDigests(allocator: Allocator, rows: []const version.RelationRow) ![]version.Hash {
178 const bounds = try boundaries(allocator, rows);
179 defer allocator.free(bounds);
180 const digests = try allocator.alloc(version.Hash, bounds.len);
181 for (bounds, digests) |bound, *value| value.* = digest(rows[bound.start..bound.end]);
182 return digests;
183 }
184
185 test "chunk boundaries cover rows contiguously within limits" {
186 const allocator = std.testing.allocator;
187 const rows = try buildRows(allocator, 500, 11, 1);
188 defer version.freeRelationRows(allocator, rows);
189 const bounds = try boundaries(allocator, rows);
190 defer allocator.free(bounds);
191 try std.testing.expect(bounds.len > 1);
192 var cursor: usize = 0;
193 for (bounds) |bound| {
194 try std.testing.expectEqual(cursor, bound.start);
195 try std.testing.expect(bound.end > bound.start);
196 try std.testing.expect(bound.end - bound.start <= max_rows);
197 cursor = bound.end;
198 }
199 try std.testing.expectEqual(rows.len, cursor);
200 }
201
202 test "chunk boundaries empty rows produce no chunks" {
203 const allocator = std.testing.allocator;
204 const bounds = try boundaries(allocator, &.{});
205 defer allocator.free(bounds);
206 try std.testing.expectEqual(@as(usize, 0), bounds.len);
207 }
208
209 test "chunk boundaries split oversized rows" {
210 const allocator = std.testing.allocator;
211 const big = try allocator.alloc(u8, max_chunk_bytes);
212 defer allocator.free(big);
213 @memset(big, 0xab);
214 var rows = [_]version.RelationRow{
215 .{ .rowid = 1, .bytes = big },
216 .{ .rowid = 2, .bytes = big },
217 };
218 const bounds = try boundaries(allocator, rows[0..]);
219 defer allocator.free(bounds);
220 try std.testing.expectEqual(@as(usize, 2), bounds.len);
221 }
222
223 test "streaming chunker matches slice boundaries and digests" {
224 const allocator = std.testing.allocator;
225 const seeds = [_]u64{ 3, 17, 91, 257 };
226 for (seeds) |seed| {
227 const rows = try buildRows(allocator, 700, seed, 1);
228 defer version.freeRelationRows(allocator, rows);
229 const bounds = try boundaries(allocator, rows);
230 defer allocator.free(bounds);
231
232 var chunker = Chunker{};
233 defer chunker.deinit(allocator);
234 var rows_scratch: [max_rows]version.RelationRow = undefined;
235 var emitted: usize = 0;
236 for (rows) |row_value| {
237 if (!try chunker.append(allocator, row_value.rowid, row_value.bytes)) continue;
238 const chunk_rows = chunker.view(&rows_scratch);
239 const bound = bounds[emitted];
240 try std.testing.expectEqual(bound.end - bound.start, chunk_rows.len);
241 try std.testing.expect(version.same(digest(rows[bound.start..bound.end]), digest(chunk_rows)));
242 chunker.reset();
243 emitted += 1;
244 }
245 if (chunker.pending() != 0) {
246 const chunk_rows = chunker.view(&rows_scratch);
247 const bound = bounds[emitted];
248 try std.testing.expectEqual(bound.end - bound.start, chunk_rows.len);
249 try std.testing.expect(version.same(digest(rows[bound.start..bound.end]), digest(chunk_rows)));
250 chunker.reset();
251 emitted += 1;
252 }
253 try std.testing.expectEqual(bounds.len, emitted);
254 }
255 }
256
257 test "chunk digest distinguishes content and order" {
258 const allocator = std.testing.allocator;
259 const rows = try buildRows(allocator, 16, 7, 1);
260 defer version.freeRelationRows(allocator, rows);
261 const same_digest = digest(rows);
262 try std.testing.expect(version.same(same_digest, digest(rows)));
263 var swapped = try allocator.dupe(version.RelationRow, rows);
264 defer allocator.free(swapped);
265 std.mem.swap(version.RelationRow, &swapped[0], &swapped[1]);
266 try std.testing.expect(!version.same(same_digest, digest(swapped)));
267 }
268
269 test "chunk boundaries stable under single row insert" {
270 const allocator = std.testing.allocator;
271 const rows = try buildRows(allocator, 400, 23, 2);
272 defer version.freeRelationRows(allocator, rows);
273 const before = try chunkDigests(allocator, rows);
274 defer allocator.free(before);
275
276 const inserted_bytes = try allocator.dupe(u8, "inserted row payload bytes");
277 defer allocator.free(inserted_bytes);
278 var with_insert = try allocator.alloc(version.RelationRow, rows.len + 1);
279 defer allocator.free(with_insert);
280 const middle = rows.len / 2;
281 @memcpy(with_insert[0..middle], rows[0..middle]);
282 with_insert[middle] = .{ .rowid = rows[middle].rowid - 1, .bytes = inserted_bytes };
283 @memcpy(with_insert[middle + 1 ..], rows[middle..]);
284
285 const after = try chunkDigests(allocator, with_insert);
286 defer allocator.free(after);
287 var shared: usize = 0;
288 for (after) |value| {
289 if (containsDigest(before, value)) shared += 1;
290 }
291 try std.testing.expect(after.len >= 3);
292 try std.testing.expect(shared + 3 >= after.len);
293 }
294
295 fn buildHashes(allocator: Allocator, count: usize, seed: u64) ![]version.Hash {
296 var prng = std.Random.DefaultPrng.init(seed);
297 const random = prng.random();
298 const hashes = try allocator.alloc(version.Hash, count);
299 for (hashes) |*hash| random.bytes(hash[0..]);
300 return hashes;
301 }
302
303 test "page boundaries cover hashes contiguously within limits" {
304 const allocator = std.testing.allocator;
305 const hashes = try buildHashes(allocator, 3_000, 17);
306 defer allocator.free(hashes);
307 const bounds = try pageBoundaries(allocator, hashes);
308 defer allocator.free(bounds);
309 try std.testing.expect(bounds.len > 1);
310 var cursor: usize = 0;
311 for (bounds) |bound| {
312 try std.testing.expectEqual(cursor, bound.start);
313 try std.testing.expect(bound.end > bound.start);
314 try std.testing.expect(bound.end - bound.start <= page_max_hashes);
315 cursor = bound.end;
316 }
317 try std.testing.expectEqual(hashes.len, cursor);
318 }
319
320 test "page boundaries empty hashes produce no pages" {
321 const allocator = std.testing.allocator;
322 const bounds = try pageBoundaries(allocator, &.{});
323 defer allocator.free(bounds);
324 try std.testing.expectEqual(@as(usize, 0), bounds.len);
325 }
326
327 test "page boundaries stable under single hash replacement" {
328 const allocator = std.testing.allocator;
329 const hashes = try buildHashes(allocator, 3_000, 29);
330 defer allocator.free(hashes);
331 const before = try pageBoundaries(allocator, hashes);
332 defer allocator.free(before);
333 var replaced = try allocator.dupe(version.Hash, hashes);
334 defer allocator.free(replaced);
335 @memset(replaced[1_500][0..], 0xee);
336 const after = try pageBoundaries(allocator, replaced);
337 defer allocator.free(after);
338 var shared: usize = 0;
339 for (after) |bound| {
340 for (before) |candidate| {
341 if (candidate.start == bound.start and candidate.end == bound.end) {
342 shared += 1;
343 break;
344 }
345 }
346 }
347 try std.testing.expect(after.len >= 3);
348 try std.testing.expect(shared + 3 >= after.len);
349 }
350
351 test "page digest distinguishes content and order" {
352 const allocator = std.testing.allocator;
353 const hashes = try buildHashes(allocator, 64, 31);
354 defer allocator.free(hashes);
355 const same = pageDigest(hashes);
356 try std.testing.expect(version.same(same, pageDigest(hashes)));
357 var swapped = try allocator.dupe(version.Hash, hashes);
358 defer allocator.free(swapped);
359 std.mem.swap(version.Hash, &swapped[0], &swapped[1]);
360 try std.testing.expect(!version.same(same, pageDigest(swapped)));
361 }