feat(store): add snapshots and forward range iterator
This commit is contained in:
1 parent
860f3861aa
commit
f574248ccb
1 file changed
+145
+145
@@ -610,3 +610,148 @@ pub const Db = struct {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
pub const ScanOptions = struct {
|
||||||
|
start: ?[]const u8 = null,
|
||||||
|
end: ?[]const u8 = null,
|
||||||
|
startInclusive: bool = true,
|
||||||
|
endInclusive: bool = false,
|
||||||
|
};
|
||||||
|
|
||||||
|
fn prefixUpperBound(allocator: Allocator, prefix: []const u8) !?[]u8 {
|
||||||
|
var end = prefix.len;
|
||||||
|
while (end > 0 and prefix[end - 1] == 0xff) end -= 1;
|
||||||
|
if (end == 0) return null;
|
||||||
|
const upper = try allocator.alloc(u8, end);
|
||||||
|
@memcpy(upper, prefix[0..end]);
|
||||||
|
upper[end - 1] += 1;
|
||||||
|
return upper;
|
||||||
|
}
|
||||||
|
|
||||||
|
pub const Snapshot = struct {
|
||||||
|
db: *Db,
|
||||||
|
seq: u64,
|
||||||
|
|
||||||
|
pub fn get(self: Snapshot, allocator: Allocator, key: []const u8) !?[]u8 {
|
||||||
|
return self.db.getAsOf(allocator, key, self.seq);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn scan(self: Snapshot, allocator: Allocator, start: ?[]const u8, end: ?[]const u8) !RangeIterator {
|
||||||
|
return self.db.scanAsOf(allocator, start, end, self.seq);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn scanRange(self: Snapshot, allocator: Allocator, options: ScanOptions) !RangeIterator {
|
||||||
|
return self.db.scanRangeAsOf(allocator, options, self.seq);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn scanReverse(self: Snapshot, allocator: Allocator, options: ScanOptions) !ReverseRangeIterator {
|
||||||
|
return self.db.scanReverseAsOf(allocator, options, self.seq);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn scanPrefix(self: Snapshot, allocator: Allocator, prefix: []const u8) !RangeIterator {
|
||||||
|
return self.db.scanPrefixAsOf(allocator, prefix, self.seq);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
const SourceCursor = union(enum) {
|
||||||
|
mem: memtable.Cursor,
|
||||||
|
sst: sstable.Cursor,
|
||||||
|
|
||||||
|
fn valid(self: *SourceCursor) bool {
|
||||||
|
return switch (self.*) {
|
||||||
|
.mem => |*c| c.valid(),
|
||||||
|
.sst => |*c| c.valid(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
fn record(self: *SourceCursor) Record {
|
||||||
|
return switch (self.*) {
|
||||||
|
.mem => |*c| c.record(),
|
||||||
|
.sst => |*c| c.record(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
fn advance(self: *SourceCursor) void {
|
||||||
|
switch (self.*) {
|
||||||
|
.mem => |*c| c.advance(),
|
||||||
|
.sst => |*c| c.advance(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
fn pickWinner(cursors: []sstable.Cursor) ?Record {
|
||||||
|
var minKey: ?[]const u8 = null;
|
||||||
|
for (cursors) |*c| {
|
||||||
|
if (c.valid()) {
|
||||||
|
const key = c.record().key;
|
||||||
|
if (minKey == null or order(key, minKey.?) == .lt) minKey = key;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (minKey == null) return null;
|
||||||
|
var winner: ?Record = null;
|
||||||
|
for (cursors) |*c| {
|
||||||
|
if (c.valid() and order(c.record().key, minKey.?) == .eq) {
|
||||||
|
const rec = c.record();
|
||||||
|
if (winner == null or rec.seq > winner.?.seq) winner = rec;
|
||||||
|
c.advance();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return winner.?;
|
||||||
|
}
|
||||||
|
|
||||||
|
pub const RangeIterator = struct {
|
||||||
|
resultAllocator: Allocator,
|
||||||
|
arena: std.heap.ArenaAllocator,
|
||||||
|
cursors: []SourceCursor,
|
||||||
|
start: ?[]u8,
|
||||||
|
end: ?[]u8,
|
||||||
|
startInclusive: bool,
|
||||||
|
endInclusive: bool,
|
||||||
|
maxSeq: u64,
|
||||||
|
|
||||||
|
pub fn next(self: *RangeIterator) !?Entry {
|
||||||
|
while (true) {
|
||||||
|
var minKey: ?[]const u8 = null;
|
||||||
|
for (self.cursors) |*c| {
|
||||||
|
if (c.valid()) {
|
||||||
|
const key = c.record().key;
|
||||||
|
if (minKey == null or order(key, minKey.?) == .lt) minKey = key;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (minKey == null) return null;
|
||||||
|
if (self.end) |bound| {
|
||||||
|
const past = if (self.endInclusive) order(minKey.?, bound) == .gt else order(minKey.?, bound) != .lt;
|
||||||
|
if (past) return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
var winner: ?Record = null;
|
||||||
|
for (self.cursors) |*c| {
|
||||||
|
if (c.valid() and order(c.record().key, minKey.?) == .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 (!self.startInclusive) {
|
||||||
|
if (self.start) |bound| {
|
||||||
|
if (order(winner.?.key, bound) == .eq) 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: *RangeIterator) void {
|
||||||
|
self.arena.deinit();
|
||||||
|
if (self.start) |bound| self.resultAllocator.free(bound);
|
||||||
|
if (self.end) |bound| self.resultAllocator.free(bound);
|
||||||
|
self.resultAllocator.free(self.cursors);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
Reference in new issue
Block a user