tiny.sql.chunk
Defined in tiny.sql.
API (16)
Actions
Public operations.
Chunker.appendChunker.deinitChunker.pendingChunker.resetChunker.viewboundariesdigestpageBoundariespageDigest
Types and contracts
Public types and contracts.
Values and defaults
Public values and defaults.
Source
Source: lib/sql/src/chunk.zig
zig
const std = @import("std");const version = @import("version.zig");const Allocator = std.mem.Allocator;pub const min_rows: usize = 8;pub const max_rows: usize = 128;pub const max_chunk_bytes: usize = 64 * 1024;const boundary_mask: u64 = 31;const boundary_seed: u64 = 0x74696e795f73716c;pub const Bounds = struct { start: usize, end: usize,};pub fn boundaries(allocator: Allocator, rows: []const version.RelationRow) Allocator.Error![]Bounds { var list: std.ArrayList(Bounds) = .empty; defer list.deinit(allocator); var start: usize = 0; var bytes: usize = 0; for (rows, 0..) |row_value, index| { bytes += row_value.bytes.len; if (!shouldSplit(index + 1 - start, bytes, row_value)) continue; try list.append(allocator, .{ .start = start, .end = index + 1 }); start = index + 1; bytes = 0; } if (start < rows.len) try list.append(allocator, .{ .start = start, .end = rows.len }); return try list.toOwnedSlice(allocator);}fn shouldSplit(length: usize, bytes: usize, appended: version.RelationRow) bool { return length >= max_rows or bytes >= max_chunk_bytes or (length >= min_rows and isBoundary(appended));}pub const Chunker = struct { rowids: [max_rows]i64 = undefined, offsets: [max_rows]usize = undefined, lengths: [max_rows]usize = undefined, row_count: usize = 0, bytes: std.ArrayList(u8) = .empty, pub fn deinit(self: *Chunker, allocator: Allocator) void { self.bytes.deinit(allocator); self.* = undefined; } pub fn append(self: *Chunker, allocator: Allocator, rowid: i64, row_bytes: []const u8) Allocator.Error!bool { std.debug.assert(self.row_count < max_rows); const offset = self.bytes.items.len; try self.bytes.appendSlice(allocator, row_bytes); self.rowids[self.row_count] = rowid; self.offsets[self.row_count] = offset; self.lengths[self.row_count] = row_bytes.len; self.row_count += 1; const appended = version.RelationRow{ .rowid = rowid, .bytes = self.bytes.items[offset..][0..row_bytes.len], }; return shouldSplit(self.row_count, self.bytes.items.len, appended); } pub fn pending(self: *const Chunker) usize { return self.row_count; } pub fn view(self: *const Chunker, rows_scratch: *[max_rows]version.RelationRow) []version.RelationRow { std.debug.assert(self.row_count > 0); std.debug.assert(self.row_count <= max_rows); for ( rows_scratch[0..self.row_count], self.rowids[0..self.row_count], self.offsets[0..self.row_count], self.lengths[0..self.row_count], ) |*slot, rowid, offset, length| { slot.* = .{ .rowid = rowid, .bytes = self.bytes.items[offset..][0..length] }; } return rows_scratch[0..self.row_count]; } pub fn reset(self: *Chunker) void { std.debug.assert(self.row_count > 0); self.row_count = 0; self.bytes.clearRetainingCapacity(); }};pub fn digest(rows: []const version.RelationRow) version.Hash { var hasher = std.crypto.hash.sha2.Sha256.init(.{}); hasher.update("tiny.sql.history.row_chunk"); update64(&hasher, @intCast(rows.len)); for (rows) |row_value| { update64(&hasher, @bitCast(row_value.rowid)); update64(&hasher, @intCast(row_value.bytes.len)); hasher.update(row_value.bytes); } var value: version.Hash = undefined; hasher.final(&value); return value;}pub const page_min_hashes: usize = 16;pub const page_max_hashes: usize = 512;const page_boundary_mask: u64 = 63;pub fn pageBoundaries(allocator: Allocator, hashes: []const version.Hash) Allocator.Error![]Bounds { var list: std.ArrayList(Bounds) = .empty; defer list.deinit(allocator); var start: usize = 0; for (hashes, 0..) |hash, index| { const length = index + 1 - start; const split = length >= page_max_hashes or (length >= page_min_hashes and isPageBoundary(hash)); if (!split) continue; try list.append(allocator, .{ .start = start, .end = index + 1 }); start = index + 1; } if (start < hashes.len) try list.append(allocator, .{ .start = start, .end = hashes.len }); return try list.toOwnedSlice(allocator);}pub fn pageDigest(hashes: []const version.Hash) version.Hash { var hasher = std.crypto.hash.sha2.Sha256.init(.{}); hasher.update("tiny.sql.history.chunk_index_page"); update64(&hasher, @intCast(hashes.len)); for (hashes) |hash| hasher.update(hash[0..]); var value: version.Hash = undefined; hasher.final(&value); return value;}fn isBoundary(row_value: version.RelationRow) bool { var hasher = std.hash.Wyhash.init(boundary_seed); var encoded: [8]u8 = undefined; std.mem.writeInt(u64, encoded[0..], @bitCast(row_value.rowid), .big); hasher.update(encoded[0..]); hasher.update(row_value.bytes); return (hasher.final() & boundary_mask) == 0;}fn isPageBoundary(hash: version.Hash) bool { return (std.mem.readInt(u64, hash[0..8], .big) & page_boundary_mask) == 0;}fn update64(hasher: *std.crypto.hash.sha2.Sha256, value: u64) void { var encoded: [8]u8 = undefined; std.mem.writeInt(u64, encoded[0..], value, .big); hasher.update(encoded[0..]);}fn buildRows(allocator: Allocator, count: usize, seed: u64, stride: i64) ![]version.RelationRow { var prng = std.Random.DefaultPrng.init(seed); const random = prng.random(); const rows = try allocator.alloc(version.RelationRow, count); var built: usize = 0; errdefer version.freeRelationRows(allocator, rows[0..built]); for (rows, 0..) |*row_value, index| { const length = 16 + random.uintLessThan(usize, 48); const bytes = try allocator.alloc(u8, length); random.bytes(bytes); row_value.* = .{ .rowid = stride * @as(i64, @intCast(index + 1)), .bytes = bytes, }; built += 1; } return rows;}fn containsDigest(digests: []const version.Hash, needle: version.Hash) bool { for (digests) |candidate| { if (version.same(candidate, needle)) return true; } return false;}fn chunkDigests(allocator: Allocator, rows: []const version.RelationRow) ![]version.Hash { const bounds = try boundaries(allocator, rows); defer allocator.free(bounds); const digests = try allocator.alloc(version.Hash, bounds.len); for (bounds, digests) |bound, *value| value.* = digest(rows[bound.start..bound.end]); return digests;}test "chunk boundaries cover rows contiguously within limits" { const allocator = std.testing.allocator; const rows = try buildRows(allocator, 500, 11, 1); defer version.freeRelationRows(allocator, rows); const bounds = try boundaries(allocator, rows); defer allocator.free(bounds); try std.testing.expect(bounds.len > 1); var cursor: usize = 0; for (bounds) |bound| { try std.testing.expectEqual(cursor, bound.start); try std.testing.expect(bound.end > bound.start); try std.testing.expect(bound.end - bound.start <= max_rows); cursor = bound.end; } try std.testing.expectEqual(rows.len, cursor);}test "chunk boundaries empty rows produce no chunks" { const allocator = std.testing.allocator; const bounds = try boundaries(allocator, &.{}); defer allocator.free(bounds); try std.testing.expectEqual(@as(usize, 0), bounds.len);}test "chunk boundaries split oversized rows" { const allocator = std.testing.allocator; const big = try allocator.alloc(u8, max_chunk_bytes); defer allocator.free(big); @memset(big, 0xab); var rows = [_]version.RelationRow{ .{ .rowid = 1, .bytes = big }, .{ .rowid = 2, .bytes = big }, }; const bounds = try boundaries(allocator, rows[0..]); defer allocator.free(bounds); try std.testing.expectEqual(@as(usize, 2), bounds.len);}test "streaming chunker matches slice boundaries and digests" { const allocator = std.testing.allocator; const seeds = [_]u64{ 3, 17, 91, 257 }; for (seeds) |seed| { const rows = try buildRows(allocator, 700, seed, 1); defer version.freeRelationRows(allocator, rows); const bounds = try boundaries(allocator, rows); defer allocator.free(bounds); var chunker = Chunker{}; defer chunker.deinit(allocator); var rows_scratch: [max_rows]version.RelationRow = undefined; var emitted: usize = 0; for (rows) |row_value| { if (!try chunker.append(allocator, row_value.rowid, row_value.bytes)) continue; const chunk_rows = chunker.view(&rows_scratch); const bound = bounds[emitted]; try std.testing.expectEqual(bound.end - bound.start, chunk_rows.len); try std.testing.expect(version.same(digest(rows[bound.start..bound.end]), digest(chunk_rows))); chunker.reset(); emitted += 1; } if (chunker.pending() != 0) { const chunk_rows = chunker.view(&rows_scratch); const bound = bounds[emitted]; try std.testing.expectEqual(bound.end - bound.start, chunk_rows.len); try std.testing.expect(version.same(digest(rows[bound.start..bound.end]), digest(chunk_rows))); chunker.reset(); emitted += 1; } try std.testing.expectEqual(bounds.len, emitted); }}test "chunk digest distinguishes content and order" { const allocator = std.testing.allocator; const rows = try buildRows(allocator, 16, 7, 1); defer version.freeRelationRows(allocator, rows); const same_digest = digest(rows); try std.testing.expect(version.same(same_digest, digest(rows))); var swapped = try allocator.dupe(version.RelationRow, rows); defer allocator.free(swapped); std.mem.swap(version.RelationRow, &swapped[0], &swapped[1]); try std.testing.expect(!version.same(same_digest, digest(swapped)));}test "chunk boundaries stable under single row insert" { const allocator = std.testing.allocator; const rows = try buildRows(allocator, 400, 23, 2); defer version.freeRelationRows(allocator, rows); const before = try chunkDigests(allocator, rows); defer allocator.free(before); const inserted_bytes = try allocator.dupe(u8, "inserted row payload bytes"); defer allocator.free(inserted_bytes); var with_insert = try allocator.alloc(version.RelationRow, rows.len + 1); defer allocator.free(with_insert); const middle = rows.len / 2; @memcpy(with_insert[0..middle], rows[0..middle]); with_insert[middle] = .{ .rowid = rows[middle].rowid - 1, .bytes = inserted_bytes }; @memcpy(with_insert[middle + 1 ..], rows[middle..]); const after = try chunkDigests(allocator, with_insert); defer allocator.free(after); var shared: usize = 0; for (after) |value| { if (containsDigest(before, value)) shared += 1; } try std.testing.expect(after.len >= 3); try std.testing.expect(shared + 3 >= after.len);}fn buildHashes(allocator: Allocator, count: usize, seed: u64) ![]version.Hash { var prng = std.Random.DefaultPrng.init(seed); const random = prng.random(); const hashes = try allocator.alloc(version.Hash, count); for (hashes) |*hash| random.bytes(hash[0..]); return hashes;}test "page boundaries cover hashes contiguously within limits" { const allocator = std.testing.allocator; const hashes = try buildHashes(allocator, 3_000, 17); defer allocator.free(hashes); const bounds = try pageBoundaries(allocator, hashes); defer allocator.free(bounds); try std.testing.expect(bounds.len > 1); var cursor: usize = 0; for (bounds) |bound| { try std.testing.expectEqual(cursor, bound.start); try std.testing.expect(bound.end > bound.start); try std.testing.expect(bound.end - bound.start <= page_max_hashes); cursor = bound.end; } try std.testing.expectEqual(hashes.len, cursor);}test "page boundaries empty hashes produce no pages" { const allocator = std.testing.allocator; const bounds = try pageBoundaries(allocator, &.{}); defer allocator.free(bounds); try std.testing.expectEqual(@as(usize, 0), bounds.len);}test "page boundaries stable under single hash replacement" { const allocator = std.testing.allocator; const hashes = try buildHashes(allocator, 3_000, 29); defer allocator.free(hashes); const before = try pageBoundaries(allocator, hashes); defer allocator.free(before); var replaced = try allocator.dupe(version.Hash, hashes); defer allocator.free(replaced); @memset(replaced[1_500][0..], 0xee); const after = try pageBoundaries(allocator, replaced); defer allocator.free(after); var shared: usize = 0; for (after) |bound| { for (before) |candidate| { if (candidate.start == bound.start and candidate.end == bound.end) { shared += 1; break; } } } try std.testing.expect(after.len >= 3); try std.testing.expect(shared + 3 >= after.len);}test "page digest distinguishes content and order" { const allocator = std.testing.allocator; const hashes = try buildHashes(allocator, 64, 31); defer allocator.free(hashes); const same = pageDigest(hashes); try std.testing.expect(version.same(same, pageDigest(hashes))); var swapped = try allocator.dupe(version.Hash, hashes); defer allocator.free(swapped); std.mem.swap(version.Hash, &swapped[0], &swapped[1]); try std.testing.expect(!version.same(same, pageDigest(swapped)));}Source: lib/sql/src/root.zig:25
zig
pub const chunk = @import("chunk.zig");Audit
| Definitions | 17 |
|---|---|
| Public names | 17 |
| Members | 7 |
| Version | 26.7.0 |
| Revision | daab053ee433 |