diff --git a/src/db.zig b/src/db.zig index f5d8fa5..32522d1 100644 --- a/src/db.zig +++ b/src/db.zig @@ -238,3 +238,230 @@ pub const Db = struct { std.posix.fsync(self.dir.fd) catch {}; } + pub fn compact(self: *Db) !void { + if (self.sstables.items.len <= 1) return; + const fileId = self.nextFileId; + var nameBuf: [64]u8 = undefined; + const name = try sstName(&nameBuf, fileId); + + var expected: usize = 0; + for (self.sstables.items) |reader| expected += @intCast(reader.numEntries); + + const cursors = try self.allocator.alloc(sstable.Cursor, self.sstables.items.len); + defer self.allocator.free(cursors); + for (self.sstables.items, 0..) |reader, i| cursors[i] = reader.cursor(null); + + { + var builder = try Builder.init(self.allocator, self.dir, name, expected); + defer builder.deinit(); + while (pickWinner(cursors)) |winner| { + if (winner.kind == .value) try builder.add(winner); + } + 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(); + + for (self.sstables.items) |old| { + var oldNameBuf: [64]u8 = undefined; + const oldName = sstName(&oldNameBuf, old.fileId) catch unreachable; + old.deinit(); + self.allocator.destroy(old); + self.dir.deleteFile(oldName) catch {}; + } + self.sstables.clearRetainingCapacity(); + reader.level = 0; + try self.sstables.append(reader); + self.nextFileId += 1; + try self.writeManifest(); + std.posix.fsync(self.dir.fd) catch {}; + } + + pub fn maybeCompact(self: *Db) !void { + while (true) { + if (self.countLevel(0) >= self.l0Trigger) { + try self.compactL0(); + continue; + } + var acted = false; + var level: usize = 1; + const bottom = self.maxDataLevel(); + while (level <= bottom) : (level += 1) { + if (self.levelSize(level) > self.levelLimit(level)) { + try self.compactDown(level); + acted = true; + break; + } + } + if (!acted) break; + } + } + + fn compactL0(self: *Db) !void { + var inputs = std.ArrayList(*Reader).init(self.allocator); + defer inputs.deinit(); + var lo: ?[]const u8 = null; + var hi: ?[]const u8 = null; + for (self.sstables.items) |reader| { + if (reader.level != 0) continue; + try inputs.append(reader); + if (reader.index.len == 0) continue; + if (lo == null or order(reader.minKey, lo.?) == .lt) lo = reader.minKey; + if (hi == null or order(reader.maxKey, hi.?) == .gt) hi = reader.maxKey; + } + if (inputs.items.len == 0) return; + if (lo) |lowBound| { + for (self.sstables.items) |reader| { + if (reader.level != 1 or reader.index.len == 0) continue; + if (rangesOverlap(reader.minKey, reader.maxKey, lowBound, hi.?)) try inputs.append(reader); + } + } + const dropTombstones = !self.hasDataBelow(1); + try self.runMerge(inputs.items, 1, dropTombstones); + } + + fn compactDown(self: *Db, level: usize) !void { + var picked: ?*Reader = null; + for (self.sstables.items) |reader| { + if (reader.level != level or reader.index.len == 0) continue; + if (picked == null or order(reader.minKey, picked.?.minKey) == .lt) picked = reader; + } + if (picked == null) return; + var inputs = std.ArrayList(*Reader).init(self.allocator); + defer inputs.deinit(); + try inputs.append(picked.?); + const target = level + 1; + for (self.sstables.items) |reader| { + if (reader.level != target or reader.index.len == 0) continue; + if (rangesOverlap(reader.minKey, reader.maxKey, picked.?.minKey, picked.?.maxKey)) try inputs.append(reader); + } + const dropTombstones = !self.hasDataBelow(target); + try self.runMerge(inputs.items, target, dropTombstones); + } + + fn runMerge(self: *Db, inputs: []*Reader, targetLevel: usize, dropTombstones: bool) !void { + const cursors = try self.allocator.alloc(sstable.Cursor, inputs.len); + defer self.allocator.free(cursors); + for (inputs, 0..) |reader, i| cursors[i] = reader.cursor(null); + + var outputs = std.ArrayList(*Reader).init(self.allocator); + errdefer { + for (outputs.items) |produced| { + produced.deinit(); + self.allocator.destroy(produced); + } + outputs.deinit(); + } + + var estimate: usize = 0; + for (inputs) |reader| estimate += @intCast(reader.numEntries); + if (estimate == 0) estimate = 1; + + var builder: ?Builder = null; + var outputId: u64 = 0; + var outputName: [64]u8 = undefined; + errdefer if (builder) |*b| b.deinit(); + + while (pickWinner(cursors)) |winner| { + if (dropTombstones and winner.kind == .tombstone) continue; + if (builder == null) { + outputId = self.nextFileId; + self.nextFileId += 1; + const name = try sstName(&outputName, outputId); + builder = try Builder.init(self.allocator, self.dir, name, estimate); + } + try builder.?.add(winner); + if (builder.?.logicalBytes >= self.sstableTargetBytes) { + const done = builder.?; + builder = null; + try self.finishOutput(done, outputId, targetLevel, &outputs); + } + } + if (builder) |done| { + builder = null; + try self.finishOutput(done, outputId, targetLevel, &outputs); + } + + for (inputs) |old| { + self.removeReader(old); + var oldNameBuf: [64]u8 = undefined; + const oldName = sstName(&oldNameBuf, old.fileId) catch unreachable; + old.deinit(); + self.allocator.destroy(old); + self.dir.deleteFile(oldName) catch {}; + } + for (outputs.items) |produced| try self.sstables.append(produced); + outputs.deinit(); + + try self.writeManifest(); + std.posix.fsync(self.dir.fd) catch {}; + } + + fn finishOutput(self: *Db, builderIn: Builder, fileId: u64, targetLevel: usize, outputs: *std.ArrayList(*Reader)) !void { + var builder = builderIn; + { + errdefer builder.deinit(); + try builder.finish(); + } + builder.deinit(); + + var nameBuf: [64]u8 = undefined; + const name = try sstName(&nameBuf, fileId); + 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 = targetLevel; + try outputs.append(reader); + } + + fn removeReader(self: *Db, target: *Reader) void { + for (self.sstables.items, 0..) |reader, i| { + if (reader == target) { + _ = self.sstables.orderedRemove(i); + return; + } + } + } + + fn countLevel(self: *Db, level: usize) usize { + var n: usize = 0; + for (self.sstables.items) |reader| { + if (reader.level == level) n += 1; + } + return n; + } + + fn levelSize(self: *Db, level: usize) usize { + var total: usize = 0; + for (self.sstables.items) |reader| { + if (reader.level == level) total += reader.buffer.len; + } + return total; + } + + fn levelLimit(self: *Db, level: usize) usize { + var limit = self.l1BaseBytes; + var i: usize = 1; + while (i < level) : (i += 1) limit *= self.levelMultiplier; + return limit; + } + + fn maxDataLevel(self: *Db) usize { + var maxLevel: usize = 0; + for (self.sstables.items) |reader| { + if (reader.level > maxLevel) maxLevel = reader.level; + } + return maxLevel; + } + + fn hasDataBelow(self: *Db, level: usize) bool { + for (self.sstables.items) |reader| { + if (reader.level > level and reader.numEntries > 0) return true; + } + return false; + } +