feat(store): add reverse range iterator and manifest loading
This commit is contained in:
1 parent
f574248ccb
commit
c312659931
1 file changed
+130
+130
@@ -755,3 +755,133 @@ pub const RangeIterator = struct {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const ReverseSource = union(enum) {
|
||||||
|
mem: memtable.ReverseCursor,
|
||||||
|
sst: sstable.ReverseCursor,
|
||||||
|
|
||||||
|
fn valid(self: *ReverseSource) bool {
|
||||||
|
return switch (self.*) {
|
||||||
|
.mem => |*c| c.valid(),
|
||||||
|
.sst => |*c| c.valid(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
fn record(self: *ReverseSource) Record {
|
||||||
|
return switch (self.*) {
|
||||||
|
.mem => |*c| c.record(),
|
||||||
|
.sst => |*c| c.record(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
fn advance(self: *ReverseSource) void {
|
||||||
|
switch (self.*) {
|
||||||
|
.mem => |*c| c.advance(),
|
||||||
|
.sst => |*c| c.advance(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn deinit(self: *ReverseSource) void {
|
||||||
|
switch (self.*) {
|
||||||
|
.mem => {},
|
||||||
|
.sst => |*c| c.deinit(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
pub const ReverseRangeIterator = struct {
|
||||||
|
resultAllocator: Allocator,
|
||||||
|
arena: std.heap.ArenaAllocator,
|
||||||
|
cursors: []ReverseSource,
|
||||||
|
start: ?[]u8,
|
||||||
|
startInclusive: bool,
|
||||||
|
maxSeq: u64,
|
||||||
|
|
||||||
|
pub fn next(self: *ReverseRangeIterator) !?Entry {
|
||||||
|
while (true) {
|
||||||
|
var maxKey: ?[]const u8 = null;
|
||||||
|
for (self.cursors) |*c| {
|
||||||
|
if (c.valid()) {
|
||||||
|
const key = c.record().key;
|
||||||
|
if (maxKey == null or order(key, maxKey.?) == .gt) maxKey = key;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (maxKey == null) return null;
|
||||||
|
if (self.start) |bound| {
|
||||||
|
const past = if (self.startInclusive) order(maxKey.?, bound) == .lt else order(maxKey.?, bound) != .gt;
|
||||||
|
if (past) return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
var winner: ?Record = null;
|
||||||
|
for (self.cursors) |*c| {
|
||||||
|
if (c.valid() and order(c.record().key, maxKey.?) == .eq) {
|
||||||
|
const rec = c.record();
|
||||||
|
if (rec.seq <= self.maxSeq and (winner == null or rec.seq > winner.?.seq)) winner = rec;
|
||||||
|
c.advance();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (winner == null) continue;
|
||||||
|
if (winner.?.kind == .tombstone) continue;
|
||||||
|
|
||||||
|
const arena = self.arena.allocator();
|
||||||
|
return Entry{
|
||||||
|
.key = try arena.dupe(u8, winner.?.key),
|
||||||
|
.value = try arena.dupe(u8, winner.?.value),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn deinit(self: *ReverseRangeIterator) void {
|
||||||
|
self.arena.deinit();
|
||||||
|
for (self.cursors) |*c| c.deinit();
|
||||||
|
if (self.start) |bound| self.resultAllocator.free(bound);
|
||||||
|
self.resultAllocator.free(self.cursors);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
fn rangesOverlap(aMin: []const u8, aMax: []const u8, bMin: []const u8, bMax: []const u8) bool {
|
||||||
|
return order(aMin, bMax) != .gt and order(bMin, aMax) != .gt;
|
||||||
|
}
|
||||||
|
|
||||||
|
fn loadManifest(allocator: Allocator, dir: std.fs.Dir) !std.AutoHashMap(u64, usize) {
|
||||||
|
var map = std.AutoHashMap(u64, usize).init(allocator);
|
||||||
|
errdefer map.deinit();
|
||||||
|
|
||||||
|
const file = dir.openFile(manifestName, .{}) catch |err| switch (err) {
|
||||||
|
error.FileNotFound => return map,
|
||||||
|
else => return err,
|
||||||
|
};
|
||||||
|
defer file.close();
|
||||||
|
const data = try file.readToEndAlloc(allocator, std.math.maxInt(usize));
|
||||||
|
defer allocator.free(data);
|
||||||
|
if (data.len < 8) return map;
|
||||||
|
|
||||||
|
const storedCrc = encode.readU32(data, 0);
|
||||||
|
const body = data[4..];
|
||||||
|
if (encode.crc(body) != storedCrc) return map;
|
||||||
|
|
||||||
|
const count = encode.readU32(body, 0);
|
||||||
|
var pos: usize = 4;
|
||||||
|
var i: u32 = 0;
|
||||||
|
while (i < count) : (i += 1) {
|
||||||
|
if (pos + 12 > body.len) break;
|
||||||
|
const fileId = encode.readU64(body, pos);
|
||||||
|
const level = encode.readU32(body, pos + 8);
|
||||||
|
try map.put(fileId, level);
|
||||||
|
pos += 12;
|
||||||
|
}
|
||||||
|
return map;
|
||||||
|
}
|
||||||
|
|
||||||
|
fn sstName(buf: []u8, id: u64) ![]const u8 {
|
||||||
|
return std.fmt.bufPrint(buf, "sst-{d:0>10}.cbt", .{id});
|
||||||
|
}
|
||||||
|
|
||||||
|
fn parseSstId(name: []const u8) ?u64 {
|
||||||
|
const prefix = "sst-";
|
||||||
|
const suffix = ".cbt";
|
||||||
|
if (!std.mem.startsWith(u8, name, prefix)) return null;
|
||||||
|
if (!std.mem.endsWith(u8, name, suffix)) return null;
|
||||||
|
const digits = name[prefix.len .. name.len - suffix.len];
|
||||||
|
return std.fmt.parseInt(u64, digits, 10) catch null;
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user