本文为CMU15-445(2021Fall)的lab记录。

推荐博客： https://blog.csdn.net/twentyonepilots/article/details/120868216，逻辑写得比较清楚

CMU-15445官方网页 https://15445.courses.cs.cmu.edu/fall2021/assignments.html

Project #1 - Buffer Pool

TASK #1 - LRU剔除策略

可以先把146. LRU 缓存做一做，涉及的底层数据结构是一样的，同样都使用了链表和哈希表，逻辑上可能还比这个lab难一点

lab的逻辑没有课上讲的复杂，可以不使用timestamp

主要完成的是lru_repalcer.cpp这个文件，主要数据结构：

std::list<frame_id_t> list_; // 存储空闲的frame_id
std::unordered_map<frame_id_t, std::list<frame_id_t>::iterator> map_; // 用来加速查找的过程

lru replacer 管理buffer pool中的空闲frame_id，这些frame_id刚入list中管理，前面是新加入的，后面则是旧的。当需要牺牲一个页面时，从旧的那一端取。

lru replacer管理的是空闲frame的id?

是的，可以往下看，lrt replacer是与buffer pool manager类一起使用的，buffer pool manager被pin住的frame是不能被刷盘的，而那些没有被pin的frame就交给lru replacer管理，这些没有被pin的frame可以被刷盘然后存放新的磁盘数据。由于磁盘速度与内存速度相差过大，因此我们的目标是尽可能将数据多保存在内存一会，但是内存的容量是有限的，必须对保存的内容进行选择，尽可能将热点数据保存在内存中。而lru 剔除策略就很好地满足了这个要求，当buffer pool的容量不够时，将最近最少使用的prame剔除即可。

TASK #2 - Buffer Pool Management Instance

这个project的难点可能是对page_id 、frame_id 和page_table的理解。

首先page_id是对磁盘上的页的编号，frame_id 是对buffer_pool中页面的编号，由于buffer_pool需要从磁盘上不断取出、刷新页面，所以buffer_pool中的页面对应的磁盘上的具体页面是时刻变化的，所以我们需要page_table 映射这两者的关系。

  /** Page table for keeping track of buffer pool pages. */
  std::unordered_map<page_id_t, frame_id_t> page_table_;

上图是一个关于page在内存和磁盘中如何被管理的逻辑视图。

bufferpool在初始化时会在内存中开辟一块内存，用来存放从磁盘上取得的page，图示的pool_size假定为4，似乎很小，但是很多测试用例就是这个大小，所以之后的两个实验中不要忘记 unpin 操作！
```
 pages_ = new Page[pool_size_];  // 这里就是bufferpool中存放数据页的地方
```

使用disk_manager将磁盘上指定的page取出来并存放到 pages_数组对应的位置,并修改page_table

....
// 比如将磁盘上page_id  = 2的那个页读取并存储到内存page_数组中，其frame_id = 1
disk_manager_->ReadPage(page_id = 2, pages_[1]->GetData());
...
page_table[1] = 2; // 然后修改page_table的映射关系即可

lru_repalcer 只管理没有被pin住的页面的frame_id

关于LRU的优化

看面经有很多关于LRU的优化问题，所以我去看了一些资料然后整理整理。

参考资料：

《Mysql是怎样运行的》
https://xiaolincoding.com/os/3_memory/cache_lru.html#如何避免预读失效造成的影响
CMU15445 lecture 5

传统LRU的问题

Sequential flooding : 如果数据库执行一个顺序扫描操作，会把读到的所有页都加入lru的新端而将其他页面victim掉，而这些被牺牲的页却可能是热点数据，在lru链表中“最近使用的页”可能是我们最不需要的页面。（想想我们执行顺序扫描，将满足条件的记录读出来，那些不满足条件的页面虽然在内存中但是已经没有用了）

MySQL的优化方案

mysql将lru链表分为 young和old两个区域，young区域存储热数据(使用频率高)，old区域存储冷数据，默认的比例是old区域占全链表的3/8.

假设我们执行sql语句： select colA from tableA where colA = something

首先扫描进内存的页被放入old端，表示这是冷数据，当我们对old区域的数据再次访问的频率达到一定程度时，会将它加入 young 区域的头部。那么从old转入young的门槛是什么？

首先考虑访问次数，如果我们将访问old区域页面第二次的页面加入young区域怎么样？答案是不可行，因为mysql在读取一条记录是，就算是访问了一次页面。显然一个页面的记录有很多，如果我们将访问的次数作为门槛，还是达不到保持young区域为热点数据的目的。mysql的做法是：在对于某个处于old区域的页面进行第一次扫描时，记录这个访问时间，如果后续访问时间与第一次访问的时间在某个时间间隔内，就不会将它加入young的头部。这个时间间隔默认为1s。当然如果没有对处于old区域的页面进行后续访问，该页面同样不同移动入young区域。

Project #2 - Extendible Hash Index

很难很多坑，难度是Project1的三倍不止。前期和gradescope斗智斗勇差点放弃，幸好有大佬搞来了测试代码下来能够在本地测试，否则给我两个月都难写出来。

Extendible Hash Index 原理

首先要搞明白的就是它的工作原理，否则寸步难行。

参考资料：

上面两个资料已经将基本的东西讲得很清楚了，先看第一个搞清楚术语。

还有几个点可能要略作补充：

资料中 directory 会保存一个指针指向bucket，但这只是逻辑上的概念，帮助我们理解它的概念。阅读lab的源码，可以知道direcotry保存的是各个bucket_page的page_id而不是bucket_page的指针, 必须使用buffer_pool_manager获取在磁盘\内存中的对应页面。
无论是bucket的分裂或者合并都需要用到GetSpiltImageIndex()方法, 这个方法的不同实现可能导致之后hashtable实现的逻辑不同。贴一下我的实现：
```
uint32_t HashTableDirectoryPage::GetSplitImageIndex(uint32_t bucket_idx) {
  uint32_t mid = (1 << global_depth_) / 2;
  if (bucket_idx < mid) {
    return mid + bucket_idx;
  }
  return bucket_idx - mid;
}
```
思路：逻辑上将整个bucket 分为两半，先判断bucket_idx在哪一半，然后返回另一半的对应偏移的值就是镜像index。对照下图看，无论是 ld = 1还是 ld = 2的bucket，应该都是没有问题的：
merge的逻辑可能不是很清楚，这个先看看官网的提示，然后又参考了其他的博客。首先注意官方代码注释提醒了不要递归地合并(分裂的逻辑却是要递归的)；触发merge的条件是当delete一个key后，如果对应的bucket为空则进行merge，在merge好一个bucekt后，应该扫描directory中的所有bucket看看是否能够再次合并空bucket。下面图解这种情况：

某一时刻，我们有下面的逻辑示意图, 即 page_id 为5 和8 的page为空，其他页都不为空。

下一时刻，上层调用remove删去page_id为7的最后一条记录，7为空页，此时发生merge，删除页面7，调整对应指针到splitimage，即目录中0号元素所指的bucket。

此时满足diretory shrink的条件：

shrink后，进行额外的可能的merge操作 扫描整个directory发现page5也为空，故与page8合并

等到最后全部删除时：

应该满足 gd <= 1条件, 如果没有进行额外的merge操作，最后会违反这个条件导致不通过

TASK #1 - Page Layouts

HASH TABLE DIRECTORY PAGE

先实现目录页，它主要有4个成员属性

page_id ：目录页的pageid
global_depth : 整个hash表的全局深度
local_depth: 这是一个长度为512的char数组，记录每个bucket的局部深度
bucket_page_id : 这是一个长度为512的int32数组，记录对应的bucket页的page_id, 也就是图示中看到的一个个箭头

512是hash表bucket页面数量的最大值，可能不会用到这么多，我们可以用 global_depth算出hash表的实际bucket数量 = 2 ^ global_depth

其他没啥好注意的，上面说的 GetSplitImage 方法想明白了就行

HASH TABLE BUCKET PAGE

这是存放键值对的页面。而且我们要支持存储相同的key不同value的键值对,这对之后的修改逻辑有很大影响。

三个主要成员属性，

char occupied_[(BUCKET_ARRAY_SIZE - 1) / 8 + 1]; // 元素被插入时设置为1，被移除时不会设置为0， 可以用来加速某些方法的执行
char readable_[(BUCKET_ARRAY_SIZE - 1) / 8 + 1]; // 元素被插入和移除时都做相应修改
MappingType array_[0];// Do not add any members below array_, as they will overlap.

使用位图管理bucket页的键值对的占用，occupied 数组表示某个位置是否被占用过（曾今占用现在被移除，或者现在还在占用中），readale 数组表示某个位置现在是否存在有效数据。

Insert ：扫描readable数组，取出可读数据比较键值对和将要插入的键值对是否都相同，如果相同说明插入失败。可以使用occupied数组加速扫描过程，因为当一个位置没有occupied时，那么说明它之后的所有位置都不会被占用，过去没有被占用，现在也没有被占用。找到位置后则设置相应的readable 、和occupied的位置为1，最后执行插入。

必须扫描完整的readable数组得到插入的位置。本人之前以为readable数组和occupied数组一样是"连续的"，所以当扫描到一个readable数组的为为0时，就以为整个bucket的键值对已经扫描完了就执行了插入。但其实readable数组是不连续的因为有移除操作，比如此时readable数组是这样的 [1 1 1 1 0 0], 我们移除一个元素 [1 0 1 1 0], 我们再插入一个元素发现第二个位置有空位，但此时不能断定插入成功，我们需要扫描后续的所有可读数据进行判重, 如果不重复则说明插入成功，最后我们再次得到[1 1 1 1 0]
Remove : 没啥好说，扫描readable数组比较键值对是否时要删除的，然后只需要设置 readable数组对应位位0即可，不修改occupied数组。同样可以用occupied数组加速这个过程。
Search: 简单，同样可用occupied数组加速

整个页面存储多少的键值对是通过宏运算确定好的。

(BUCKET_ARRAY_SIZE - 1) / 8 + 1 这个计算公式已经根据 Key 和 Value的实际类型计算出了一个bucket页面最多存储多少个键值对，是一个不会造成内存越界的最大值。所以我们不能在头文件中添加其他的类成员属性 , 因为这可能造成数据的 “overlap”

TASK #2 - 实现hash表

这应该是整个 Lab 中最烦最劝退的任务。

我们要实现 extendible_hash_table.cpp和相应头文件的代码，三个主要的public 方法为， search、 insert 、remove

难点在两个个private 方法： SplitInsert---可能由insert触发（bucket满）、 Merge---可能由remove触发（bucket空）

search

search(getvalue)的逻辑很简单就是先计算key的hash值，根据global_depth计算到一个bucket_index, 也就是说key被散列到了这个bucekt_index对应的bucket页面中，我们已经在第一个task中已经实现了如何在单个bucket页中查找key，我们使用对应方法在这个bucket中查找即可。麻烦的是下面insert 和 merge的逻辑。

insert

如果一个Bucket的空位数目足够，那么insert的逻辑和search几乎一样简单，但是一个bucket满时，我们就要进行页分裂。对应单独的一个方法 SplitInsert，我们可以在 insert 逻辑中判断插入是否成功，如果因为bucket满而插入不成功(也有可能时key-value重复而不成功)，则调用SplitInsert进行分裂插入。

SplitInsert是一个递归方法，老师有对该函数写了下面得注释 --- Performs insertion with an optional bucket splitting. If the page is still full after the split, then recursively split. This is exceedingly rare, but possible. 函数声明如下

bool HASH_TABLE_TYPE::SplitInsert(Transaction *transaction, const KeyType &key, const ValueType &value)

大致的逻辑为：

hash 传入参数key，计算这个键应该被hash到哪一个bucket。然后判断这个bucket是否满，如果不满（这就是递归终止条件）则对bucket执行正常插入然后直接返回。
bucket已满则要分裂bucket。根据该buckut页面的local_depth 是否等于 global_depth有两种做法
- 如果local_depth == global_depth, 那么目录页面的容量得首先增加一倍，然后分裂bucket， "分裂"指的是用bmp新建一个Page当作Bucket，将它的page_id放入目录页中保存bucket pageId的数组中，当然得是在对应的bucketIdx中
- 如果local_depth < global_depth, 那么仅分裂bucket即可

bucket分裂后应该更新目录页中对应的local_depths 、 page_ids 、 global_depth，使它们可以通过完整性检查：

/**
   * VerifyIntegrity
   *
   * Verify the following invariants:
   * (1) All LD <= GD.
   * (2) Each bucket has precisely 2^(GD - LD) pointers pointing to it.
   * (3) The LD is the same at each index with the same bucket_page_id
   */

分裂bucket后，将原页面的键值对rehash到两个bucket中，这两个bucket互为镜像。
- rehash过程先检测每个key是否hash到原bucket中，如果是就不用动这个key-value对；如果不是则在这个原bucket中删除这个对，然后将这个键值对添加到image bucket中
完成分裂后，递归调用 splitinsert , 而且在这之前不要忘记unpin

伪代码如下：

bool HASH_TABLE_TYPE::Insert(Transaction *transaction, const KeyType &key, const ValueType &value) {
    bucket = hash_key_to_bucket(key);
    insert_succeed = bucket->insert(key, value);
    if (!insert_succeed && bucket is full) {
        insert_succeed = SplitInsert(key,value);
    }
    return insert_succeed;
}

bool HASH_TABLE_TYPE::SplitInsert(Transaction *transaction, const KeyType &key, const ValueType &value) {
    bucket = hash_key_to_bucket(key);
    // recursion termination condition
    if (bucket is not full) {
        // just apply insert in the bucket 
        bucket.insert(key, value);
        return true;
    }
    
    // bucket is full , we need split it
    // gd : global_depth of directory; 
    // ld : local_depth of the bucket
    if (gd == ld) {
        // if gd == ld ，expand directory
        ExpensionDirectory();
    }
    // if gd > ld, only need to split bucket
    image_bucket = buffer_pool_manager->NewPage();
    ajdust ld 、 gd、 page_id to verify invariants
	RehashKeyValue(bucket, image_bucket);
    // must upin pages, otherwise there is no space for bmp to get pages from disk or create pages
    upin pages just used
    // recursive call
    return SplitInsert(key, value)
}

remove

和search是差不多的逻辑，在对某个bucket成功remove后，判断这个bucket是否为空，如果为空则调用merge。由于merge不是递归函数，但是有可能还会有额外的合并，所以我们定义一个ExtraMerge方法进行额外地合并。官方注释提示我们可以在以下几个条件满足时可以跳过合并的步骤，：

  * There are three conditions under which we skip the merge:
   * 1. The bucket is no longer empty.
   * 2. The bucket has local depth 0.
   * 3. The bucket's local depth doesn't match its split image's local depth.
   * Note: we do not merge recursively.

直接上伪代码吧：

bool HASH_TABLE_TYPE::Remove(Transaction *transaction, const KeyType &key, const ValueType &value) {
    bucket = hash_key_to_bucket(key);
    remove_succeed = bucket->remove(key, value);
    if (remove_succeed && bucket is empty) {
        Merge(key,value);
        ExtarMerge();
    }
}

void HASH_TABLE_TYPE::Merge(Transaction *transaction, const KeyType &key, const ValueType &value) {
    empty_bucket = hash_key_to_bucket_using_bmp(key);
    image_bucket = GetImageBucket(empty_bucket);
    // judge if we actually need merge 
    if ( ld_of_empty_bucket > 0 && ld_of_empty_bucket == ld_of_image_bucket &&
  		bucket_empty_page_id != bucket_image_page_id &&
      	empty_bucket->IsEmpty()) {
        // don't forget to unpin the empty page, otherwise we cannot delete this page due to the restrction of bmp
        bmp_unpin(empty_bucket);
        bmp_delete(empty_bucket);
        
        ajdust ld 、 gd、 page_id to verify invariants;
        
        if (all lds < gd) {
            ShrinkDirectory(); // we can just decrement gd by 1, since the actual size of deretory is calculated by gd
        }
    }
    unpin all pages   
}

void HASH_TABLE_TYPE::ExtraMerge() {
    for (every bucket in deretory) {
        if (bucket is empty) {
            just apply the merge process like  Merge() does
        }
    }
    unpin_pages
}

TASK #3 - Concurrency Control

这一个任务是让我们保证哈希表的线程安全性。在extendible_hash_table.h 中已经定义了一个表锁：

// extendible_hash_table.h 
...
ReaderWriterLatch table_latch_;

这个锁用来对整张表进行加锁。那么为了提高并发度，我们需要降低锁的粒度，还有一个在 Page.h中

// Page.h
...
ReaderWriterLatch rwlatch_;
/** Acquire the page write latch. */
inline void WLatch() { rwlatch_.WLock(); }
/** Release the page write latch. */
inline void WUnlatch() { rwlatch_.WUnlock(); }
/** Acquire the page read latch. */
inline void RLatch() { rwlatch_.RLock(); }
/** Release the page read latch. */
inline void RUnlatch() { rwlatch_.RUnlock(); }

每个Bucket都有一把读写锁，在哪里呢？明明在BucketPage中没有读写锁啊？在在Page类中！注意Bucket page 类不是 page的一个子类，我们只是通过reinterpret_cast将page.data_解释为了BucketPage而已，所以要使用Page中的锁，我们需要重新reinterpret回去，可以这样写：

  reinterpret_cast<Page *>(bucket_page)->WLatch();
	// modify bucket
  reinterpret_cast<Page *>(bucket_page)->WUnlatch();

  reinterpret_cast<Page *>(bucket_page)->RLatch();
	// read bucket
  reinterpret_cast<Page *>(bucket_page)->RUnlatch();

为什么不用dynamic cast呢，因为Page 和 BucketPage根本没有继承关系，编译器会报错，这属于C++语法的部分，请见C++语法那部分

'bustub::HashTableBucketPage<int, int, bustub::IntComparator>' is not polymorphic

我们的BucketPage 是由Page的data_ 属性重新解释为 BucketPage得来的：

template <typename KeyType, typename ValueType, typename KeyComparator>
HASH_TABLE_BUCKET_TYPE *HASH_TABLE_TYPE::FetchBucketPage(page_id_t bucket_page_id) {
   // 这里转换的是 char* -> HASH_TABLE_BUCKET_TYPE *, 没有继承关系，所以用 reinterpret_cast
  auto bucket_page =
      reinterpret_cast<HASH_TABLE_BUCKET_TYPE *>(buffer_pool_manager_->FetchPage(bucket_page_id)->GetData());
  return bucket_page;
}

接下来记录一下我的加锁策略。

Search(GetValue) : 为table上读锁，在bucket上也使用读锁
Insert : 由于此方法不会修改目录页但是修改bucket页，所以table使用读锁，bucket使用写锁
SplitInsert ：由于此方法会修改目录页也修改bucket页，所以table肯定是用写锁的，而bucket我们可以不加锁，因为其他方法肯定会先对table加锁，无论使用写锁还是读锁锁都于本方法(SplitInsert方法)使用的写锁互斥，所以我们不用在本方法中对bucket再加锁。
Remove : 和Insert的策略一样
Merge 和 ExtraMerge 和SpiltInsert一样，只需要为table加写锁，不用管bucket

注意Insert 和 Remove方法中都要调用其他方法，比如Insert方法会调用SplitInset，这是要注意解锁操作与调用顺序，否则会造成死锁。拿Insert举例来说，考虑加锁的逻辑后如下：

bool HASH_TABLE_TYPE::Insert(Transaction *transaction, const KeyType &key, const ValueType &value) {
   // lock table
   table->ReadLatch.Lock();
   
   bucket = hash_key_to_bucket(key);
   // lcok the bucket
   bucket->WriteLatch.Lock();
   insert_succeed = bucket->insert(key, value);
   bucket->WriteLatch.UnLock();
   
   // must unlock table, before we call splitinsert
   table->ReadLatch.UnLock();
   
   if (!insert_succeed && bucket is full) {
       insert_succeed = SplitInsert(key,value);
   }
   return insert_succeed;
}

bool HASH_TABLE_TYPE::SplitInsert(Transaction *transaction, const KeyType &key, const ValueType &value) {
   // lock table， because we use write lock of table so we don't care about bucket
   table->WriteLatch.Lock();
   bucket = hash_key_to_bucket_using_bmp(key);
   // recursion termination condition
   if (bucket is not full) {
       // just apply insert in the bucket 
       bucket.insert(key, value);
       // unlock before return ！
       table->WriteLatch.UnLock();
       return true;
   }
   
   // bucket is full , we need split it
   // gd : global_depth of directory; 
   // ld : local_depth of the bucket
   if (gd == ld) {
       ExpensionDirectory();
       CopyPageId();
   }
   // now we need split bucket
   image_bucket = buffer_pool_manager->NewPage();
   ajdust ld 、 gd、 page_id to verify invariants
   RehashKeyValue(bucket, image_bucket);
   // must upin pages, otherwise there is no space for bmp to get pages from disk or create pages
   upin pages just used
       
   // unlock table before recursive call
   table->WriteLatch.UnLock();    
   // recursive call
   return SplitInsert(key, value)
}

Task3其实没有什么难度，注意加锁解锁的调用顺序就行了，Debug过程中如果是死锁错误，那应该是好调的；如果在加锁的时候出现SEGV错误，可能不是加锁策略的问题，而是没有正确地执行 unpin 操作！

建议完成了hashtable的主要功能后再考虑线程安全， gradescope上的线程安全测试用例是完全独立的。

DEBUG

说一下我碰到的几个比较烦但是只要稍加注意就能避免的BUG。

不要忘记 unping 不使用的page。 gradescope的测试代码的bufferpool大小小得可怜，有些只有4个page的容量，所以一定要及时unpin避免bufferpool爆满而不能再获取页面
注意使用bmp的unpin方法的flag参数是true，还是flase。如果是true说明我们在本线程中修改过这个页面，所以bufferpool会用这个页面刷新磁盘上对应的页。如果搞错了逻辑，特别是该true的时候却设置为false，那不用说多线程场景了，就是单线程场景也很有可能会出现SEGV(使用未定义的内存)错误，因为bufferpool会将修改过的页面淘汰但是却没有刷新硬盘，等下次从硬盘上取数据就不符合预期了。
如果注释了一大片代码后，也请 make format一下。是的注释也有代码风格问题 : )。
bucket page的方法需要操作readable数组，请记住它是不连续的，详见上文的分析。
对于有加锁的代码块，一定在return前释放锁，很容易出现在一个if分支里return但是忘记解锁的情况，然后就导致死锁而超时。建议使用lock_guard。
数组越界错误。这本来是很容易诊断的bug，但是gradescope可能会超时而不报memory_test方面的错误，导致debug的方向不对。
搞清楚到底是上读锁还是写锁
C++语法：
1. 使用初始化列表进行初始化时，其初始化顺序已被成员定义的顺序定死，一定多加注意放置一些隐晦的BUG。如果两个成员变量相互依赖，一定要按照顺序初始化！
2. C++初学者注意引用，不要对着拷贝一顿修改，然后debug的时候很奇怪地发现原数据结构没有被修改！
```
std::unordered_map<string, vector<int>> umap;
// ... // 
auto vec = umap["copy"]; 
vec.push_back(1);  // 这是对着一个拷贝做修改

auto& vec = umap["reference"];
vec.push_back(1);  // 这才会在原数据结构上修改
```

建议：

不要做过早的优化，否则如果出现了一个Bug，很难定位。

建议是先完成单线程版本的，然后再完成多线程版本的，最后再考虑某些优化，比如读写锁的选择，bufferpool managerd的交互优化(flag是true还是false)等。

补充

动态hash的优点？

https://loonytek.com/2016/05/17/extendible-hashing/

https://eng.libretexts.org/Bookshelves/Computer_Science/Databases_and_Data_Structures/Book%3A_Data_Structures_(Wikibook)/09%3A_Hash_Tables/9.07%3A_Other_hash_table_algorithms

mysql的inodb中有hash索引吗？

mysql的inodb中存在hash索引，但是不能由用户指定使用hash索引取代默认的B+树索引。mysql的hash索引是一个内存数据结构，当系统检测到在内存建立hash索引进行快速查询能够提高性能时，会自动创建hash索引。

参考资料：

MySQL :: MySQL 8.0 Reference Manual :: 15.5.3 Adaptive Hash Index

PROJECT #3 - Query Execution

本实验难点在源码阅读理解上，没啥好说的只能一个一个地点进去看。

依然还是推荐这篇博客的内容熟悉相关api。

首先要对火山模型有个总览：sql语句会被拆分成一个树形图，其中的每个节点就是一个executor，这个实验就是要实现各个executor。而将整个sql语句拆分成多个executor的步骤则没有让我们实现。

TASK #1 - EXECUTORS

完成9个 executor类的编写

SeqScan

The SeqScanExecutor iterates over a table and return its tuples, one-at-a-time. A sequential scan is specified by a SeqScanPlanNode. The plan node specifies the table over which the scan is performed. The plan node may also contain a predicate; if a tuple does not satisfy the predicate, it is not produced by the scan.

我的做法是在.h文件中添加一个成员属性 TableIterator, 用来保存表中遍历的位置，方便再Next方法中调用。在Init方法中初始化迭代器

 tableheap_iterator_ = table_info_->table_->Begin(exec_ctx_->GetTransaction());

在Next方法中只需要对tableheap_iterator_进行递增操作，就可以实现整张表的遍历动作。

官网提示plan也会包含一个判断条件，我们可以这样使用它，相当于 Select ... From... Where ... 中对Where语句指定的元组进行筛选：

while (tableheap_iterator_ != table_info_->table_->Bnd()) {
    auto predicate = plan_->GetPredicate();
    if (predicate != nullptr && !predicate->Evaluate(&(*tableheap_iterator_), &(table_info_->schema_)).GetAs<bool>()) {
      // 不满足predicate, 且如果隔离等级为read_commited 则立刻释放锁
      ++tableheap_iterator_;
      continue;
    }
    ...
}
...

Insert

有两种类型的插入

一种称为 raw insert ，将要插入的值就存储在plan node中
另一种则从子执行器中取得tuple，将这个从子执行器得到的元组插入目标table中。即 INSERT INTO .. SELECT ..

所以在Init 和 next中因该分两种情况，在init中如果有子执行器，则应该先init子执行器

在插入tuple后还应该更新索引，API的使用如下:

 ...
if (insert_successed) {
      // 插入成功则更新索引
      auto indexs = exec_ctx_->GetCatalog()->GetTableIndexes(table_info_->name_);
      // 更新所有的索引
      for (auto &index_info : indexs) {
        Tuple key_tuple = tuple_to_be_inserted.KeyFromTuple(table_info_->schema_, index_info->key_schema_,
                                                           index_info->index_->GetKeyAttrs());
        index_info->index_->InsertEntry(key_tuple, rid_to_be_inserted, exec_ctx_->GetTransaction());
      }
  }

Update

update的修改对象总是来自于一个子执行器

总体逻辑和delete差不多，但是比delete还简单些

同样要更新索引，但是索引没有相应的api，所以可以先delete然后insert

Delete

和Update的逻辑相似，要删除的tuple总是来自于seqscan子执行器，将对应的rid的tuple， Mark delete 就行了

Nested Loop Join

无论课本还是课堂上的理论都是非常简单的

但这就是理论和实践的差距，上面的伪代码没有将火山模型考虑进去，火上模型要求我们为每个excutor些一个Next方法，一次返回一个有效的tuple，并且提前return出while循环，这就可能要求我们保存左表或者右表的一些状态，我的做法是在类的成员变量中保存状态

也要注意处理输入表为空的情况！

class NestedLoopJoinExecutor : public AbstractExecutor {
  ...
  Tuple left_tuple_; // 从左表保存一个tuple，逻辑上表示一个游标
  RID left_tuple_rid_;
  bool left_table_not_empty_; // 左表是否为空
}

NestLoopJoinExecutor 有左右两个自执行器，一般都是seqscan子执行器。对每一个left_tuple都对右表中的所有tuple判断它们是否满足predicate。

另外grade_scope线上测试有IO_cost 测试，会判断你对两个表的迭代次数是否有误。比如A表10条记录，B表10条记录，那么总共的“IO次数”(应该是算的对两张表调用next总次数) = 10 * 10 = 100。所以也要注意这里的逻辑，我之前写的逻辑迭代了101次（就多了一次），没有通过。

关于NestLoopJoin的改进： BlockNestLoopJoin算法。

Hash Join

需要构建一个哈希表，哈希表的key是左表的join对应的value，而哈希表的value是对应的tuples（放于vector）。基础数据结构选择unordered_map，按照官网的提示编写自定义键的 hash模板类和 == 函数,(参考)。我们也可以使用 map，只需要重载 < 就可以了。

个人选用的是std::unordered_map，键是一个自定义类型，值是一个Tuple数组，该数组的每个Tuple都有相同的JoinKey

// 关键类成员
std::unordered_map<JoinKey, std::vector<Tuple>> join_map_;

对于自定义类型，如果我们想要使用它，那么我们要做两件事：

第一要对自定义类重载 == 运算符。
第二要再std的命名空间中，特化hash类模板。

// hash_join_executor.h
namespace bustub {
    struct JoinKey {
      Value value_;
      bool operator==(const JoinKey &other) const { return value_.CompareEquals(other.value_) == CmpBool::CmpTrue; }
    };
}  // namespace bustub

// 注意是在std的命名空间中
namespace std {
    template <>
    struct hash<bustub::JoinKey> {
      std::size_t operator()(const bustub::JoinKey &join_key) const {
        size_t curr_hash = 0;
        if (!join_key.value_.IsNull()) {
            // 仿照SimpleAggregationHashTable 的写法提供hash函数
          curr_hash = bustub::HashUtil::CombineHashes(curr_hash, bustub::HashUtil::HashValue(&join_key.value_));
        }
        return curr_hash;
      }
    };
}  // namespace std

在init阶段，遍历左表的tuple，将每个tuple中参与join的value计算出来，将(value, tuple)放入join_map_中

在Next阶段就拿右表的joinvalue在哈希表中寻找存放这个joinkey的Tuple数组，一个一个遍历并连接即可

HashJoin 和 Sort-Merge Join https://15445.courses.cs.cmu.edu/fall2021/notes/10-joins.pdf

Hash join can only be used for equal-joins on the complete join key. （哈希连接只使用与等值连接）

注意这个实验的HashJoin只能进行等值连接！

Aggregation

这个与HashJoin相比反而简单，但是它的Api反倒很复杂

aggregate key 是 Sum() Count() 等函数括号里对应的那个key

group_atrs 故名思意就是group 后的哪个属性

看两个例子

SELECT COUNT(colB), SUM(colB), MIN(colB), MAX(colB) from test_3;

这个sql语句没有group by，所以SimpleAggregationHashTable表中的key只有一个，即为空

Init阶段对每一个chile_executor的tuple，只需要调用MakeAggregateKey 和 MakeAggregateValue先计算出 key 和value，然后再调用SimpleAggregationHashTable::InsertCombine（key,value）即可，它会根据聚合类型更新hash表中对应的value。
SELECT COUNT(colA), colB FROM test_1 GROUP BY colB；

这个语句对colB进行groupby，所以hash表的key是colB的值

Limit

没啥好说的，用成员变量记录个数就行了

Distinct

这里只需要一个集合数据结构去重就可以了，那么是set还是unordered_set呢？可能得看个人的实现了，我是这样的：

// 同样应该提供 == 和 hash()
namespace bustub{
    struct DistinKey {
      std::vector<Value> values_;

      bool operator==(const DistinKey &other) const {
        for (uint32_t i = 0; i < other.values_.size(); i++) {
          if (values_[i].CompareEquals(other.values_[i]) != CmpBool::CmpTrue) {
              // 只要有一个 不等就false
            return false;
          }
        }
        return true;
      }
    };
}  // namespace bustub

namespace std {
template <>
struct hash<bustub::DistinKey> {
  std::size_t operator()(const bustub::DistinKey &distinct_key) const {
    size_t curr_hash = 0;
    for (const auto &value : distinct_key.values_) {
      if (!value.IsNull()) {
        curr_hash = bustub::HashUtil::CombineHashes(curr_hash, bustub::HashUtil::HashValue(&value));
      }
    }
    return curr_hash;
  }
};
}  // namespace std
...
std::unordered_set<DistinKey> set_;

我这样的选法似乎不能使用 std::set, 因为 set要我们提供 < 操作符号重载，但是我的key是 std::vector values，它是一个"矢量"，如何在一个矢量上执行 < 操作呢？ bustub没有给出定义，所以不太能够写出准确的 < 定义。但是bustub 给了我们values.CompareEquals()函数，所以我们可以对vector中的value一一判断是否相等，像上面做的那样。

Debug

有个值得注意的点是对 RID的理解，它是对一个在表中元组的定位信息，如果一个元组不是从表中得到的(比如SecScan从表中得到的Rid是有效的)，像这样
```
*tuple = Tuple(values, output_shema_);
```
直接构造一个tuple，它是没有rid的，如果执行tuple->GetRid()得到的是一个非法的RID。

把元组插入表中，才能够得到一个有效的RID
```
table_info_->table_->InsertTuple(tuple, &rid_to_be_inserted, exec_ctx_->GetTransaction());
// rid_to_be_inserted 会被赋予正确的值
```

PROJECT #4 - Concurrency Control

本实验难度应该是第二大，不是很熟悉并发编程，所以debug起来也还是很痛苦。

TASK #1 - Lock Manager

课上的理论听得好好的，但是一到编程时间就蒙了。比如隔离级别的实现，本project不要求实现serailizable隔离级别，但尽管如此，我刚开始还是不知道要怎样具体地实现它们.

看了一些博客以及看了测试用例后渐渐摸清了, 推荐这个博客 https://blog.csdn.net/twentyonepilots/article/details/120868216, 写得比较清楚

总结以下 :

首先，各个隔离级别下可能发生的一致性问题有：

不可重复读和幻读的区别？

两种情况下，都是指在两次读操作返回的结果不同。区别就在于，幻读是多出了一些不存在的记录，而不可重复读是结果集已存在的记录发生了改变。

从造成原因来看，两者的区别很大，幻读出现的原因是 我们不可能对不存在的记录加锁，不可重复出现的原因是 读锁在commit之前被释放，而导致其他事务修改了对应的记录(这个记录是已存在的)

各个隔离级别的具体方案是：

isolation = READ_UNCOMMITTED : 不需要读锁，只需要写锁
isolation = READ_COMMITED : 读锁，写锁都需要，但是读锁在读完就释放，写时上写锁，在commit时才释放写锁 (不需要2pl,看测试代码就懂了)
isolation = REPEATABLE_READ ：使用二阶段锁，在growing阶段只能获取锁，在shrinking阶段只能释放锁，而且所有的锁都在commit时才释放(成了 rigorous 2pl)
isolation = SERIALIZABLE ：本lab不要求，理论上需要上一级别的所有要求 + index locks。因为幻读的出现的原因是我们不可能对不存在的记录加锁，所以我们只能对索引加锁了。sql语句运行过程中，如果插入一条新数据，之后势必会更新索引，如果新加入的索引在 index lock 保护的范围内，那么就回滚对记录所作的插入操作，因此index lock 起到了消除幻读的作用。

关键数据结构：

  class LockRequest {
   public:
    LockRequest(txn_id_t txn_id, LockMode lock_mode) : txn_id_(txn_id), lock_mode_(lock_mode), granted_(false) {}
    txn_id_t txn_id_;
    LockMode lock_mode_;
    bool granted_;
  };

  class LockRequestQueue {
   public:
    std::list<LockRequest> request_queue_; // 对某一RID的锁请求，全部都在这各list上等待
    std::condition_variable cv_;  // for notifying blocked transactions on this rid
    bool upgrading_ = false;      // 标记是否正在升级， while循环等待写锁释放时，upgrading为true，
                                  // 退出whilie循环后，将读锁升级并将ubgrading设置为false；
    bool has_writer_ = false;     // 标记这个等待队列中是否有读锁
    int sharing_count_ = 0;       // 获得 share lock 的事务个数
  };

...
  std::mutex latch_; // 这个mutex用来保护 lock_table_ ，在对lock_table_做修改时一定要首先获取这个互斥锁
  std::unordered_map<RID, LockRequestQueue> lock_table_; // 管理锁的table

以RID为key为申请锁的行记录创建一个LockRequestQueue，此后对该记录的上锁请求都加入到这个请求队列中。

has_writer 和 sharing_count 是我自己添加的属性，方便在等待条件变量时判断可上锁的条件。

而且在某一时刻在ConditionalVariable循环等待的请求，当sharing_count != 0时，读请求不会进入等待循环而写请求会，因为读写互斥，而读读不互斥（这里存在一个问题，就是读者可能会饿死写者，因为得等到sharingcount = 0是，写请求才会被唤醒）。

那么RID是什么？

RID唯一地标识了某条行记录的位置信息，它由两个成员变量组成：

page_id, 标识这条行记录位于磁盘上的哪个页（由diskManager管理）

slot_num，标识这条行记录在某个页的哪个槽

接下来分别记录三个Lock 和一个Unlock操作的大致逻辑

LockShared

- 2pl检测，只在 RR 隔离级别下检查事务的状态是否为Growing
- 为本事务对应的记录生成一个读锁请求，放到请求队列中，分两种情况
	- locktable中原先不存在对应Rid的LockRequestQueue，那么使用emplace_back api生成新的lockrequest
	- locktable中原先存在对应Rid的LockRequestQueue，表示已经有其他请求在该记录上执行操作
		- 执行cv.wait()循环,等待has_writer = false
		- 在wait循环外，表示本事务已经得到该记录写锁，那么执行一系列更新操作，（granted、sharing_count修改）
- txn->GetSharedLockSet()->emplace(rid);

LockExclusive

大致逻辑几乎与LockShared，但是cv.wait()等待has_writer = false && sharing_count = 0

LockUpgrade

- 2pl检测，只在 RR 隔离级别下检查事务的状态是否为Growing
- 如果该队列的upgrading = true，表示有一个锁正在升级，直接返回false
- txn->GetSharedLockSet()->erase(rid); 去除事务中原来的读锁
- 由于执行锁升级，所以在locktable中一定存在相应的lockrequest，修改它的sharing_count, graned_, lock_mode,并将upgrading 设置为true
	- 循环cv.wait()等待has_writer = false && sharing_count = 0
- txn->GetExclusiveLockSet()->emplace(rid); 升级完成添加获得的写锁

Unlock

-  txn->GetSharedLockSet()->erase(rid); // 删除本事务对应rid的读锁
-  txn->GetExclusiveLockSet()->erase(rid); // 删除本事务对应的写锁， 这两条语句不可能同时会执行有效动作，因为同一个RID的记录不可能同时被授予写锁和读锁
-  2pl处理，只有在 repeatable 隔离级别下才将事务状态改为 Shrinking
-  根据事务id，遍历 lockrequest_list 中的lockrequest，获得对应事务id的lockrequest
-  按照lockrequest的模式执行不同的操作
    - 如果lock模式为读锁， 那么sharing count --， 此时再判断是否 sharing count = 0 ，如果是的话就调用cv_.notify_all()，唤醒写请求;
	- 如果lock模式为写锁，那么将has_writer 设为false， 调用cv_.notify_all()，唤醒读或写请求;

以上所有操作都必须在获取latch_ 互斥锁才能进行，并且为了配合条件变量的使用，要配合unique_lock使用

std::unique_lock<std::mutex> unique_lk(latch_);
...
cv_.wait(uniq_lk);
...

从逻辑上说，lockshared 和 lockexclusive函数都应该判断是否已经对该rid进行了加锁，如果已经加锁了就返回false；对于锁升级来说，应该检查对应rid的记录是否持有读锁，如果没有则返回false。我当初没有想到这个逻辑，但是测试代码好像不会检测，所以也过了。

LockUpgrade的真实使用场景：

一个update executor，首先他会从子节点(seqscan executor)取出对应的记录，根据隔离级别的不同，这条记录从子节点返回时，它可能被加了读锁或者没有加锁。在没有加锁的情况下，update executor 应该对该记录执行LockExclusive，如果该记录已经加了读锁，那么应该对其执行LockUpgrade，将读锁升级为写锁。

TASK #2 - Deadlock Prevention

采用wound-wait(young wait for old)的方法预防死锁

具体说就是当两个事务，同时对一个记录加互相排斥的锁时，应该检查哪个事务时先运行的。比如T1先运行要对A加X锁，然而检测到T2也对A加了X锁，这时判断T1和T2哪个先运行，T1检测到是自己先运行，则Abort T2,把T2的锁抢为己有。如果T1检测到是T2先运行，那么就挂入锁申请队列等待T2释放X锁。

以上讨论同样适合X锁与S锁，因为X锁和S锁同样是互斥的。但是记住S锁互相是兼容的。

加入死锁检测的逻辑大致要该两个地方的逻辑，一个是在加入lockrequest队列时，运行wound-wait算法，另一个时当本事务被其他事务Abort时，可能正在cv.wait()循环中，所以我们在这个循环中还要检测自身事务的状态是否为Aborted，如果是则退出循环，返回false

用 LockShared 作为例子，添加死锁预防的后逻辑如下

- 2pl检测，只在 RR 隔离级别下检查事务的状态是否为Growing
- 为本事务对应的记录生成一个读锁请求，放到请求队列中，分两种情况
	- locktable中原先不存在对应Rid的LockRequestQueue，那么使用emplace_back api生成新的lockrequest
	- locktable中原先存在对应Rid的LockRequestQueue，表示已经有其他请求在该记录上执行操作
		- 执行woundwait算法，检查 LockRequestQueue中所有 `写` 请求，判断它们的txn_id是否大于本事务，如果是的话就表示这个发出这个请求的事务比本事务年轻，因此abort它; 
		- 调用一次cv.notify_all() 唤醒所有被abort的事务
		- 执行cv.wait()循环,等待has_writer = false，并且还要判断自身事务状态是否为abort，如果是就表示其他事务在执行 wound-wait算法时把本事物abort了，因此 break 退出循环
         - 检查本事务是否被abort，如果是则返回false
		- 到这里，表示本事务已经得到该记录写锁，那么执行一系列更新操作，（granted、sharing_count修改）
- txn->GetSharedLockSet()->emplace(rid);

TASK #3 - Concurrent Query Execution

修改project3中的 seq_scan delete update 执行器，使它们能够满足各种隔离级别下的特性。

seq_scan : 只涉及读锁,
- 在Next方法开头判断隔离级别是否为 READ_UNCOMMITTED, 如果是就不用加读锁, 其他两个隔离条件都需要加读锁.
- 在Next方法返回前, 判断隔离级别是否为REPEATABLE_READ, 如果不是则立刻释放读锁,如果是则不释放,由上层调用者在事务commit时一起释放
delete: 涉及写锁和锁升级, 首先应该意识到delete从它的子执行器(也就是scan executor)中获得tuple, 视隔离级别不同, 取得的tuple有没有上锁或者已经上读锁的两种可能
- 获得子执行器传出的tuple后, 判断当前事务的隔离级别, 如果是REPEATABLE_READ, 说明该tuple已经被上了读锁，那么对该tuple执行锁升级
- 否则对该tuple上写锁
- 在该防范中不需要解锁, transaction manager类会在事务commit后自动释放全部的锁
update: 同上
insert : 似乎没有要求？因为insert插入的记录是在前一刻不存在的，因此行记录锁是锁不住的。这会导致幻读的发生，但是实验并没有要求我们实现Serializable这个隔离级别，所以没有对insert executor做出具体要求，如果要实现Serializable隔离级别，则在RR隔离级别上加上index lock才行。

Debug

花费时间最长的是关于Reapeatable Read相关的测试

测试代码会这样测试你是否正确实现了Reapeatable Read隔离级别：

开启两个thread，记为thread1，thread2
thread1 执行两次Selcet操作，两次查询的数据的RID相同
thread2 执行一次Update操作，修改的数据与thread1的数据的RID相同
测试代码使用sleep操作保证调度顺序保证为：
- thread1第一次select -> thread2 update数据 -> thread1第二次select
如果正确实现了Repeatable隔离级别，那么thread1的第二次select结果与第一次的结果相同
这里的关键就是使得thread2在执行update时被阻塞。
UpdateExecutor有一个seqscan子Executor，子执行器已经对数据上了读锁，且由于隔离级别为Reapeatable Read，因此子执行器上的读锁不会归还直到事务结束。这时主执行器，在对记录进行update时就要使用LockUpdate方法了。
LockUpdate方法一定要确保在这种调度顺序下，thread2会在条件变量上等待，等待该RID的读锁和写锁数量为0。

具体如何调试？
说来惭愧，本人不是很会用GDB，且这个实验的并发读不高，所以一直使用printf和LOG_DEBUG在调式。主要观察thread2 的update方法是否在thread1 commit事务前结束执行。如果是的话，说明哪里有问题，那就打印更加详细的LOG，把等待队列的has_writer、readcount、是否陷入等待、是否执行唤醒操作、wound-wait算法是否正确abort了新事务等等，都打印出来，一个一个排查。

易错：当一个事务被wakeup时，并不代表它获取了锁，有可能它作为一个老事务被新事物abort了，因此一定要检查这种情况！

其他

读写锁

这个项目的读写锁是老师自己实现好的，感觉可以摸索摸索

与操作系统导论那本书的P272相比，bustub的读写锁实现有一个mutex和两个条件变量，多出了一个条件变量，一把mutex锁用来保护整个读写锁，reader和writer分别在各自的条件变量上等待。以及最重要的是，在RLOCK方法中判断了读者数量是否到达了最大数量，如果读者达到了最大数量，那么将不会获取锁，这就预防了写者被饿死。

// Identification: src/include/common/rwlatch.h
namespace bustub {
/**
 * Reader-Writer latch backed by std::mutex.
 */
class ReaderWriterLatch {
  using mutex_t = std::mutex;
  using cond_t = std::condition_variable;
  static const uint32_t MAX_READERS = UINT_MAX;

 public:
  ReaderWriterLatch() = default;
  ~ReaderWriterLatch() { std::lock_guard<mutex_t> guard(mutex_); }
  DISALLOW_COPY(ReaderWriterLatch); // 不允许赋值操作
  /**
   * Acquire a write latch.
   */
  void WLock() {
    std::unique_lock<mutex_t> latch(mutex_);
    while (writer_entered_) {
      reader_.wait(latch);
    }
    writer_entered_ = true;
    while (reader_count_ > 0) {
      writer_.wait(latch);
    }
  }

  /**
   * Release a write latch.
   */
  void WUnlock() {
    std::lock_guard<mutex_t> guard(mutex_);
    writer_entered_ = false;
    reader_.notify_all();
  }

  /**
   * Acquire a read latch.
   */
  void RLock() {
    std::unique_lock<mutex_t> latch(mutex_);
    while (writer_entered_ || reader_count_ == MAX_READERS) {
      reader_.wait(latch);
    }
    reader_count_++;
  }

  /**
   * Release a read latch.
   */
  void RUnlock() {
    std::lock_guard<mutex_t> guard(mutex_);
    reader_count_--;
    if (writer_entered_) {
      if (reader_count_ == 0) {
        writer_.notify_one();
      }
    } else {
      if (reader_count_ == MAX_READERS - 1) {
        reader_.notify_one();
      }
    }
  }

 private:
  mutex_t mutex_;
  cond_t writer_;
  cond_t reader_;
  uint32_t reader_count_{0};
  bool writer_entered_{false};
};

}  // namespace bustub

工厂模式

发现了一个简单的工厂模式的应用

bool Execute(const AbstractPlanNode *plan, std::vector<Tuple> *result_set, Transaction *txn,
               ExecutorContext *exec_ctx) {
    // Construct and executor for the plan
    auto executor = ExecutorFactory::CreateExecutor(exec_ctx, plan);

    // Prepare the root executor
    executor->Init();
	// ...... 下略
}

CreateExecutor方法如下所示：

返回值是一个使用unique_ptr管理的AbstractExecutor指针
方法内根据paln的种类创建对应的具体的Executor
各个具体的Executor则继承AbstractExector

std::unique_ptr<AbstractExecutor> ExecutorFactory::CreateExecutor(ExecutorContext *exec_ctx,
                                                                  const AbstractPlanNode *plan) {
  switch (plan->GetType()) {
    // Create a new sequential scan executor
    case PlanType::SeqScan: {
      return std::make_unique<SeqScanExecutor>(exec_ctx, dynamic_cast<const SeqScanPlanNode *>(plan));
    }

    // Create a new index scan executor
    case PlanType::IndexScan: {
      return std::make_unique<IndexScanExecutor>(exec_ctx, dynamic_cast<const IndexScanPlanNode *>(plan));
    }

    // Create a new insert executor
    case PlanType::Insert: {
      auto insert_plan = dynamic_cast<const InsertPlanNode *>(plan);
      auto child_executor =
          insert_plan->IsRawInsert() ? nullptr : ExecutorFactory::CreateExecutor(exec_ctx, insert_plan->GetChildPlan());
      return std::make_unique<InsertExecutor>(exec_ctx, insert_plan, std::move(child_executor));
    }

    // Create a new update executor
    // ... 下略
  }
}

class SeqScanExecutor : public AbstractExecutor {
 public: 
 // 下略
}

class IndexScanExecutor : public AbstractExecutor {
 public: 
 // 下略
}

class InsertExecutor : public AbstractExecutor {
 public: 
 // 下略
}
// ... 等等