feat(store): add write path, point reads, and flush
This commit is contained in:
1 parent
6be9d8b710
commit
57360650ba
1 file changed
+101
+101
@@ -137,3 +137,104 @@ pub const Db = struct {
|
||||
self.allocator.free(self.dirPath);
|
||||
}
|
||||
|
||||
pub fn put(self: *Db, key: []const u8, value: []const u8) !void {
|
||||
try self.write(.value, key, value);
|
||||
}
|
||||
|
||||
pub fn delete(self: *Db, key: []const u8) !void {
|
||||
try self.write(.tombstone, key, &.{});
|
||||
}
|
||||
|
||||
fn write(self: *Db, kind: Kind, key: []const u8, value: []const u8) !void {
|
||||
const seq = self.nextSeq;
|
||||
try self.wal.append(kind, seq, key, value);
|
||||
try self.wal.sync();
|
||||
try self.memtable.upsert(kind, seq, key, value);
|
||||
self.nextSeq += 1;
|
||||
if (self.memtable.byteSize >= self.flushThreshold) try self.flush();
|
||||
}
|
||||
|
||||
pub fn apply(self: *Db, batch: *WriteBatch) !void {
|
||||
if (batch.ops.items.len == 0) return;
|
||||
var records = try self.allocator.alloc(Record, batch.ops.items.len);
|
||||
defer self.allocator.free(records);
|
||||
for (batch.ops.items, 0..) |op, i| {
|
||||
records[i] = .{ .key = op.key, .value = op.value, .seq = self.nextSeq + @as(u64, @intCast(i)), .kind = op.kind };
|
||||
}
|
||||
try self.wal.appendBatch(records);
|
||||
try self.wal.sync();
|
||||
for (records) |record| {
|
||||
try self.memtable.upsert(record.kind, record.seq, record.key, record.value);
|
||||
}
|
||||
self.nextSeq += records.len;
|
||||
if (self.memtable.byteSize >= self.flushThreshold) try self.flush();
|
||||
}
|
||||
|
||||
pub fn get(self: *Db, allocator: Allocator, key: []const u8) !?[]u8 {
|
||||
return self.getAsOf(allocator, key, std.math.maxInt(u64));
|
||||
}
|
||||
|
||||
pub fn getAsOf(self: *Db, allocator: Allocator, key: []const u8, maxSeq: u64) !?[]u8 {
|
||||
var bestSeq: ?u64 = null;
|
||||
var bestKind: Kind = .value;
|
||||
var owned: ?[]u8 = null;
|
||||
errdefer if (owned) |v| allocator.free(v);
|
||||
|
||||
if (self.memtable.findAsOf(key, maxSeq)) |rec| {
|
||||
bestSeq = rec.seq;
|
||||
bestKind = rec.kind;
|
||||
owned = try allocator.dupe(u8, rec.value);
|
||||
}
|
||||
for (self.sstables.items) |reader| {
|
||||
const hit = try reader.findCachedAsOf(&self.blockCache, allocator, key, maxSeq) orelse continue;
|
||||
if (bestSeq == null or hit.seq > bestSeq.?) {
|
||||
if (owned) |v| allocator.free(v);
|
||||
bestSeq = hit.seq;
|
||||
bestKind = hit.kind;
|
||||
owned = hit.value;
|
||||
} else {
|
||||
allocator.free(hit.value);
|
||||
}
|
||||
}
|
||||
|
||||
if (bestSeq == null) return null;
|
||||
if (bestKind == .tombstone) {
|
||||
if (owned) |v| allocator.free(v);
|
||||
return null;
|
||||
}
|
||||
return owned;
|
||||
}
|
||||
|
||||
pub fn snapshot(self: *Db) !Snapshot {
|
||||
try self.flush();
|
||||
return .{ .db = self, .seq = if (self.nextSeq == 0) 0 else self.nextSeq - 1 };
|
||||
}
|
||||
|
||||
pub fn flush(self: *Db) !void {
|
||||
if (self.memtable.count == 0) return;
|
||||
const fileId = self.nextFileId;
|
||||
var nameBuf: [64]u8 = undefined;
|
||||
const name = try sstName(&nameBuf, fileId);
|
||||
|
||||
{
|
||||
var builder = try Builder.init(self.allocator, self.dir, name, self.memtable.count);
|
||||
defer builder.deinit();
|
||||
var cur = self.memtable.cursor(null);
|
||||
while (cur.valid()) : (cur.advance()) try builder.add(cur.record());
|
||||
try builder.finish();
|
||||
}
|
||||
|
||||
const reader = try self.allocator.create(Reader);
|
||||
errdefer self.allocator.destroy(reader);
|
||||
reader.* = try Reader.open(self.allocator, self.dir, name, fileId);
|
||||
errdefer reader.deinit();
|
||||
reader.level = 0;
|
||||
try self.sstables.insert(0, reader);
|
||||
|
||||
self.nextFileId += 1;
|
||||
self.memtable.clear();
|
||||
try self.wal.reset();
|
||||
try self.writeManifest();
|
||||
std.posix.fsync(self.dir.fd) catch {};
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user