feat(store): add leveled compaction

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;
+    }
+