2021 CMU 15-445 实验笔记

陆陆续续终于把 CMU 15-445 刷完了（中间插了个 TinyKV），这也算是自己数据库的启蒙之课。编码耗时共计 98 小时 43 分钟。

我个人给整个项目难度评级：Project 1 < Project 4 < Project 3 << Project 2。其中 Project 2 难度最大，主要没啥参考资料，今年的是 Extendible Hash Table，不是噩梦 B+ 树（其实也挺噩梦的）。

我个人认为 15-445 并不是代码通过 Gradescope 就算可以了，很多东西即使你做完了还是模模糊糊的，强烈建议跟着 PPT 和《数据库系统概念 第七版》过一遍，着重看 Query Processing，Transaction 和 Concurrency Control，其中事务这块更是重中之重。

这里可以看看我自己总结的事务并发控制：https://www.inlighting.org/archives/database-concurrency-control/ 。

如果想直接要答案源码的，发我邮件咨询就行了。

Project 0

Project 0 相当于一个热身项目，用于检查学员是否具备正常的 C++ 能力来进行这一门课程。

因为我学习这门课程前不会 C++，所以我没能力，因此我没做。。。

Project 1

Project 1 要我们实现一个 buffer pool，实验分为三个部分，我逐步说明。

LRU Replacement Policy

这个实验一开始主要是被方法名搞懵了，实际上其方法名是对应上层 BufferPool 来说的。LRU 管理的是 frame，存放 page 的那个 frame，而不是 page 本身。比如上层 BufferPool Pin() 了一个 page，然后上层找到该 page 的 frame，然后 LRU 需要移除这个 frame，不进行淘汰（因为上层在使用中）。反之，如果上层 BufferPool UnPin() 了一个 page，然后就要把该 page 对应的 frame 加入 LRU，等待被移除。

此外每个方法注意加锁，可以使用 std::lock_guard<std::mutex> 来进行处理，类似 go 语言的 defer ，可以优雅的解决锁释放的问题。

Buffer Pool Manager Instance

具体流程我不讲，大家自己琢磨琢磨就知道了，我就说说我几个犯了错误的地方。

NewPgImp(page_id_t *page_id) 中，不要一开始就调用 AllocatePage() 分配 pageId，只有当真的有空闲的 page 可以使用时，再调用 AllocatePage()分配一个 pageId 给它，不然你会过不去 gradescope 上面的 [RoundRobinNewPage] 这个测试点。至于为啥，你看看 AllocatePage() 的实现就知道了。

每一次获得一个新的 page，或者删除一个 page 时，请调用 page->ResetMemory() 方法将其数据重置掉，而不是放任不管，想着后面可以直接覆盖。

UnpinPgImp(page_id_t page_id, bool is_dirty) 时不要直接 page->is_dirty_ = is_dirty ，相反应该是：

if (is_dirty) {
  page->is_dirty_ = is_dirty; // 不然会直接把之前的 is_dirty 状态给覆盖了。
}

最后注意加锁！

Parallel Buffer Pool Manager

我在 Parallel Buffer Pool Manager 中维护了一个 next_instance_ 变量，用于判断下一次分配 page 的 Buffer Pool Manager 是谁，分配 page 的 round-robin 代码如下：

Page *ParallelBufferPoolManager::NewPgImp(page_id_t *page_id) {
  std::lock_guard<std::mutex> guard(latch_);
  for (size_t i = 0; i < num_instances_; i++) {
    BufferPoolManager *manager = *(managers_ + next_instance_);
    Page *page = manager->NewPage(page_id);
    next_instance_ = (next_instance_ + 1) % num_instances_;
    if (page != nullptr) {
      return page;
    }
  }
  return nullptr;
}

注意，这里只有 NewPgImp(page_id_t *page_id) 方法需要加锁，别的地方加锁没必要，不然还要 parallel 干啥。

Project 2

Project 2 是让我们实现一个 Extendible Hash Table，只能说很难，难度系数是 Project 1 的两倍，中间一度有点想放弃（主要网上还没别人的代码参考）。整个项目大约花了 10 天吧。

关于 Extendible Hash Table 的算法实现，可以看我的另一篇文章：https://www.inlighting.org/archives/extendible-hash-table-algorithm ，这里我说说我遇到的一些坑。

Bucket

先从 bucket 开始说起，首先就是 IsReadable() 和 IsOccupied() 两个函数。在这里，如果一个元素被创建了，那么他的 readable_ 和 occupied_ 均要被标记。如果被删除了，你只需要将 readable_ 的标记清除即可，occupied_ 不用管，仍然占用。

Bucket 标记元素是否被占用的是 char 数组，一个 char 是 8 bit，能表示 8 个数据，设置 readable 和 occupied 时位运算是肯定跑不了了。

关于插入和查询操作，你直接遍历查找就行了，是的，你没有听错，就是一个一个遍历。一个 bucket 只占一个 page 的大小，4 KB 的空间你也玩不出什么数据结构。虽然常规下，Extendible Hash Table 的 bucket 应该使用前缀树，但是它太占空间了。

请不要在 bucket page 中定义额外的成员变量：一开始我想为了提升性能，在 bucket 中定义了一个 NumReadable 变量，用于统计当前 bucket 有几个可读的元素，这样判断 IsFull 和 IsEmpty 可以不需要遍历。但是实际上官方给你定义的数据结构有时候会正好占满 4096 KB，如果你自己定义了某个成员变量，会使得这个 bucket 超出范围了，然后你会越界访问到 Page 里面的内容，然后就莫名其妙的报错。我被这个问题卡了很久，不然早过了。

Directory

这块其实没啥好说的，你自己实现好 Hash Table 的 grow 和 shrink 的逻辑即可。锁也不用加。

Hash Table

Hash Table 这块锁的设计就有讲究，我个人建议的是，先不加锁实现，等能过基本的 Insert，Remove 测试点时，再加锁。加锁直接用全局的 table_latch_ 加写锁，先保证测试用例都能过了，100 分了，再考虑优化性能。我一开始全局写锁，gradescope 是 100 分了，不过 leaderboard 那里没有分数。

这里讲讲我优化后的锁设计：

Insert() 时，table_latch_ 是 ReadLock，对应的 bucket 为 WriteLock。这很好理解，因为你只对一个 bucket 就行修改操作。

SplitInsert() 时，因为一个 bucket 容量不够，你需要扩容，这里会涉及到 directory 的操作，所以这里我使用 table_latch_ 的 WriteLock，锁住全局。同理，合并 directory 的操作也需要 table_latch_ 的 WriteLock 锁住全局。

GetValue() 操作不用说，table_latch_ 和 bucket lock 均使用 ReadLock。

FetchDirectoryPage() 这块我使用了一个独立的锁，因为我在这个方法里面涉及到创建 directory 的逻辑，就是当 HashTable 刚被创建的时候需要一个初始的 directory 是一个 local depth 为 0 的 bucket。当然你也可以不用这么麻烦，直接在 Hash Table 的构造方法里面创建就行了。

注意事项

及时的 Unpin 不需要的 page，我就这么说吧，gradescope 中有些测试用例的 buffer pool size 只有 3，也就是 Hash Table 运行最小需要的 page 数量。（1 个给 directory，2 个给 bucket，因为 bucket 分裂的时候需要 2 个）。

善用 assert 语句，比如 Unpin 等操作时通过 assert 确定其是成功执行的。还有一些地方通过 assert 来确定数据是按照你的想法在执行。这样能帮助你更快的定位出程序的问题。

比如下面这段程序：

uint32_t mask = dir_page->GetLocalDepthMask(split_bucket_index);
for (uint32_t i = 0; i < origin_array_size; i++) {
  MappingType tmp = origin_array[i];
  uint32_t target_bucket_index = Hash(tmp.first) & mask;
  page_id_t target_bucket_index_page = dir_page->GetBucketPageId(target_bucket_index);
  assert(target_bucket_index_page == split_bucket_page_id || target_bucket_index_page == split_image_bucket_page_id);
  if (target_bucket_index_page == split_bucket_page_id) {
    assert(split_bucket->Insert(tmp.first, tmp.second, comparator_));
  } else {
    assert(split_image_bucket->Insert(tmp.first, tmp.second, comparator_));
  }
}

当一个 bucket 分裂后，我们需要将这个 bucket 中原有的数据分流。按照 split 逻辑我们肯定知道，分流的数据必定落在原来的 bucket page 和 split image bucket page 两个 bucket 中（注意是 page 哦，而不是 bucket 的 index）。这里我们可以使用 assert 进行确认，提前定位 bug。

Project 3

Project 3 中我们需要基于火山模型（Volcano model）实现基本的 SQL 语句，没啥难的，无非就是一些 API 不知道，要花点时间看源码。

常用代码：

根据 SELECT 的字段生成对应 tuple：

std::vector<Value> values;
for (size_t i = 0; i < plan_->OutputSchema()->GetColumnCount(); i++) {
  values.push_back(plan_->OutputSchema()->GetColumn(i).GetExpr()->Evaluate(tuple, schema_));
}

*tuple = Tuple(values, plan_->OutputSchema());

判断 tuple 是否满足 predicate 条件：

const AbstractExpression *predict = plan_->GetPredicate();
if (predict != nullptr && !predict->Evaluate(tuple, plan_->OutputSchema()).GetAs<bool>()) {
  // Satisfy predicate
}

如果存在 child executor，需要先 init 它：

void InsertExecutor::Init() {
  // ...
  child_executor_->Init();
  // ...
}

插入索引：

删除索引类似。

for (const auto &index : catalog_->GetTableIndexes(table_info_->name_)) {
  index->index_->InsertEntry(
    tuple->KeyFromTuple(table_info_->schema_, *index->index_->GetKeySchema(), index->index_->GetKeyAttrs()), *rid,
    exec_ctx_->GetTransaction());
}

具体实现

Sequential Scan

数据通过 TableHeap 的 Next() 获取，根据 SELECT 的字段生成对应 tuple。如果存在 predicate 条件则额外进行判断是否满足。

Insert

调用 TableHeap 的 InsertTuple() 方法，插入成功后需插入对应的索引。

Update

删除原来的索引，调用 GenerateUpdatedTuple() 生成新的 tuple，通过 TableHeap 的 UpdateTuple() 更新原有 tuple，最后再插入 新索引。

Delete

调用 TableHeap 的 MarkDelete() 删除对应的 tuple，再删除索引即可。

Nested Loop Join

没啥复杂的，主要是判断 join 的 predicate 条件的 API 复杂，示例代码如下：

if (plan_->Predicate() == nullptr || plan_->Predicate()
    ->EvaluateJoin(&left_tuple, left_executor_->GetOutputSchema(),
                   &right_tuple, right_executor_->GetOutputSchema())
    .GetAs<bool>()) {
  std::vector<Value> output;
  for (const auto &col : GetOutputSchema()->GetColumns()) {
    output.push_back(col.GetExpr()->EvaluateJoin(&left_tuple, left_executor_->GetOutputSchema(), &right_tuple,
                                                 right_executor_->GetOutputSchema()));
  }
  tmp_results_.push(Tuple(output, GetOutputSchema()));
}

Hash Join

Hash Join 需要自己仿照 SimpleAggregationHashTable 自己写一个 Hash Table，底层直接用 std::unorder_map 就行，不需要使用 Extendible Hash Table。

在 Init 时先把所有 left_child 的 tuple 插入 hash table，之后在 Next() 时每次匹配一个 right tuple 即可。

Aggregation

和 Hash Join 类似，就是输出时需要判断是否存在 having 条件，如果存在，判断是否满足。

// 判断Having条件，符合返回，不符合则继续查找
if (plan_->GetHaving() == nullptr ||
    plan_->GetHaving()->EvaluateAggregate(agg_key.group_bys_, agg_value.aggregates_).GetAs<bool>()) {
  std::vector<Value> ret;
  for (const auto &col : plan_->OutputSchema()->GetColumns()) {
    ret.push_back(col.GetExpr()->EvaluateAggregate(agg_key.group_bys_, agg_value.aggregates_));
  }
  *tuple = Tuple(ret, plan_->OutputSchema());
  return true;
}
return Next(tuple, rid);

Limit

太简单，直接贴出来得了。

void LimitExecutor::Init() {
  numbers_ = 0;
  child_executor_->Init();
}

bool LimitExecutor::Next(Tuple *tuple, RID *rid) {
  if (!child_executor_->Next(tuple, rid)) {
    return false;
  }

  numbers_++;
  return numbers_ <= plan_->GetLimit();
}

Distinct

一样需要和 Hash Join 一样实现一个自己的 hash table，然后通过 hash 表去重即可。

Project 4

事务的并发控制，建议过完实验后，看一遍书和 PPT，再回来看代码，会有更加深刻的理解。不然你有可能只是面向测试用例编程。

LockManager

这里直接和 deadlock prevention 一起讲了。

2PL 下不同隔离级别的行为：

Read uncommitted：读取不需要获得 shared lock，写需要获得 exclusive lock，用完直接放锁，不需要遵守 2PL 的两个 phase 规则。

Read committed：读取和写入均需要锁，用完直接放锁，不需要遵守 2PL 的两个 phase 规则。

Repeatable read：读取和写入均需要锁，需要遵守 2PL 的两个 phase 规则，只有在事务 commit 或 abort 时统一放锁。

等待获取锁细节：

在 LockRequestQueue 中排队获取锁时，我采用的设计是事务从小到大排列（older->younger）。假设一个事务 $T$ 申请锁后会将其锁追加在 RequestQueue 末尾，然后遍历整个 RequestQueue，如果存在大于事务 $T$ （也就是 younger）且会冲突的锁，则 abort 掉拥有该锁的事务。如果在遍历 RequestQueue 的过程中发生过 abort 行为，遍历完成后就 notify_all() 一次，尝试唤醒阻塞线程。

等待获取锁请使用 while 循环，而不是 if。

while (NeedWait(txn, lock_queue)) {
  lock_queue->cv_.wait(guard);
  if (CheckAbort(txn)) {
    return false;
  }
}

LockShared

1. 如果已经 abort，直接 return false。
2. 如果 IsolationLevel 是 READ_UNCOMMITTED，直接 abort，它不需要读锁。
3. 如果不是处于 2PL 的 GROWING 阶段，直接 abort。
4. 如果已经获取过 shared lock，直接 return true。
5. 添加锁到 RequestQueue 和 txn 的 SharedLockSet 中，之后尝试等待获取锁。
6. 获得锁成功后，将锁的 granted_ 设置为 true。

LockExclusive

和 LockShared 差不多，就是锁冲突的形式不一样，Exclusive 和任意锁都是冲突的。

LockUpgrade

同样和 LockShared 差不多，下面说几个不同点：

1. 如果已经有了对应 rid 的 exclusive lock，说明之前可能已经 upgrade 成功，直接 return true。
2. 只有当比你 older 的 txn 不含有 exclusive lock 时，你才可以 upgrade 你的 shared lock。
3. 获得锁后，修改锁的信息，更改事务的 LockSet。

虽然 LockRequestQueue 提供了一个 upgrading_ 属性，不过我并没用到过。

Unlock

1. 如果不含对应的锁，直接 return false。
2. 如果当前事务隔离级别是 REPETABLE_READ，且处于 2PL 的 GROWING 阶段，将 2PL 设置为 SHRINKING 阶段。
3. 移除事务的 LockSet 中对应的锁。

Execution

seq_scan_executor

如果不是 READ_UNCOMMITTED，读取均需要获取 shared lock。如果是 READ_COMMITTED，读完后需要立刻释放 shared lock。

insert_executor

任意隔离级别均需要获取 exclusive lock（如果本来有 shared lock，则通过 upgrade 升级得到）。READ_UNCOMMITTED 和 READ_COMMITTED 写入完成后立刻释放 exclusive lock。REPEATABLE_READ 会在整个事务 commit 时统一 unlock，不需要我们自己编写代码。

update_executor

同上

delete_executor

同上

总结

受益匪浅，就是不知道大脑能记多久，感谢 CMU！。