From 860f3861aa7d7bb86140222c15ae6a8d4002778d Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 26 Aug 2026 09:00:00 +0200 Subject: [PATCH] feat(store): add manifest persistence, statistics, and scan api --- src/db.zig | 145 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/src/db.zig b/src/db.zig index 32522d1..cbc5212 100644 --- a/src/db.zig +++ b/src/db.zig @@ -465,3 +465,148 @@ pub const Db = struct { return false; } + fn writeManifest(self: *Db) !void { + var buf = std.ArrayList(u8).init(self.allocator); + defer buf.deinit(); + const bodyWriter = encode.ByteWriter.init(&buf); + try bodyWriter.putU32(@intCast(self.sstables.items.len)); + for (self.sstables.items) |reader| { + try bodyWriter.putU64(reader.fileId); + try bodyWriter.putU32(@intCast(reader.level)); + } + + var frame = std.ArrayList(u8).init(self.allocator); + defer frame.deinit(); + const frameWriter = encode.ByteWriter.init(&frame); + try frameWriter.putU32(encode.crc(buf.items)); + try frameWriter.slice(buf.items); + + const tmpName = "MANIFEST.tmp"; + { + const file = try self.dir.createFile(tmpName, .{ .truncate = true }); + defer file.close(); + try file.writeAll(frame.items); + try file.sync(); + } + try self.dir.rename(tmpName, manifestName); + } + + pub fn levelOf(self: *Db, fileId: u64) ?usize { + for (self.sstables.items) |reader| { + if (reader.fileId == fileId) return reader.level; + } + return null; + } + + pub fn latestSeq(self: *Db, key: []const u8) ?u64 { + var best: ?u64 = null; + if (self.memtable.find(key)) |rec| best = rec.seq; + for (self.sstables.items) |reader| { + const rec = reader.find(key) orelse continue; + if (best == null or rec.seq > best.?) best = rec.seq; + } + return best; + } + + pub fn stats(self: *Db) Stats { + var result = Stats{ + .memtableEntries = self.memtable.count, + .memtableBytes = self.memtable.byteSize, + .sstableCount = self.sstables.items.len, + .cacheHits = self.blockCache.stats.hits, + .cacheMisses = self.blockCache.stats.misses, + .nextSeq = self.nextSeq, + }; + for (self.sstables.items) |reader| { + result.liveEntries += reader.numEntries; + result.physicalBytes += reader.physicalBytes; + if (reader.level < statsmod.maxTrackedLevels) { + result.levels[reader.level].tables += 1; + result.levels[reader.level].entries += reader.numEntries; + result.levels[reader.level].physicalBytes += reader.physicalBytes; + } + } + return result; + } + + pub fn scan(self: *Db, allocator: Allocator, start: ?[]const u8, end: ?[]const u8) !RangeIterator { + return self.scanRangeAsOf(allocator, .{ .start = start, .end = end }, std.math.maxInt(u64)); + } + + pub fn scanAsOf(self: *Db, allocator: Allocator, start: ?[]const u8, end: ?[]const u8, maxSeq: u64) !RangeIterator { + return self.scanRangeAsOf(allocator, .{ .start = start, .end = end }, maxSeq); + } + + pub fn scanRange(self: *Db, allocator: Allocator, options: ScanOptions) !RangeIterator { + return self.scanRangeAsOf(allocator, options, std.math.maxInt(u64)); + } + + pub fn scanRangeAsOf(self: *Db, allocator: Allocator, options: ScanOptions, maxSeq: u64) !RangeIterator { + const cursorCount = 1 + self.sstables.items.len; + var cursors = try allocator.alloc(SourceCursor, cursorCount); + errdefer allocator.free(cursors); + cursors[0] = .{ .mem = self.memtable.cursor(options.start) }; + for (self.sstables.items, 0..) |reader, i| cursors[i + 1] = .{ .sst = reader.cursor(options.start) }; + + var startCopy: ?[]u8 = null; + errdefer if (startCopy) |s| allocator.free(s); + if (options.start) |bound| startCopy = try allocator.dupe(u8, bound); + var endCopy: ?[]u8 = null; + if (options.end) |bound| endCopy = try allocator.dupe(u8, bound); + + return .{ + .resultAllocator = allocator, + .arena = std.heap.ArenaAllocator.init(allocator), + .cursors = cursors, + .start = startCopy, + .end = endCopy, + .startInclusive = options.startInclusive, + .endInclusive = options.endInclusive, + .maxSeq = maxSeq, + }; + } + + pub fn scanPrefix(self: *Db, allocator: Allocator, prefix: []const u8) !RangeIterator { + return self.scanPrefixAsOf(allocator, prefix, std.math.maxInt(u64)); + } + + pub fn scanPrefixAsOf(self: *Db, allocator: Allocator, prefix: []const u8, maxSeq: u64) !RangeIterator { + const upper = try prefixUpperBound(allocator, prefix); + defer if (upper) |u| allocator.free(u); + return self.scanRangeAsOf(allocator, .{ .start = prefix, .end = upper }, maxSeq); + } + + pub fn scanReverse(self: *Db, allocator: Allocator, options: ScanOptions) !ReverseRangeIterator { + return self.scanReverseAsOf(allocator, options, std.math.maxInt(u64)); + } + + pub fn scanReverseAsOf(self: *Db, allocator: Allocator, options: ScanOptions, maxSeq: u64) !ReverseRangeIterator { + const cursorCount = 1 + self.sstables.items.len; + var cursors = try allocator.alloc(ReverseSource, cursorCount); + errdefer allocator.free(cursors); + var built: usize = 0; + errdefer { + var i: usize = 0; + while (i < built) : (i += 1) cursors[i].deinit(); + } + cursors[0] = .{ .mem = self.memtable.cursorRev(options.end, options.endInclusive) }; + built = 1; + for (self.sstables.items, 0..) |reader, i| { + cursors[i + 1] = .{ .sst = try reader.reverseCursor(allocator, options.end, options.endInclusive) }; + built += 1; + } + + var startCopy: ?[]u8 = null; + if (options.start) |bound| startCopy = try allocator.dupe(u8, bound); + + return .{ + .resultAllocator = allocator, + .arena = std.heap.ArenaAllocator.init(allocator), + .cursors = cursors, + .start = startCopy, + .startInclusive = options.startInclusive, + .maxSeq = maxSeq, + }; + } +}; +