feat(sstable): add block builder and writer

src/sstable.zig

@@ -0,0 +1,157 @@
+const std = @import("std");
+const encode = @import("encode.zig");
+const Bloom = @import("bloom.zig").Bloom;
+const cache = @import("cache.zig");
+const compress = @import("compress.zig");
+
+const Allocator = std.mem.Allocator;
+const Kind = encode.Kind;
+const Record = encode.Record;
+const BlockCache = cache.BlockCache;
+const BlockKey = cache.BlockKey;
+const Codec = compress.Codec;
+const readU32 = encode.readU32;
+const readU64 = encode.readU64;
+
+pub const OwnedHit = struct {
+    seq: u64,
+    kind: Kind,
+    value: []u8,
+};
+
+fn order(a: []const u8, b: []const u8) std.math.Order {
+    return std.mem.order(u8, a, b);
+}
+
+const magic: u32 = 0xC0BA1701;
+const version: u32 = 1;
+const footerLen = 4 + 4 + 8 + 4 + 8 + 4 + 8 + 8;
+const recordHeader = 1 + 8 + 4 + 4;
+const defaultBlockTarget = 4096;
+
+pub const Builder = struct {
+    allocator: Allocator,
+    file: std.fs.File,
+    offset: u64,
+    blockBuf: std.ArrayList(u8),
+    blockFirstKey: std.ArrayList(u8),
+    indexBuf: std.ArrayList(u8),
+    numEntries: u64,
+    logicalBytes: u64,
+    maxSeq: u64,
+    bloom: Bloom,
+    blockTarget: usize,
+
+    pub fn init(allocator: Allocator, dir: std.fs.Dir, name: []const u8, expectedKeys: usize) !Builder {
+        const file = try dir.createFile(name, .{ .truncate = true });
+        return .{
+            .allocator = allocator,
+            .file = file,
+            .offset = 0,
+            .blockBuf = std.ArrayList(u8).init(allocator),
+            .blockFirstKey = std.ArrayList(u8).init(allocator),
+            .indexBuf = std.ArrayList(u8).init(allocator),
+            .numEntries = 0,
+            .logicalBytes = 0,
+            .maxSeq = 0,
+            .bloom = try Bloom.init(allocator, expectedKeys, 0.01),
+            .blockTarget = defaultBlockTarget,
+        };
+    }
+
+    pub fn deinit(self: *Builder) void {
+        self.blockBuf.deinit();
+        self.blockFirstKey.deinit();
+        self.indexBuf.deinit();
+        self.bloom.deinit();
+        self.file.close();
+    }
+
+    pub fn add(self: *Builder, record: Record) !void {
+        if (self.blockBuf.items.len == 0) {
+            self.blockFirstKey.clearRetainingCapacity();
+            try self.blockFirstKey.appendSlice(record.key);
+        }
+        const writer = encode.ByteWriter.init(&self.blockBuf);
+        try writer.byte(@intFromEnum(record.kind));
+        try writer.putU64(record.seq);
+        try writer.putU32(@intCast(record.key.len));
+        try writer.putU32(@intCast(record.value.len));
+        try writer.slice(record.key);
+        try writer.slice(record.value);
+
+        self.bloom.add(record.key);
+        self.numEntries += 1;
+        self.logicalBytes += recordHeader + record.key.len + record.value.len;
+        if (record.seq > self.maxSeq) self.maxSeq = record.seq;
+
+        if (self.blockBuf.items.len >= self.blockTarget) try self.flushBlock();
+    }
+
+    fn flushBlock(self: *Builder) !void {
+        if (self.blockBuf.items.len == 0) return;
+        const originalLen: u32 = @intCast(self.blockBuf.items.len);
+
+        const encoded = try compress.encode(self.allocator, self.blockBuf.items);
+        defer encoded.deinit(self.allocator);
+
+        var physical = std.ArrayList(u8).init(self.allocator);
+        defer physical.deinit();
+        const physWriter = encode.ByteWriter.init(&physical);
+        try physWriter.byte(@intFromEnum(encoded.codec));
+        try physWriter.putU32(originalLen);
+        try physWriter.slice(encoded.bytes);
+
+        const physLen: u32 = @intCast(physical.items.len);
+        try self.file.writeAll(physical.items);
+        var crcBuf: [4]u8 = undefined;
+        std.mem.writeInt(u32, &crcBuf, encode.crc(physical.items), encode.endian);
+        try self.file.writeAll(&crcBuf);
+
+        const indexWriter = encode.ByteWriter.init(&self.indexBuf);
+        try indexWriter.putU64(self.offset);
+        try indexWriter.putU32(physLen);
+        try indexWriter.putU32(@intCast(self.blockFirstKey.items.len));
+        try indexWriter.slice(self.blockFirstKey.items);
+
+        self.offset += physLen + 4;
+        self.blockBuf.clearRetainingCapacity();
+    }
+
+    pub fn finish(self: *Builder) !void {
+        try self.flushBlock();
+
+        var bloomBuf = std.ArrayList(u8).init(self.allocator);
+        defer bloomBuf.deinit();
+        try self.bloom.serialize(encode.ByteWriter.init(&bloomBuf));
+        const bloomOffset = self.offset;
+        const bloomLen: u32 = @intCast(bloomBuf.items.len);
+        try self.writeBlock(bloomBuf.items);
+
+        const indexOffset = self.offset;
+        const indexLen: u32 = @intCast(self.indexBuf.items.len);
+        try self.writeBlock(self.indexBuf.items);
+
+        var footer = std.ArrayList(u8).init(self.allocator);
+        defer footer.deinit();
+        const fw = encode.ByteWriter.init(&footer);
+        try fw.putU32(magic);
+        try fw.putU32(version);
+        try fw.putU64(indexOffset);
+        try fw.putU32(indexLen);
+        try fw.putU64(bloomOffset);
+        try fw.putU32(bloomLen);
+        try fw.putU64(self.numEntries);
+        try fw.putU64(self.maxSeq);
+        try self.file.writeAll(footer.items);
+        try self.file.sync();
+    }
+
+    fn writeBlock(self: *Builder, bytes: []const u8) !void {
+        try self.file.writeAll(bytes);
+        var crcBuf: [4]u8 = undefined;
+        std.mem.writeInt(u32, &crcBuf, encode.crc(bytes), encode.endian);
+        try self.file.writeAll(&crcBuf);
+        self.offset += bytes.len + 4;
+    }
+};