feat(store): add database open, recovery, and lifecycle
This commit is contained in:
1 parent
bb1625d405
commit
6be9d8b710
1 file changed
+139
+139
@@ -0,0 +1,139 @@
|
||||
const std = @import("std");
|
||||
const encode = @import("encode.zig");
|
||||
const memtable = @import("memtable.zig");
|
||||
const sstable = @import("sstable.zig");
|
||||
const cache = @import("cache.zig");
|
||||
const statsmod = @import("stats.zig");
|
||||
const Wal = @import("wal.zig").Wal;
|
||||
|
||||
const Allocator = std.mem.Allocator;
|
||||
const Kind = encode.Kind;
|
||||
const Record = encode.Record;
|
||||
const Memtable = memtable.Memtable;
|
||||
const Reader = sstable.Reader;
|
||||
const Builder = sstable.Builder;
|
||||
const BlockCache = cache.BlockCache;
|
||||
|
||||
pub const WriteBatch = @import("batch.zig").WriteBatch;
|
||||
pub const Stats = statsmod.Stats;
|
||||
|
||||
fn order(a: []const u8, b: []const u8) std.math.Order {
|
||||
return std.mem.order(u8, a, b);
|
||||
}
|
||||
|
||||
const walName = "wal.log";
|
||||
const manifestName = "MANIFEST";
|
||||
|
||||
pub const Options = struct {
|
||||
flushThreshold: usize = 4 * 1024 * 1024,
|
||||
blockCacheBytes: usize = 8 * 1024 * 1024,
|
||||
l0CompactionTrigger: usize = 4,
|
||||
l1BaseBytes: usize = 8 * 1024 * 1024,
|
||||
levelMultiplier: usize = 10,
|
||||
sstableTargetBytes: usize = 2 * 1024 * 1024,
|
||||
};
|
||||
|
||||
pub const Entry = struct {
|
||||
key: []const u8,
|
||||
value: []const u8,
|
||||
};
|
||||
|
||||
pub const Db = struct {
|
||||
allocator: Allocator,
|
||||
dir: std.fs.Dir,
|
||||
dirPath: []u8,
|
||||
memtable: Memtable,
|
||||
wal: Wal,
|
||||
sstables: std.ArrayList(*Reader),
|
||||
blockCache: BlockCache,
|
||||
nextFileId: u64,
|
||||
nextSeq: u64,
|
||||
flushThreshold: usize,
|
||||
l0Trigger: usize,
|
||||
l1BaseBytes: usize,
|
||||
levelMultiplier: usize,
|
||||
sstableTargetBytes: usize,
|
||||
|
||||
pub fn open(allocator: Allocator, dirPath: []const u8, options: Options) !Db {
|
||||
try std.fs.cwd().makePath(dirPath);
|
||||
var dir = try std.fs.cwd().openDir(dirPath, .{ .iterate = true });
|
||||
errdefer dir.close();
|
||||
|
||||
var sstables = std.ArrayList(*Reader).init(allocator);
|
||||
errdefer {
|
||||
for (sstables.items) |reader| {
|
||||
reader.deinit();
|
||||
allocator.destroy(reader);
|
||||
}
|
||||
sstables.deinit();
|
||||
}
|
||||
|
||||
var ids = std.ArrayList(u64).init(allocator);
|
||||
defer ids.deinit();
|
||||
var walker = dir.iterate();
|
||||
while (try walker.next()) |item| {
|
||||
if (item.kind != .file) continue;
|
||||
if (parseSstId(item.name)) |id| try ids.append(id);
|
||||
}
|
||||
std.mem.sort(u64, ids.items, {}, std.sort.desc(u64));
|
||||
|
||||
var maxSeq: u64 = 0;
|
||||
var maxId: u64 = 0;
|
||||
for (ids.items) |id| {
|
||||
var nameBuf: [64]u8 = undefined;
|
||||
const name = try sstName(&nameBuf, id);
|
||||
const reader = try allocator.create(Reader);
|
||||
errdefer allocator.destroy(reader);
|
||||
reader.* = try Reader.open(allocator, dir, name, id);
|
||||
errdefer reader.deinit();
|
||||
try sstables.append(reader);
|
||||
if (reader.maxSeq > maxSeq) maxSeq = reader.maxSeq;
|
||||
if (id > maxId) maxId = id;
|
||||
}
|
||||
|
||||
var levelMap = try loadManifest(allocator, dir);
|
||||
defer levelMap.deinit();
|
||||
for (sstables.items) |reader| {
|
||||
reader.level = levelMap.get(reader.fileId) orelse 0;
|
||||
}
|
||||
|
||||
var table = try Memtable.init(allocator);
|
||||
errdefer table.deinit();
|
||||
var wal = try Wal.open(allocator, dir, walName);
|
||||
errdefer wal.close();
|
||||
const walMaxSeq = try wal.replay(&table);
|
||||
if (walMaxSeq > maxSeq) maxSeq = walMaxSeq;
|
||||
|
||||
const dirPathCopy = try allocator.dupe(u8, dirPath);
|
||||
|
||||
return .{
|
||||
.allocator = allocator,
|
||||
.dir = dir,
|
||||
.dirPath = dirPathCopy,
|
||||
.memtable = table,
|
||||
.wal = wal,
|
||||
.sstables = sstables,
|
||||
.blockCache = BlockCache.init(allocator, options.blockCacheBytes),
|
||||
.nextFileId = maxId + 1,
|
||||
.nextSeq = maxSeq + 1,
|
||||
.flushThreshold = options.flushThreshold,
|
||||
.l0Trigger = options.l0CompactionTrigger,
|
||||
.l1BaseBytes = options.l1BaseBytes,
|
||||
.levelMultiplier = options.levelMultiplier,
|
||||
.sstableTargetBytes = options.sstableTargetBytes,
|
||||
};
|
||||
}
|
||||
|
||||
pub fn close(self: *Db) void {
|
||||
self.wal.close();
|
||||
for (self.sstables.items) |reader| {
|
||||
reader.deinit();
|
||||
self.allocator.destroy(reader);
|
||||
}
|
||||
self.sstables.deinit();
|
||||
self.blockCache.deinit();
|
||||
self.memtable.deinit();
|
||||
self.dir.close();
|
||||
self.allocator.free(self.dirPath);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user