Summary
在本章中，您将：
要将测试用例复制到启动器代码中并运行它们，
- 实现合并某些SST文件并生成新SST文件的compaction逻辑。
- 实现逻辑以更新LSM状态并管理文件系统上的SST文件。
- 更新LSM读取路径以合并LSM级别。
cargo x copy-test --week 2 --day 1
cargo x scheck
Task 1-Compaction Implementation
在本任务中，您将实现执行合并的核心逻辑——将一组SST文件合并为一个sorted run(一个有序的集合，集合内每个元素唯一)。您需要修改：
src/compact.rs具体来说，就是需要实现
force_full_compaction和compact函数。force_full_compaction是决定要合并哪些文件并更新LSM状态。compact执行实际的合并工作，合并一些SST文件并返回一组新的SST文件。您的compaction实现应该获取存储引擎中的所有SST，通过使用
MergeIterator对它们进行合并，然后使用SST构建器将结果写入到新文件中。如果文件过大，则需要拆分SST文件。compaction完成后，您可以更新LSM状态，以将所有新排序的添加到LSM树的第一级。并且，您需要在LSM树中删除未使用的文件。在您的实现中，您的SST应该只存储在两个地方：L0 SST和L1 SST。也就是说，LSM状态中的层结构应该使用一个vector(Vec<(usize, Vec<usize>)>)保存。在LsmStorageState中，我们将LSM的L1保存在levels字段里面。Compaction不应该阻塞L0刷新，因此在合并文件时不应该使用状态锁。您只应该在更新LSM状态时在合并过程结束时获取状态锁，并在完成状态修改后立即释放锁。
您可以假设用户将确保只有一个合并正在进行。
force_full_compaction在任何时候都只会在一个线程中调用。被放入第1级的SST应该按照它们的第一个键进行排序，并且不应该有重叠的键范围。剧透：Compaction伪代码
fn force_full_compaction(&self) { let ssts_to_compact = { let state = self.state.read(); state.l0_sstables + state.levels[0] }; let new_ssts = self.compact(FullCompactionTask(ssts_to_compact))?; { let state_lock = self.state_lock.lock(); let state = self.state.write(); state.l0_sstables.remove(/* the ones being compacted */); state.levels[0] = new_ssts; // new SSTs added to L1 }; std::fs::remove(ssts_to_compact)?; }在您的Compaction实现中，您现在只需要处理FullCompaction，其中的任务信息包含您将需要Compact的SST。您还需要确保SST的顺序是正确的，以便新的SST中保存的是最新版本的key-value。
因为我们总是合并所有的SST，如果我们发现一个key的多个版本，我们可以简单地保留最新的一个。如果最新版本是删除标记，我们不需要在生成的SST文件中保留它。这不适用于后面几章中的合并策略。
有些事情你可能需要考虑。
- 你的实现是如何处理L0 flush与compaction并行的?(在执行compaction时不使用状态锁，并且还需要考虑在compaction进行时产生的新L0文件。)
- 如果您的实现在合并完成后立即删除原始SST文件，是否会导致系统出现问题?(在macOS/Linux上通常没有，因为在没有文件句柄被持有之前，操作系统不会实际删除文件。)
LSM-Tree的合并策略有很多，这里并非阅读其他资料中提到的size-tiered、leveled，而是简单的只有两层：L0、L1，其中L0为刚从imm_mentable转储而来的，SST之间是无序且存在范围重复。L1保存的是上一次Compaction的结果有序，SST之间不存在范围重复。
FullCompaction则是将所有的l0_sstables(L0)和levels(L1)合并至levels(L1)
测试用例
先看看测试用例week2_day1.rs的test_task1_full_compaction方法：
- 插入0-v1
- 转储
sst - 插入0-v2
- 插入1-v2
- 插入2-v2
- 转储
sst - 删除0
- 删除2
- 转储
sst
在进行合并前，转储了三个sst，新转储的sst都在L0中，合并后的SST文件保存在L1中：
- 插入0-v3
- 插入2-v3
- 转储
sst - 删除1
- 转储
sst
在进行合并前，新转储了两个sst，新转储的sst都在L0中，合并过程需要带上L1中的SST，合并后的SST文件也是保存在L1中：
force_full_compaction
- 获取当前
L0、L1中的SST，构造一个ForceFullCompaction任务 - 进行
compact合并操作 - 移除
sstables存储的历史SST文件 - 清空
l0_sstables、levels(L1) - 将新生成的
sst_id保存至L1 sstables插入新SST
pub fn force_full_compaction(&self) -> Result<()> {
let snapshot = {
let state = self.state.read();
state.clone()
};
// 1、获取当前L0、L1中的SST，构造一个ForceFullCompaction任务
let l0_sstables = snapshot.l0_sstables.clone();
let l1_sstables = snapshot.levels[0].1.clone();
let compaction_task = CompactionTask::ForceFullCompaction {
l0_sstables: l0_sstables.clone(),
l1_sstables: l1_sstables.clone(),
};
println!("force full compaction: {:?}", compaction_task);
// 2、进行compact合并操作
let sstables = self.compact(&compaction_task)?;
{
let mut guard = self.state.write();
let mut snapshot = guard.as_ref().clone();
// 3、移除sstables存储的历史文件
for sst in l0_sstables.iter().chain(l1_sstables.iter()) {
snapshot.sstables.remove(sst);
}
// 4、清空l0_sstables、levels(L1)
snapshot.l0_sstables.clear();
snapshot.levels[0].1.clear();
for sst in sstables {
// 5. 将新生成的sst_id保存至L1
snapshot.levels[0].1.push(sst.sst_id());
// 6、sstables插入新SST
snapshot.sstables.insert(sst.sst_id(), sst);
}
*guard = Arc::new(snapshot)
}
Ok(())
}
compact
使用MergeIterator合并数据，利用SsTableBuilder构建新的SST，在往SsTableBuilder中添加数据后需要判断大小是否超出限制，若超出限制则生成SST。同时判断当前数据的value值是否为空字符串，空字符串代表被删除的记录不需要保存。
// 生成迭代器
let mut iters = Vec::with_capacity(l0_sstables.len() + l1_sstables.len());
for id in l0_sstables.iter().chain(l1_sstables.iter()) {
iters.push(Box::new(SsTableIterator::create_and_seek_to_first(
snapshot.sstables.get(id).unwrap().clone(),
)?));
}
// 使用MergeIterator合并数据
let mut iter = MergeIterator::create(iters);
let mut builder = SsTableBuilder::new(self.options.block_size);
while iter.is_valid() {
let key = iter.key();
let value = iter.value();
// 空字符串代表被删除的记录不需要保存
if !value.is_empty() {
// 利用SsTableBuilder构建新的SST
builder.add(key, value);
}
iter.next().unwrap();
// 超出限制，若超出限制则生成SST
if builder.estimated_size() >= self.options.target_sst_size {
let id = self.next_sst_id();
let sst = builder.build(id, None, self.path_of_sst(id))?;
result.push(Arc::new(sst));
builder = SsTableBuilder::new(self.options.block_size);
}
}
let id = self.next_sst_id();
let sst = builder.build(id, None, self.path_of_sst(id))?;
result.push(Arc::new(sst));
builder = SsTableBuilder::new(self.options.block_size);
Ok(result)
Task 2-Concat Iterator
在此任务中，您需要修改：
src/iterors/concat_iterator.rs现在，您已经在系统中创建了sorted run，可以对读取路径进行简单的优化。您不总是需要为SST创建合并迭代器。如果SST属于一个sorted run，则可以创建一个concat迭代器，它只是按顺序迭代每个SST中的键，因为一个排序运行中的SST不包含重叠的键范围，并且它们按其第一个键进行排序。我们不希望提前创建所有的SST迭代器（因为创建一次迭代器都会导致一次块读取），因此我们只在这个迭代器中存储SST对象。
需求提出的原因是因为SsTableIterator的构造函数create_and_seek_to_first会默认调用一次read_block产生磁盘的IO。因为L1中的SST是排好序的，所以可以通过元信息就能判断所需的key在那个SST。
create_and_seek_to_first
从头开始迭代，只要创建第一个SST的迭代器就行，产生一次IO：
pub fn create_and_seek_to_first(sstables: Vec<Arc<SsTable>>) -> Result<Self> {
if sstables.is_empty() {
return Ok(Self {
current: None,
next_sst_idx: 0,
sstables,
});
}
Ok(Self {
current: Some(SsTableIterator::create_and_seek_to_first(sstables[0].clone()).unwrap()),
next_sst_idx: 1,
sstables: sstables,
})
}
在进行next操作后，还会通过move_until_valid函数，判断当前迭代器是否有效。若无效再创建下一个SST的迭代器。
fn next(&mut self) -> Result<()> {
self.current.as_mut().unwrap().next()?;
self.move_until_valid()?;
Ok(())
}
create_and_seek_to_key
通过first_key元信息，判断在那个SST中。可以精确定位到具体的SST，在找到的这个SST前的SST都没有创建迭代器的操作。
pub fn create_and_seek_to_key(sstables: Vec<Arc<SsTable>>, key: KeySlice) -> Result<Self> {
let idx: usize = sstables
.partition_point(|table| table.first_key().as_key_slice() <= key)
.saturating_sub(1);
if idx >= sstables.len() {
return Ok(Self {
current: None,
next_sst_idx: sstables.len(),
sstables,
});
}
let mut iter = Self {
current: Some(SsTableIterator::create_and_seek_to_key(
sstables[idx].clone(),
key,
)?),
next_sst_idx: idx + 1,
sstables,
};
iter.move_until_valid()?;
Ok(iter)
}
Task 3-Integrate with the Read Path
在此任务中，您需要修改：
src/lsm_iterator.rs src/lsm_storage.rs src/compact.rs现在我们有了LSM树的两级结构，可以更改读取路径以使用新的concat迭代器来优化读取路径。
您需要更改LsmStorageIterator的内部迭代器类型。之后，您可以构造一个合并memtables和L0 SST的两个合并迭代器，以及另一个合并迭代器，将该迭代器与L1 concat迭代器合并。
您还可以更改您的compaction实现以利用concat迭代器。
您需要为concat迭代器实现num_active_iterator，以便测试用例可以测试您的实现是否正在使用concat迭代器，并且它应该始终为1。
要以交互方式测试您的实现，
cargo run --bin mini-lsm-cli-ref -- --compaction none # reference solution cargo run --bin mini-lsm-cli -- --compaction none # your solution然后，
fill 1000 3000 flush fill 1000 3000 flush full_compaction fill 1000 3000 flush full_compaction get 2333 scan 2000 2333
同样的需要，运行需要带上参数--path：
cargo run --bin mini-lsm-cli-ref -- --compaction none --path /tmp/lsm
cargo run --bin mini-lsm-cli -- --compaction none --path /tmp/lsm
get点查
需要使用SstConcatIterator添加L1中SST的读取。
先读取levels[0]中所有的SST，使用SstConcatIterator::create_and_seek_to_key查找。
let mut l1_sst = vec![];
for table in self.state.read().levels[0].1.iter() {
let table = self.state.read().sstables[table].clone();
if table.bloom.is_some()
&& !table
.bloom
.as_ref()
.unwrap()
.may_contain(farmhash::fingerprint32(_key))
{
continue;
}
l1_sst.push(table.clone());
}
let l1_iter = SstConcatIterator::create_and_seek_to_key(l1_sst,KeySlice::from_slice(_key))?;
if l1_iter.is_valid() && l1_iter.key().raw_ref() == _key {
if l1_iter.value().is_empty() {
return Ok(None);
}
return Ok(Some(Bytes::copy_from_slice(l1_iter.value())));
}
scan范围查找
先修改LsmIteratorInner的类型：
type LsmIteratorInner = TwoMergeIterator<
TwoMergeIterator<MergeIterator<MemTableIterator>, MergeIterator<SsTableIterator>>,
SstConcatIterator,
>;
再修改scan函数，先读取levels[0]中所有的SST，使用SstConcatIterator::create_and_seek_to_first创建l1的迭代器。
let mut l1_sst = Vec::with_capacity(snapshot.levels[0].1.len());
for &sst_id in snapshot.levels[0].1.iter() {
l1_sst.push(snapshot.sstables[&sst_id].clone());
}
let l1_iter = SstConcatIterator::create_and_seek_to_first(l1_sst)?;
let iter = TwoMergeIterator::create(merge_memtable_iter, l0_iter)?;
let iter = TwoMergeIterator::create(iter, l1_iter)?;