LSM-Tree（Log-Structured Merge Tree）是一种高效的键值存储数据结构，广泛应用于NoSQL数据库和大数据处理系统中。其核心思想是通过分层、有序地利用磁盘顺序写入的性能优势，优化写入操作，同时牺牲部分读取性能以换取更高的写入吞吐量。
在互联网的各个基础设施中，不论是数据库还是缓存亦或是大数据框架，LSM-Tree这个数据结构都是很常见的身影。
我每天都在使用这个存储引擎，但是对它的了解还流于表面，所以我想要自己实现一次LSM-Tree加深理解。
本次实现我们采用了Zig语言，简要的实现LSM-Tree的核心功能（读写、数据压缩、持久化，不包含MVCC的内容）。
Zig是一种新兴的系统编程语言，其设计目标是提供现代特性的同时保持低复杂性。
本项目极大的受到了Mini-Lsm这个项目的启发，强烈推荐大家学习这个项目！
在开始自己编写之前，我先简单介绍一下LSM-Tree（Log-Structured Merge Tree）的架构以及读写流程。
LSM-Tree它结合了日志和索引的特点，优化了写入和读取性能。每次写入都是采用append-only的方式，所以写入性能很高。
而作为代价，追加写入会造成存储放大，LSM-Tree时采用了多层SSTable的方式将数据堆叠在硬盘上。所以需要一个合并压缩的过程来回收过多的空间。
写流程
读流程
首先，我们先实现 LSM 存储引擎的内存结构---Memtable。我们选择跳表实现作为 Memtable 的数据结构，因为它支持无锁的并发读写。我们不会深入介绍跳表的工作原理(Redis的同学应该不陌生这个东西)，简单来说，它是一个易于实现的有序键值映射。
Skiplist的实现非常简单，这里我利用Zig编译时的能力实现了一个泛型版本的跳表src/skiplist.zig，有兴趣的小伙伴可以直接去仓库中参观代码。
基于SkipList的能力，我们即可包装出Memtable的基本功能。
我们这个LSM支持WAL功能的，即写入内存表之前要先写入磁盘日志，方便在意外宕机重启后可以恢复数据。
WAL的能力我就不想自己再实现了，于是从网上扒了一个C的实现（Zig集成C语言非常便捷，可以参考与 C 交互）。
map: Map,
lock: RwLock,
wal: ?Wal,
id: usize,
allocator: std.mem.Allocator,
arena: std.heap.ArenaAllocator,
approximate_size: atomic.Value(usize) = atomic.Value(usize).init(0),
fn putToList(self: *Self, key: []const u8, value: []const u8) !void {
{
self.lock.lock();
defer self.lock.unlock();
try self.map.insert(kk, vv);
}
_ = self.approximate_size.fetchAdd(@intCast(key.len + value.len), .monotonic);
}
fn putToWal(self: *Self, key: []const u8, value: []const u8) !void {
// [key-size: 4bytes][key][value-size: 4bytes][value]
if (self.wal) |w| {
var buf = std.ArrayList(u8).init(self.arena.allocator());
var bw = buf.writer();
try bw.writeInt(u32, @intCast(key.len), .big);
_ = try bw.write(key);
try bw.writeInt(u32, @intCast(value.len), .big);
_ = try bw.write(value);
try w.append(buf.items);
}
}
// 写入Memtable，先写WAL，再写skiplist table
pub fn put(self: *Self, key: []const u8, value: []const u8) !void {
try self.putToWal(key, value);
try self.putToList(key, value);
}
pub fn get(self: *Self, key: []const u8, val: *[]const u8) !bool {
self.lock.lockShared();
defer self.lock.unlockShared();
var vv: []const u8 = undefined;
if (try self.map.get(key, &vv)) {
val.* = vv;
return true;
}
return false;
}
注意到这里我们没有实现删除的功能，这里我仿照了RocksDB中的墓碑机制，用空值代表删除，所以删除被put(key, "")取代。
接下来，我们就着手开始实现LSM中另外一个重要元素 --- SSTable。
SSTable（Sorted String Table）是一种不可变的（Immutable）磁盘文件，内部按Key有序排列，存储键值对数据。每个SSTable文件生成后不再修改，更新和删除操作通过追加新记录或标记删除，最终通过合并（Compaction）清理冗余数据。 每当LSM-Tree中的MemTable体积超出阈值，就会将内存中的数据写入SsTable。
每个SSTable由多个Block组成，每个Block是一组KV的package。
Block的编码格式如下：
为了构建一个Block，我们实现了一个BlockBuilder的模块，这部分代码见src/block.zig：
pub const Block = struct {
data_v: std.ArrayList(u8),
offset_v: std.ArrayList(u16),
}
pub const BlockBuilder = struct {
allocator: std.mem.Allocator,
offset_v: std.ArrayList(u16),
data_v: std.ArrayList(u8),
block_size: usize,
first_key: []u8,
pub fn add(self: *Self, key: []const u8, value: ?[]const u8) !bool {
std.debug.assert(key.len > 0); // key must not be empty
const vSize = if (value) |v| v.len else 0;
if ((self.estimated_size() + key.len + vSize + 3 * @sizeOf(u16) > self.block_size) and !self.is_empty()) {
return false;
}
try self.doAdd(key, value);
if (self.first_key.len == 0) {
self.first_key = try self.allocator.dupe(u8, key);
}
return true;
}
fn doAdd(self: *Self, key: []const u8, value: ?[]const u8) !void {
// add the offset of the data into the offset array
try self.offset_v.append(@intCast(self.data_v.items.len));
const overlap = calculate_overlap(self.first_key, key);
var dw = self.data_v.writer();
// encode key overlap
try dw.writeInt(u16, @intCast(overlap), .big);
// encode key length
try dw.writeInt(u16, @intCast(key.len - overlap), .big);
// encode key content
_ = try dw.write(key[overlap..]);
// encode value length
if (value) |v| {
try dw.writeInt(u16, @intCast(v.len), .big);
// encode value content
_ = try dw.write(v);
} else {
try dw.writeInt(u16, 0, .big);
}
}
pub fn build(self: *Self) !Block {
if (self.isEmpty()) {
@panic("block is empty");
}
return Block.init(
try self.data_v.clone(),
try self.offset_v.clone(),
);
}
}
可能有同学注意到，我们写key的时候没有直接将key值写入，而且只写了key与当前block的第一个key不重叠的suffix部分。由于block中的key都是有序的，所以一个block中的key有很大概率是前缀类似的，所以这里是一个空间优化的小技巧，例如：
Key: foo, foo1, foo2, foo3 ....
我们写入block时，只需要写：
foo|1|2|3|.... 很多有序表的实现中都会用到这个小技巧。
有了block的实现，我们可以进一步来定义SSTable的格式。一个SSTable由多个Block、block元数据以及布隆过滤器构成。
布隆过滤器是一种概率性数据结构，用于维护一组键。您可以向布隆过滤器中添加键，并且可以知道在添加到布隆过滤器中的键集中可能存在或必须不存在的键。
在SSTable中添加布隆过滤器可以有效提升查询key的效率。
元数据包含了block的第一个与最后一个key以及block在sst中的offset信息，记录元数据主要为了在后续的检索中可以快速定位某个key落在哪个block中。
同样的套路，为了构建SSTable，我们先实现一个SSTableBuilder，部分代码见src/ss_table.zig
pub const SsTableBuilder = struct {
allocator: std.mem.Allocator,
builder: BlockBuilder, // 刚才实现的block构建装置
first_key: ?[]const u8,
last_key: ?[]const u8,
meta: std.ArrayList(BlockMeta),
block_size: usize,
data: std.ArrayList(u8),
bloom: BloomFilterPtr, // 布隆过滤器
pub fn add(self: *Self, key: []const u8, value: []const u8) !void {
try self.setFirstKey(key);
try self.bloom.get().insert(key); // 写入布隆过滤器
if (try self.builder.add(key, value)) {
try self.setLastKey(key);
return;
}
// block is full
try self.finishBlock();
std.debug.assert(try self.builder.add(key, value));
try self.resetFirstKey(key);
try self.setLastKey(key);
}
// 写入一个block的数据
fn finishBlock(self: *Self) !void {
if (self.builder.isEmpty()) {
return;
}
var bo = self.builder;
// reset block
defer bo.reset();
self.builder = BlockBuilder.init(self.allocator, self.block_size);
var blk = try bo.build();
defer blk.deinit();
const encoded_block = try blk.encode(self.allocator); // block序列化
defer self.allocator.free(encoded_block);
// 记录block的元数据
try self.meta.append(.{
.allocator = self.allocator,
.offset = self.data.items.len,
.first_key = try self.allocator.dupe(u8, self.first_key.?),
.last_key = try self.allocator.dupe(u8, self.last_key.?),
});
const cksm = hash.Crc32.hash(encoded_block); // 写入4b的校验值
try self.data.appendSlice(encoded_block);
try self.data.writer().writeInt(u32, cksm, .big);
}
// 构建为一个SSTable
pub fn build(
self: *Self,
id: usize,
block_cache: ?BlockCachePtr, // 读取block数据的缓存，减少block的反序列化成本
path: []const u8,
) !SsTable {
var arena = std.heap.ArenaAllocator.init(self.allocator);
defer arena.deinit();
const allocator = arena.allocator();
try self.finishBlock();
const w = self.data.writer();
// 写入元数据及其offset
const meta_offset = self.data.items.len;
const meta_b = try BlockMeta.batchEncode(self.meta.items, allocator);
_ = try w.write(meta_b);
try w.writeInt(u32, @intCast(meta_offset), .big);
// 写入布隆过滤器及其offset
const bloom_offset = self.data.items.len;
const encoded_bloom = try self.bloom.get().encode(allocator);
_ = try w.write(encoded_bloom);
try w.writeInt(u32, @intCast(bloom_offset), .big);
const file = try FileObject.init(path, self.data.items);
errdefer file.deinit();
const fk = self.meta.items[0].first_key;
const lk = self.meta.getLast().last_key;
return .{
.allocator = self.allocator,
.file = file,
.block_metas = try self.meta.toOwnedSlice(),
.meta_offset = meta_offset,
.block_cache = block_cache,
.bloom = self.bloom.clone(),
.id = id,
.first_key = try self.allocator.dupe(u8, fk),
.last_key = try self.allocator.dupe(u8, lk),
.max_ts = 0,
};
}
}
有了SSTable和MemTable，我们就有了LSM-Tree需要的两个最重要的材料，后续的读写不过是对这两类材料的组合拼装。
在实现写操作之前，我们先假想一下LSM-Tree的数据结构: 
最终，我们实现的LSM数据结构如下：
pub const StorageState = struct {
allocator: std.mem.Allocator,
mem_table: MemTablePtr, // 当前正在写的MemTable
imm_mem_tables: std.ArrayList(MemTablePtr), // 冷MemTable数组
l0_sstables: std.ArrayList(usize), // 第一层的SSTable数组
levels: std.ArrayList(std.ArrayList(usize)), // 后续多层的SSTable数组
sstables: std.AutoHashMap(usize, SsTablePtr), // sst_id => SSTable
}
pub const StorageInner = struct {
const Self = @This();
allocator: std.mem.Allocator,
state: StorageState,
state_lock: std.Thread.RwLock = .{},
next_sst_id: atomic.Value(usize),
path: []const u8,
options: StorageOptions,
compaction_controller: CompactionController,
block_cache: BlockCachePtr,
terminate: std.Thread.ResetEvent = .{},
wg: std.Thread.WaitGroup = .{},
}
先不考虑逐层压缩的逻辑，只考虑一层SSTable的简单情况，写逻辑可以简化为如下流程：
pub fn writeBatch(self: *Self, records: []const WriteBatchRecord) !void {
for (records) |record| {
switch (record) {
.put => |pp| {
try self.state.getMemTable().put(pp.key, pp.value);
},
.delete => |dd| {
// we use "" as the tombstone value
try self.state.getMemTable().put(dd, "");
},
}
// 尝试把当前MemTable压入冷数据
try self.tryFreeze(self.state.getMemTable().getApproximateSize());
}
}
fn forceFreezeMemtable(self: *Self) !void {
const next_sst_id = self.getNextSstId();
// 生成一个新的MemTable
var new_mm: MemTable = undefined;
{
if (self.options.enable_wal) {
const mm_path = try pathOfWal(self.allocator, self.path, next_sst_id);
defer self.allocator.free(mm_path);
new_mm = MemTable.init(next_sst_id, self.allocator, mm_path);
} else {
new_mm = MemTable.init(next_sst_id, self.allocator, null);
}
}
errdefer new_mm.deinit();
var old_mm: *MemTable = undefined;
{
self.state_lock.lock();
defer self.state_lock.unlock();
var old_mm_ptr = self.state.mem_table;
old_mm = old_mm_ptr.get();
defer old_mm_ptr.release();
self.state.mem_table = try MemTablePtr.create(self.allocator, new_mm);
// 将写满的MemTable压入冷数据
try self.state.imm_mem_tables.append(old_mm_ptr.clone()); // newer memtable is inserted at the end
}
// TIPS：把磁盘同步放在锁的范围外面，降低锁的覆盖
try old_mm.syncWal();
}
pub fn flushNextMemtable(self: *Self) !void {
std.debug.assert(self.state.imm_mem_tables.items.len > 0);
var to_flush_table: *MemTable = undefined;
{
self.state_lock.lockShared();
defer self.state_lock.unlockShared();
// oldest memtable is at the index 0
to_flush_table = self.state.imm_mem_tables.items[0].load();
}
// 将最冷的MemTable构建为SSTable
var builder = try SsTableBuilder.init(self.allocator, self.options.block_size);
defer builder.deinit();
const sst_id = to_flush_table.id;
try to_flush_table.flush(&builder);
const sst_path = try self.pathOfSst(sst_id);
defer self.allocator.free(sst_path);
var sst = try builder.build(sst_id, self.block_cache.clone(), sst_path);
errdefer sst.deinit();
// add the flushed table to l0_sstables
{
self.state_lock.lock();
defer self.state_lock.unlock();
var m = self.state.imm_mem_tables.orderedRemove(0);
defer m.deinit();
std.debug.assert(m.load().id == sst_id);
// newest sstable is at the end
try self.state.l0_sstables.append(sst_id);
try self.state.sstables.put(sst.id, try SsTablePtr.create(self.allocator, sst));
}
}
当然，这里只实现了一半的写逻辑，数据停留在L0的SST中，后续的多层SST还没有使用。
剩下一半的写逻辑会在数据压缩的章节中介绍。
写入的过程比较好理解，但是读就略微复杂了，以上面我们实现的写结果为例子，最终我们的数据沉淀在一个3层的数据结构中，要如何高效的从其中检索数据呢？
如同写过程一般，读过程也是对各个基础单元(MemTable、SSTable、Block)读过程的组合，为了方便组合逻辑，我们要先统一各个模块的读行为。
在LSM-Tree中，所有的读行为都定义为了如下的Interface（Zig中没trait或者Interface，所以这里实例代码我用Rust描述）：
pub trait StorageIterator {
/// Get the current value.
fn value(&self) -> &[u8];
/// Get the current key.
fn key(&self) -> &[u8];
/// Check if the current iterator is empty.
fn is_empty(&self) -> bool;
/// Move to the next position.
fn next(&mut self) -> anyhow::Result<()>;
/// Number of underlying active iterators for this iterator.
fn num_active_iterators(&self) -> usize {
1
}
}
我们首先对MemTable、SSTable、Block这些模块实现读接口，代码可见：src/MemTable.zig，src/block.zig，src/ss_table.zig，这里单独简单介绍下SSTable的读接口实现思路，其他的模块实现思路类似，感兴趣的直接阅读源码即可。
pub const SsTableIterator = struct {
allocator: std.mem.Allocator,
table: SsTablePtr,
blk: BlockPtr,
blk_iterator: BlockIteratorPtr,
blk_idx: usize,
const Self = @This();
pub fn initAndSeekToFirst(allocator: std.mem.Allocator, table: SsTablePtr) !Self {
const s = try seekToFirstInner(allocator, table);
return .{
.allocator = allocator,
.table = table,
.blk_iterator = s.blk_iter,
.blk = s.blk,
.blk_idx = 0,
};
}
pub fn initAndSeekToKey(allocator: std.mem.Allocator, table: SsTablePtr, k: []const u8) !Self {
const b = try seekToKeyInner(allocator, table, k);
return .{
.allocator = allocator,
.table = table,
.blk_iterator = b.blk_iter,
.blk_idx = b.blk_idx,
.blk = b.blk,
};
}
fn seekToFirstInner(allocator: std.mem.Allocator, table: SsTablePtr) !struct {
blk: BlockPtr,
blk_iter: BlockIteratorPtr,
} {
var blk = try table.get().readBlockCached(0, allocator); // 读取第一个block
errdefer blk.release();
var blk_iter = try BlockIterator.createAndSeekToFirst(allocator, blk.clone());
errdefer blk_iter.deinit();
return .{
.blk = blk,
.blk_iter = try BlockIteratorPtr.create(allocator, blk_iter), // 从SSTable的读接口转换为Block的读接口
};
}
fn seekToKeyInner(allocator: std.mem.Allocator, table: SsTablePtr, k: []const u8) !struct {
blk_idx: usize,
blk: BlockPtr,
blk_iter: BlockIteratorPtr,
} {
const table_ptr = table.get();
var blk_idx = try table_ptr.findBlockIndex(k);
var blk = try table_ptr.readBlockCached(blk_idx, allocator);
errdefer blk.deinit();
var blk_iter = try BlockIterator.createAndSeekToKey(allocator, blk.clone(), k);
errdefer blk_iter.deinit();
var blk_iter_ptr = try BlockIteratorPtr.create(allocator, blk_iter);
errdefer blk_iter_ptr.release();
// 如果当前block读完了，跳到下一个block，并生成block的读接口
if (blk_iter.isEmpty()) {
blk_idx += 1;
if (blk_idx < table_ptr.numBlocks()) {
{
blk.deinit();
blk_iter.deinit();
}
var blk2 = try table_ptr.readBlockCached(blk_idx, allocator);
errdefer blk2.deinit();
var blk_iter2 = try BlockIterator.createAndSeekToFirst(allocator, blk2.clone());
errdefer blk_iter2.deinit();
return .{
.blk_idx = blk_idx,
.blk_iter = try BlockIteratorPtr.create(allocator, blk_iter2),
.blk = blk2,
};
}
}
return .{
.blk_idx = blk_idx,
.blk_iter = blk_iter_ptr,
.blk = blk,
};
}
pub fn key(self: Self) []const u8 {
return self.blk_iterator.get().key();
}
pub fn value(self: Self) []const u8 {
return self.blk_iterator.get().value();
}
pub fn isEmpty(self: Self) bool {
return self.blk_iterator.get().isEmpty();
}
pub fn next(self: *Self) !void {
try self.blk_iterator.get().next();
// 若当前的Block读完了，就跳到下一个block，并生成Block读接口。
if (self.blk_iterator.get().isEmpty()) {
self.blk_idx += 1;
if (self.blk_idx < self.table.get().numBlocks()) {
self.reset();
const blk = try self.table.get().readBlockCached(self.blk_idx, self.allocator);
const blk_iter = try BlockIterator.createAndSeekToFirst(self.allocator, blk.clone());
self.blk = blk;
self.blk_iterator = try BlockIteratorPtr.create(self.allocator, blk_iter);
}
}
}
};
有了几个基本元素的读接口之后，我们便遇到第一个问题：我们如何对多个MemTable做读检索？
这个时候，我们需要一个新的数据结构来实现多个读实例的合并检索---- MergeIterator
MergeIterator在内部维护一个二叉堆。堆中数据的优先级如下：
当各个迭代器key不同时，具有最小key的迭代器最优。当多个迭代器有相同的当前key时，最新的迭代器一个最优。
假设我们有如下MemTable（iter1最新，iter3最旧）:
iter1: b->del, c->4, d->5 iter2: a->1, b->2, c->3 iter3: e->4
经过合并后迭代器结果应该为：
a最小，iter2优先迭代 iter2迭代一次后，iter1与iter2 key相同，iter1优先迭代，b->2跳过 c最小，iter1优先迭代，iter2中c->3跳过 d最小，iter1优先迭代 只剩iter3，迭代iter3
最终结果：a->1, b->del, c->4, d->5, e->4
实现代码如下：
// 标准库中有二叉堆实现
const IteratorHeap = std.PriorityQueue(*HeapWrapper, Comparer.Context, Comparer.cmp);
allocator: std.mem.Allocator,
q: IteratorHeap,
current: ?*HeapWrapper,
pub fn init(allocator: std.mem.Allocator, iters: std.ArrayList(StorageIteratorPtr)) !Self {
var q = IteratorHeap.init(allocator, .{});
if (iters.items.len == 0) {
return Self{
.allocator = allocator,
.q = q,
.current = null,
};
}
// PS: the last iter has the highest priority
// 按顺序写入二叉堆
for (iters.items, 0..) |sp, i| {
if (!sp.load().isEmpty()) {
const hw = try allocator.create(HeapWrapper);
errdefer allocator.destroy(hw);
hw.* = HeapWrapper.init(i, sp.clone());
try q.add(hw);
}
}
const cc = q.removeOrNull();
return Self{
.allocator = allocator,
.q = q,
.current = cc,
};
}
pub fn key(self: Self) []const u8 {
return self.current.?.key();
}
pub fn value(self: Self) []const u8 {
return self.current.?.value();
}
pub fn isEmpty(self: Self) bool {
if (self.current) |cc| {
return cc.isEmpty();
}
return true;
}
pub fn next(self: *Self) !void {
const cc = self.current.?;
while (true) {
if (self.q.peek()) |ii| {
std.debug.assert(!ii.isEmpty());
// 如果优先堆头部迭代器A和当前正在生效的迭代器B的key相同，让迭代器A跳过重复key
if (std.mem.eql(u8, cc.key(), ii.key())) {
try ii.next();
if (ii.isEmpty()) {
_ = self.q.remove();
ii.deinit();
self.allocator.destroy(ii);
}
} else {
break;
}
}
break;
}
try cc.next(); // 迭代当前迭代器
// 如果当前优先迭代器迭代完了，就从堆中弹出最优迭代器
if (cc.isEmpty()) {
defer {
cc.deinit();
self.allocator.destroy(cc);
}
if (self.q.removeOrNull()) |h| {
self.current = h;
} else {
self.current = null;
}
return;
}
// 将当前迭代器写回二叉堆，重新计算最优迭代器
try self.q.add(cc);
self.current = self.q.removeOrNull();
}
有了MergeIterator这个工具，我们具备了在多个MemTable和多个SSTable中迭代检索的能力，但是还有个问题，我们当前有两个MergeIterator，应该如何在两个迭代器中执行迭代任务？
此时，我们再引入一个新的数据结构：TwoMergeIterator，这个是MergeIterator在元素只有两个的情况下的简化版。
TwoMergeIterator由两个迭代器构成，一个高优一个低优，每次迭代优先迭代高优，当key相同时，优先迭代高优。实现如下：
pub const TwoMergeIterator = struct {
a: StorageIteratorPtr,
b: StorageIteratorPtr,
choose_a: bool,
// 选择两个迭代器中key更小的迭代器
fn chooseA(a: *StorageIterator, b: *StorageIterator) bool {
if (a.isEmpty()) {
return false;
}
if (b.isEmpty()) {
return true;
}
return std.mem.lessThan(u8, a.key(), b.key());
}
// key相同时，跳过低优中的key
fn skipB(self: *TwoMergeIterator) !void {
const ap = self.a.load();
const bp = self.b.load();
if (!ap.isEmpty() and !bp.isEmpty() and std.mem.eql(u8, ap.key(), bp.key())) try bp.next();
}
pub fn init(a: StorageIteratorPtr, b: StorageIteratorPtr) !TwoMergeIterator {
var iter = TwoMergeIterator{
.a = a,
.b = b,
.choose_a = false,
};
try iter.skipB();
iter.choose_a = chooseA(iter.a.load(), iter.b.load());
return iter;
}
pub fn deinit(self: *TwoMergeIterator) void {
self.a.release();
self.b.release();
}
pub fn key(self: TwoMergeIterator) []const u8 {
if (self.choose_a) {
std.debug.assert(!self.a.load().isEmpty());
return self.a.load().key();
}
std.debug.assert(!self.b.load().isEmpty());
return self.b.load().key();
}
pub fn value(self: TwoMergeIterator) []const u8 {
if (self.choose_a) {
std.debug.assert(!self.a.load().isEmpty());
return self.a.load().value();
}
std.debug.assert(!self.b.load().isEmpty());
return self.b.load().value();
}
pub fn isEmpty(self: TwoMergeIterator) bool {
if (self.choose_a) {
return self.a.load().isEmpty();
}
return self.b.load().isEmpty();
}
pub fn next(self: *TwoMergeIterator) !void {
if (self.choose_a) {
try self.a.load().next();
} else {
try self.b.load().next();
}
try self.skipB();
self.choose_a = chooseA(self.a.load(), self.b.load());
}
};
至此，我们读行为所需要的武器就完备了！
让我们再来看看LSM的架构图：
我们将每个数据层中的数据标上优先级，由于LSM-Tree是append-only的，所以优先级越高的数据层中数据越新。
所以我们的读策略也很明显：按照上图中P0至P2依次检索，这部分代码实现见src/storage.zig。
// search in memtable
if (try self.state.getMemTable().get(key, value)) {
if (value.*.len == 0) {
// tomestone
return false;
}
return true;
}
// search in imm_memtables
self.state_lock.lockShared();
defer self.state_lock.unlockShared();
for (self.state.imm_mem_tables.items) |imm_table| {
if (try imm_table.load().get(key, value)) {
if (value.*.len == 0) {
// tomestone
return false;
}
return true;
}
}
// 收集L0中的迭代器
var l0_iters = std.ArrayList(StorageIteratorPtr).init(self.allocator);
defer {
for (l0_iters.items) |iter| {
var ii = iter;
ii.release();
}
l0_iters.deinit();
}
{
self.state_lock.lockShared();
defer self.state_lock.unlockShared();
for (self.state.l0_sstables.items) |sst_id| {
const sst = self.state.sstables.get(sst_id).?;
if (try sst.load().mayContain(key)) {
var ss_iter = try SsTableIterator.initAndSeekToKey(self.allocator, sst.clone(), key);
errdefer ss_iter.deinit();
try l0_iters.append(try StorageIteratorPtr.create(self.allocator, .{ .ss_table_iter = ss_iter }));
}
}
}
// 收集Levels中的迭代器
var level_iters: std.ArrayList(StorageIteratorPtr) = undefined;
{
self.state_lock.lockShared();
defer self.state_lock.unlockShared();
level_iters = try std.ArrayList(StorageIteratorPtr).initCapacity(
self.allocator,
self.state.levels.items.len,
);
for (self.state.levels.items) |level| {
var level_ssts = try std.ArrayList(SsTablePtr).initCapacity(self.allocator, level.items.len);
errdefer level_ssts.deinit();
for (level.items) |sst_id| {
const sst = self.state.sstables.get(sst_id).?;
if (try mayWithinTable(key, sst)) {
try level_ssts.append(sst.clone());
}
}
if (level_ssts.items.len > 0) {
var level_iter = try SstConcatIterator.initAndSeekToKey(
self.allocator,
level_ssts,
key,
);
errdefer level_iter.deinit();
try level_iters.append(try StorageIteratorPtr.create(self.allocator, .{ .sst_concat_iter = level_iter }));
}
}
}
// 将多个迭代器合并为一个TwoMergeIterator
var l0_merge_iter = try MergeIterators.init(self.allocator, l0_iters);
errdefer l0_merge_iter.deinit();
var levels_merge_iter = try MergeIterators.init(self.allocator, level_iters);
errdefer levels_merge_iter.deinit();
var iter = try TwoMergeIterator.init(
try StorageIteratorPtr.create(self.allocator, .{ .merge_iterators = l0_merge_iter }),
try StorageIteratorPtr.create(self.allocator, .{ .merge_iterators = levels_merge_iter }),
);
defer iter.deinit();
if (iter.isEmpty()) {
return false;
}
if (std.mem.eql(u8, iter.key(), key) and iter.value().len > 0) {
value.* = iter.value();
return true;
}
在上一节的写过程中，我们实现了从内存表到Level0的SSTable堆叠。
随着写入的持续，Lv0的SSTable会越来越多，这个时候就需要我们将Lv0中的数据合并写入至Lv2，并依次类推重复这个过程，直到堆叠到最深的层数，这个逐层合并数据的过程就是数据压缩。
LSM-Tree中数据压缩的过程大致如下：
具体的实现代码可见src/compact.zig，src/storage.zig。
简单分层压缩与原始 LSM 论文中的压缩策略相似。它为 LSM 树维护多个层级。当一个层级太大时，它会将此层级的所有 SST 与下一层合并。压缩策略由 3 个参数控制：
做好这些准备工作，我们可以逐步实现压缩逻辑：
pub const SimpleLeveledCompactionController = struct {
options: SimpleLeveledCompactionOptions,
pub fn generateCompactionTask(self: SimpleLeveledCompactionController, state: *storage.StorageState) !?SimpleLeveledCompactionTask {
if (self.options.max_levels == 1) {
return null;
}
var level_sizes = std.ArrayList(usize).init(state.allocator);
defer level_sizes.deinit();
try level_sizes.append(state.l0_sstables.items.len);
for (state.levels.items) |level| {
try level_sizes.append(level.items.len);
}
// 如果Lv0中SST数量超出阈值，触发L0级别压缩
if (state.l0_sstables.items.len >= self.options.level0_file_num_compaction_trigger) {
std.debug.print("compaction of L0 to L1 because L0 has {d} SSTS >= {d}\n", .{ state.l0_sstables.items.len, self.options.level0_file_num_compaction_trigger });
return .{
.upper_level = null,
.upper_level_sst_ids = try state.l0_sstables.clone(),
.lower_level = 1,
.lower_level_sst_ids = try state.levels.items[0].clone(),
.is_lower_level_bottom = false,
};
}
// 计算Lv[n+1]/lv[n]，如果比例小于阈值，触发Lv[n]级别压缩
for (1..self.options.max_levels) |level| {
const lower_level = level + 1;
if (level_sizes.items[level] == 0) {
continue;
}
const size_ration = level_sizes.items[lower_level] * 100 / level_sizes.items[level];
if (size_ration < self.options.size_ration_percent) {
std.debug.print("compaction of L{d} to L{d} because L{d} size ratio {d} < {d}\n", .{ level, lower_level, level, size_ration, self.options.size_ration_percent });
return .{
.upper_level = level,
.upper_level_sst_ids = try state.levels.items[level - 1].clone(),
.lower_level = lower_level,
.lower_level_sst_ids = try state.levels.items[lower_level - 1].clone(),
.is_lower_level_bottom = lower_level == self.options.max_levels,
};
}
}
return null;
}
}
有了上一小节中读过程的介绍，多层数据的压缩过程就很好理解了。
例如我们想将L1与L2的SSTable合并压缩至L2，我们只需要把L1和L2的数据放在一起创造一个迭代器，再持续从该迭代器中读出数据写入新的SSTable中，这个过程保证了新的SSTable中数据不重复且有序。
fn compactSimple(self: *Self, task: SimpleLeveledCompactionTask) !std.ArrayList(SsTablePtr) {
if (task.upper_level) |_| {
var upper_ssts = try std.ArrayList(SsTablePtr).initCapacity(
self.allocator,
task.upper_level_sst_ids.items.len,
);
var lower_ssts = try std.ArrayList(SsTablePtr).initCapacity(
self.allocator,
task.lower_level_sst_ids.items.len,
);
self.state_lock.lockShared();
for (task.upper_level_sst_ids.items) |sst_id| {
const sst = self.state.sstables.get(sst_id).?;
try upper_ssts.append(sst.clone());
}
for (task.lower_level_sst_ids.items) |sst_id| {
const sst = self.state.sstables.get(sst_id).?;
try lower_ssts.append(sst.clone());
}
self.state_lock.unlockShared();
var upper_iter = try SstConcatIterator.initAndSeekToFirst(self.allocator, upper_ssts);
errdefer upper_iter.deinit();
var lower_iter = try SstConcatIterator.initAndSeekToFirst(self.allocator, lower_ssts);
errdefer lower_iter.deinit();
var iter = try TwoMergeIterator.init(
try StorageIteratorPtr.create(self.allocator, .{ .sst_concat_iter = upper_iter }),
try StorageIteratorPtr.create(self.allocator, .{ .sst_concat_iter = lower_iter }),
);
defer iter.deinit();
return self.compactGenerateSstFromIter(&iter, task.is_lower_level_bottom);
} else {
// compact l0_sstables to l1_sstables
// ..... 代码逻辑大致与上面LvN层压缩一致，只是Lv0层的SSTable是无序的需要特殊考虑
return self.compactGenerateSstFromIter(&iter, task.is_lower_level_bottom);
}
}
fn compactGenerateSstFromIter(self: *Self, iter: *TwoMergeIterator, compact_to_bottom_level: bool) !std.ArrayList(SsTablePtr) {
var builder: SsTableBuilder = try SsTableBuilder.init(self.allocator, self.options.block_size);
defer builder.deinit();
var new_ssts = std.ArrayList(SsTablePtr).init(self.allocator);
// 持续迭代此迭代器
while (!iter.isEmpty()) {
// 如果压缩至最后一层，可以不保留墓碑值key了
if (compact_to_bottom_level) {
if (iter.value().len > 0) {
try builder.add(iter.key(), iter.value());
}
} else {
try builder.add(iter.key(), iter.value());
}
// 当写满一个SSTable后，就清空builder，把写满的SSTable入列
if (builder.estimatedSize() >= self.options.target_sst_size) {
// reset builder
defer builder.reset() catch unreachable;
const sst_id = self.getNextSstId();
const path = try self.pathOfSst(sst_id);
defer self.allocator.free(path);
var sst = try builder.build(sst_id, self.block_cache.clone(), path);
errdefer sst.deinit();
var sst_ptr = try SsTablePtr.create(self.allocator, sst);
errdefer sst_ptr.deinit();
try new_ssts.append(sst_ptr);
}
try iter.next();
}
// 剩余的数据单独一个SSTable
if (builder.estimatedSize() > 0) {
const sst_id = self.getNextSstId();
const path = try self.pathOfSst(sst_id);
defer self.allocator.free(path);
var sst = try builder.build(sst_id, self.block_cache.clone(), path);
errdefer sst.deinit();
var sst_ptr = try SsTablePtr.create(self.allocator, sst);
errdefer sst_ptr.deinit();
try new_ssts.append(sst_ptr);
}
return new_ssts;
}
这部分逻辑并不复杂，即删除此次压缩任务中的原有两层数据，用新合并的SSTable替换至较低层数据。
这里有个需要注意的点，即压缩过程是在一个线程中单独执行的，压缩过程中LSM-Tree的原数据可能发生了改变，所以这里执行SSTable删除时要注意过滤掉新数据，不能覆盖了有效数据。
并发问题是软件中的Bug集散地！
pub fn applyCompactionResult(
_: SimpleLeveledCompactionController,
state: *storage.StorageState,
task: SimpleLeveledCompactionTask,
output: []usize,
) !std.ArrayList(usize) {
var files_to_remove = std.ArrayList(usize).init(state.allocator);
errdefer files_to_remove.deinit();
if (task.upper_level) |upper_level| {
// 删除高层SSTable数据，这层数据不会在压缩过程中变更，放心删
std.debug.assert(sliceEquals(
task.upper_level_sst_ids.items,
state.levels.items[upper_level - 1].items,
));
try files_to_remove.appendSlice(task.upper_level_sst_ids.items);
state.levels.items[upper_level - 1].clearAndFree();
} else {
// 删除L0数据，需要小心
try files_to_remove.appendSlice(task.upper_level_sst_ids.items);
var new_l0_sstables = std.ArrayList(usize).init(state.allocator);
errdefer new_l0_sstables.deinit();
{
var l0_sst_compacted = std.AutoHashMap(usize, struct {}).init(state.allocator);
defer l0_sst_compacted.deinit();
for (task.upper_level_sst_ids.items) |sst_id| {
try l0_sst_compacted.put(sst_id, .{});
}
for (state.l0_sstables.items) |sst_id| {
if (!l0_sst_compacted.remove(sst_id)) { // 不在压缩任务中的SST不能删除
try new_l0_sstables.append(sst_id);
}
}
std.debug.assert(l0_sst_compacted.count() == 0);
}
state.l0_sstables.deinit();
state.l0_sstables = new_l0_sstables;
}
// 低层SSTable数据，直接删除
try files_to_remove.appendSlice(task.lower_level_sst_ids.items);
state.levels.items[task.lower_level - 1].clearAndFree();
try state.levels.items[task.lower_level - 1].appendSlice(output);
return files_to_remove;
}
// sst to remove
var ssts_to_remove = std.ArrayList(SsTablePtr).init(self.allocator);
{
var new_sst_ids = std.ArrayList(usize).init(self.allocator);
defer new_sst_ids.deinit();
self.state_lock.lock();
defer self.state_lock.unlock();
for (sstables.items) |sst| {
const id: usize = @intCast(sst.get().sstId());
try new_sst_ids.append(id);
try self.state.sstables.put(id, sst.clone());
}
var file_to_remove = try self.compaction_controller.applyCompactionResult(
&self.state,
task,
output.items,
);
defer file_to_remove.deinit();
for (file_to_remove.items) |id| {
if (self.state.sstables.fetchRemove(id)) |kv| {
try ssts_to_remove.append(kv.value);
}
}
try self.syncDir();
}
for (ssts_to_remove.items) |sst| {
const path = try self.pathOfSst(sst.get().sstId());
defer self.allocator.free(path);
try std.fs.cwd().deleteFile(path);
}
try self.syncDir();
我们使用Zig语言实现了一个LSM-Tree的核心功能，包括MemTable、SSTable、写流程、各类Iterator与数据压缩能力。通过这个项目，我收获了很多心得体会。
以往对LSM这个数据结构的多层SST设计与写过程早有耳闻，但是读流程的实现不太理解。这个项目解答了我疑惑很久的读流程的实现，特别是MergeIterator的算法设计非常巧妙。
Zig语言没有内存安全的保证，为了不想指针乱飞到处泄露，在Deepseek的帮助下实现了一个简单的智能指针，极大降低了内存管理的心智负担。
文 / 酒米
关注得物技术，每周更新技术干货
要是觉得文章对你有帮助的话，欢迎评论转发点赞～
未经得物技术许可严禁转载，否则依法追究法律责任。
微信关注我们
转载内容版权归作者及来源网站所有！
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。
为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。
Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。
Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。
扫码在手机上查看文章
扫描二维码，手机阅读更方便
有任何问题或合作意向欢迎联系我们
Email: 99873273@qq.com
QQ: 99873273