feat(store): add manifest persistence, statistics, and scan api
This commit is contained in:
1 parent
f11ba18b9d
commit
860f3861aa
1 file changed
+145
+145
@@ -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,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user