CMU 15-445实验记录（三）：Project 2 B+Tree的插入与删除

B+Tree的删除的五种情况：

叶结点被删除后没有underflow，直接删除对应的key和recordPtr即可
叶结点被删除后有underflow，从sibling节点借一个key和recordPtr，从相邻节点借了一个key过来后，两个节点的key的范围都发生了变化，为了正确地反映指针指向的key的范围，必须更新中间节点，也就是父节点的key
叶结点被删除后有underflow，但是sibling的key的数量太少，如果被借走key自己也会underflow。此时需要与一个sibling合并，合并后两个节点的key的范围都发生了变化，为了正确地反映指针指向的key的范围，必须更新中间节点，也就是父节点的key。调用Delete (5, rightTree(5))删除父节点的指针和key，删除后父节点没有发生underflow。
叶结点被删除后有underflow，与一个sibling合并，再调用Delete (5, rightTree(5))删除父节点的指针和key，删除后父节点也发生了underflow，父节点再从它的sibling借一个key和recordPtr。
叶结点被删除后有underflow，与一个sibling合并后父节点也发生了underflow。父节点underflow后无法从sibling借key，而是与sibling进行合并。父节点merge后需要删除父节点的父节点的一个key，如果没有underflow，则B+Tree的删除到此结束；如果又发生了underflow，如果不是根节点，则还需要对父节点的父节点进行处理；如果是根节点，则删除根节点后结束。

调用Delete (13, rightTree(13))删除父节点的指针和key

删除后根节点变成了空，删除根节点

B+Tree处理叶结点和内部节点的对比

从sibling转移
- 叶结点：
  
  将左边的（或右边的）sibling的最后一个（或第一个）key和ptr转移到L的第一个位置（或最后一个位置），转移后两个节点的key的范围都发生了变化，必须更新中间节点，也就是将父节点的key更新为新的右子节点的第一个key的值。
- 内部节点：
  
  将左边的（或右边的）sibling的最后一个（或第一个）的ptr转移到N的第一个指针（或最后一个指针），为了保证这个指针指向的范围不变，将指针原位置的key放入父节点中，将父节点原来的key放入N中。
与sibling进行合并：
- 叶结点：
  
  如果左sibling存在，就将L的key和recordPtr以及最后一个link ptr合并到左sibling的后面；如果右sibling存在，就将右sibling合并到L的后面。合并后由于父节点少了一个子节点，需要删除父节点中的右子树和对应的key。
- 内部节点：
  
  如果左sibling存在，将N的所有ptr合并到左sibling的后面，为了保证这些指针指向的范围不变，将父节点key和原N的key也放入左sibling中。合并后由于父节点少了一个子节点，需要删除父节点中的右子树和对应的key。
  
  如果是右sibling存在，则将右sibling所有的ptr合并到N的后面。

B+Tree的合并和转移本质上是移动ptr，为了保证这些ptr指向的范围不变，需要相应地修改B+Tree中的key。

叶结点和内部节点数量的关注点是不同的，叶结点限制的是key的数量；而内部节点限制的是指针的数量。而在具体实现中，内部节点第一个键是不使用的，所以内部节点的kv对的数量就等于指针的数量，叶结点的kv对的数量等于key的数量。

内部节点的下限：至少有一半的ptr被使用，即内部节点至少包含最大指针数除二再向上取整的ptr。上限为所有指针都被使用

叶结点的下限：至少一半的key被使用，即叶结点至少包含最大值数除二再向上取整的key。上限为所有key都被使用

具体实现

每个B+Tree的内部或叶子节点（页）（即BPlusTreeLeafPage和BPlusTreeInternalPage对象）都对应着缓冲池中的一个Page对象的内容（即data_部分），所以每一次我们读取或写入一个叶子或内部页，我们需要先通过唯一的page_id将该页从缓冲池中fetch，再使用reinterpret_cast将该页的内容（即获取的Page对象的data_部分！而不是Page对象本身！）重新解释为LeafPage对象或InternalPage对象。读或写结束后，再调用bpm的Unpin方法结束对该Page对象的操作。如果改变了LeafPage对象或InternalPage对象的属性，就相当于改变了Page对象的data_部分，所以还需要将Page对象的dirty置为true。

我们的B +树索引只能支持唯一的键。也就是说，当我们尝试将重复的键插入索引中时，它不应执行插入并返回false。如果插入导致某一页中的kv对等于max_size，需要将这个节点进行split。

header_page记录了系统中的所有索引的名字以及对应索引的root_page_id，所以每当我们对某一个索引的根节点进行修改时，或者创建了一个新的根节点时，都需要调用UpdateRootPageId方法插入（参数为true）或更新（参数为false）header_page。

B+Tree初始化时给的max_size对叶节点和内部节点的含义不同：

对叶节点来说，max_size指节点最多可以容纳的kv对数量加一，一到达leaf_max_size就需要进行分裂
对内部节点来说，max_size是节点最多可容纳的kv对数量，也就是指针的数量，到达internal_max_size+1才需要进行分裂

我也实在不明白为什么要这样设计，这个问题在测试时才发现

因此：

对叶节点来说，min_size为max_size/2
对内部节点来说，min_size为(max_size+1)/2

小于min_size就发生了underflow，需要coalesce或redistribute

我们还需要指定函数调用Unpin的规则：

执行了fetchPage或newPage操作，并且没有将该page作为返回值，那么需要在本函数内unpin该page
调用了返回值是page对象的函数（或者返回参数是page对象的函数），那么需要在本函数内unpin该page

一个page对象的使用在哪个函数内结束，哪个函数就负责Unpin这个page

实现插入算法

如果当前为空树则创建一个新的树插入其中，否则就插入到叶结点中。

INDEX_TEMPLATE_ARGUMENTS
bool BPLUSTREE_TYPE::Insert(const KeyType &key, const ValueType &value, Transaction *transaction) {if (IsEmpty()) {StartNewTree(key, value);return true;}return InsertIntoLeaf(key, value, transaction);

将一对kv插入到一个空树中

将一对kv插入到一个空树中：从缓冲池中请求一个新的page，这个page就是我们的root page，将page id赋给root_page_id_，同时这个page也是叶page，将这个page对象转化为leaf_page对象，再初始化leaf_page对象，再向其中插入kv对。更新rootPageId，结束操作后，将缓冲池中的该页Unpin。

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::StartNewTree(const KeyType &key, const ValueType &value) {Page *root_page = buffer_pool_manager_->NewPage(&root_page_id_);if (root_page == nullptr) {throw "out of memory";}LeafPage *leaf_page = reinterpret_cast<LeafPage *>(root_page->GetData());leaf_page->Init(root_page_id_, INVALID_PAGE_ID, leaf_max_size_);// true为插入，false为更新UpdateRootPageId(true);leaf_page->Insert(key, value, comparator_);buffer_pool_manager_->UnpinPage(root_page_id_, true);
}

向指定叶结点中插入kv对：在该节点中找到大于等于插入key的第一个key，将kv数组中在该key之后的所有kv对向后移动一格，将我们的kv对插入，将页的大小加一。

INDEX_TEMPLATE_ARGUMENTS
int B_PLUS_TREE_LEAF_PAGE_TYPE::Insert(const KeyType &key, const ValueType &value, const KeyComparator &comparator) {int index = KeyIndex(key, comparator);assert(index >= 0);int end = GetSize();for (int i = end; i > index; i--) {array[i].first = array[i - 1].first;array[i].second = array[i - 1].second;}array[index].first = key;array[index].second = value;IncreaseSize(1);return GetSize();
}

在叶结点中找到大于等于要插入key的第一个key的index：

INDEX_TEMPLATE_ARGUMENTS
int B_PLUS_TREE_LEAF_PAGE_TYPE::KeyIndex(const KeyType &key, const KeyComparator &comparator) const {assert(GetSize() >= 0);int l = 0;int r = GetSize() - 1;while (l <= r) {int mid = (r - l) / 2 + l;if (comparator(array[mid].first, key) < 0) {l = mid + 1;} else {r = mid - 1;}}return r + 1;
}

查找叶结点并插入其中

在B+树中查找插入元素对应的叶结点并插入其中：

查找到插入元素对应的叶结点
向查找到的叶结点中插入
1. 由于我们的B+Tree只支持唯一的key，所以如果叶结点中已经存在相同的key，则马上返回false。否则就直接插入。
2. 如果插入后该叶结点内的kv对个数大于最大值则需要进行split，产生两个新节点，将右边节点的第一个key复制后插入到父节点

1、查找到插入元素对应的叶结点

从根节点开始查找，根据page_id从缓冲池中获取每一个节点对应的page对象，将page对象重新解释为Internal_page，调用Internal_page的lookup方法，查找到插入元素对应的下一级节点的page_id，再从缓冲池中获取下一级page对象，同时Unpin刚才操作的page对象。直到查找到叶结点为止。

参数中的leftMost是用来在实现Iterator时，先调用此函数定位到最左边的叶子节点的。所以如果leftMost为true，查找时的每一次循环直接使用当前节点中kv数组的第一个page_id即可。

INDEX_TEMPLATE_ARGUMENTS
Page *BPLUSTREE_TYPE::FindLeafPage(const KeyType &key, bool leftMost) {if (IsEmpty()) {return nullptr;}page_id_t next_page_id = root_page_id_;InternalPage *internal_page;for (internal_page = reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(next_page_id)->GetData());!internal_page->IsLeafPage();internal_page = reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(next_page_id)->GetData())) {page_id_t old_page_id = next_page_id;if (leftMost) {next_page_id = internal_page->ValueAt(0);} else {next_page_id = internal_page->Lookup(key, comparator_);}buffer_pool_manager_->UnpinPage(old_page_id, false);}return reinterpret_cast<Page *>(internal_page);
}

InternalPage的Lookup方法：在某个内部节点的kv数组中查找小于等于输入key的最大的key，对应的ptr就指向包含输入key的page

INDEX_TEMPLATE_ARGUMENTS
ValueType B_PLUS_TREE_INTERNAL_PAGE_TYPE::Lookup(const KeyType &key, const KeyComparator &comparator) const {assert(GetSize() > 1);for (int i = 1; i < GetSize(); i++) {if (comparator(key, array[i].first) < 0) {return array[i - 1].second;}}return array[GetSize() - 1].second;
}

2、向该叶结点中插入

由于我们的B+Tree只支持唯一的key，所以先查找叶结点中的kv数组，如果叶结点中已经存在相同的key，则马上返回false（在返回前还需要unpin当前的叶结点）。否则就直接插入。

如果插入后该叶结点内的kv对个数大于最大值则需要进行split，产生两个新节点（实际上是一个新节点，左节点还是使用之前的指针）
再调用InsertIntoParent，将右边节点的第一个key和指向右节点的指针插入到父节点中左节点对应的kv对的后面（split后要记得unpin分裂产生的新的叶结点）。插入结束后，还是要记住Unpin叶结点，同时dirty位置为true，因为此页的内容已经被修改了。
- 如果插入后，父节点也满了，则还需要对父节点递归进行split和InsertIntoParent。直到自己变成根节点或父节点没有满为止。

INDEX_TEMPLATE_ARGUMENTS
bool BPLUSTREE_TYPE::InsertIntoLeaf(const KeyType &key, const ValueType &value, Transaction *transaction) {Page *page = FindLeafPage(key, false);LeafPage *leaf_page = reinterpret_cast<LeafPage *>(page->GetData());ValueType v;if (leaf_page->Lookup(key, &v, comparator_)) {buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), false);return false;}leaf_page->Insert(key, value, comparator_);// 到达叶节点的max_size就需要进行分裂if (leaf_page->GetSize() >= leaf_page->GetMaxSize()) {LeafPage *new_leaf_page = Split(leaf_page);InsertIntoParent(leaf_page, new_leaf_page->KeyAt(0), new_leaf_page, transaction);buffer_pool_manager_->UnpinPage(new_leaf_page->GetPageId(), true);}buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), true);return true;
}

split

对叶结点split：

对内部节点split：

split：首先从缓冲池中请求一个新的page对象，将它重新解释为Internal_page或leaf_page，再进行初始化。将输入的page中的一半的kv对移动到新的page对象中，再修改next_page_id的指向，将新的page插入到输入page的后面（如果输入的是内部节点，则不需要修改next_page_id的指向）。注意，如果是内部节点，新的page中的第一个key就是要送给父节点的key，刚好可以忽略它，这样就将leaf_page和Internal_page的MoveHalfTo的操作统一起来了。

INDEX_TEMPLATE_ARGUMENTS
template <typename N>
N *BPLUSTREE_TYPE::Split(N *node) {page_id_t new_page_id;Page *new_page = buffer_pool_manager_->NewPage(&new_page_id);if (new_page == nullptr) {throw "out of memory";}if (node->IsLeafPage()) {LeafPage *new_leaf_page = reinterpret_cast<LeafPage *>(new_page->GetData());LeafPage *old_leaf_page = reinterpret_cast<LeafPage *>(node);new_leaf_page->Init(new_page_id, old_leaf_page->GetParentPageId(), leaf_max_size_);old_leaf_page->MoveHalfTo(new_leaf_page);new_leaf_page->SetNextPageId(old_leaf_page->GetNextPageId());old_leaf_page->SetNextPageId(new_page_id);} else {InternalPage *new_internal_page = reinterpret_cast<InternalPage *>(new_page->GetData());InternalPage *old_internal_page = reinterpret_cast<InternalPage *>(node);new_internal_page->Init(new_page_id, old_internal_page->GetParentPageId(), internal_max_size_);// 注意！！内部节点分裂后，需要更新分配给新的内部节点的所有子节点的父节点指向old_internal_page->MoveHalfTo(new_internal_page, buffer_pool_manager_);}return reinterpret_cast<N *>(new_page->GetData());
}

leaf_page的MoveHalfTo方法：

MoveHalfTo：调用copyNFrom方法从被调用leaf_page对象的array中移动一半的kv对到recipient中，再更新两个leaf_page对象的大小。

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_LEAF_PAGE_TYPE::MoveHalfTo(BPlusTreeLeafPage *recipient) {int move_nums = GetSize() / 2;MappingType *start = array + GetSize() - move_nums;recipient->CopyNFrom(start, move_nums);this->IncreaseSize(-1 * move_nums);recipient->IncreaseSize(move_nums);
}

CopyNFrom：将从start开始的size个对象复制到this对象的array中

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_LEAF_PAGE_TYPE::CopyNFrom(MappingType *items, int size) {for (int cnt = 0; cnt < size; cnt++) {this->array[cnt] = *(items + cnt);}
}

internal_page的MoveHalfTo方法：

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_INTERNAL_PAGE_TYPE::MoveHalfTo(BPlusTreeInternalPage *recipient,BufferPoolManager *buffer_pool_manager) {int move_nums = GetSize() / 2;MappingType *start = array + GetSize() - move_nums;recipient->CopyNFrom(start, move_nums, buffer_pool_manager);this->IncreaseSize(-1 * move_nums);recipient->IncreaseSize(move_nums);
}

CopyNFrom：与叶节点的相同，需要注意的一点是：内部节点分裂后，需要更新分配给新的内部节点的所有子节点的父节点指针

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_INTERNAL_PAGE_TYPE::CopyNFrom(MappingType *items, int size, BufferPoolManager *buffer_pool_manager) {for (int cnt = 0; cnt < size; cnt++) {this->array[cnt] = *(items + cnt);// 注意！！内部节点分裂后，需要更新分配给新的内部节点的所有子节点的父节点指针auto child_page = reinterpret_cast<BPlusTreePage *>(buffer_pool_manager->FetchPage(array[cnt].second)->GetData());child_page->SetParentPageId(this->GetPageId());buffer_pool_manager->UnpinPage(array[cnt].second, true);}
}

InsertIntoParent

InsertIntoParent：

如果old_node就是根节点，那么需要向缓冲池请求一个新的页来创建新的根节点，初始化根节点，更新root_page_id的值（以及header_page），调用PopulateNewRoot方法使用old_node，key和new_node填充新创建的根节点，再修改old_node和new_node的父指针。最后还需要调用Unpin结束对刚才从缓冲池中获取的新的根页的操作。
找到old_node的父节点，调用InsertNodeAfter方法将new_ndoe和第一个key插入到父节点中old_ndoe的后面，再修改new_node的父指针
- 如果插入后的大小超过maxsize，还需要对父节点进行split和InsertIntoParent（split后要记得unpin分裂产生的新的叶结点）
对父节点进行unpin，结束操作。

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::InsertIntoParent(BPlusTreePage *old_node, const KeyType &key, BPlusTreePage *new_node,Transaction *transaction) {if (old_node->IsRootPage()) {// 创建新的根节点要更新root_page_id_和header_pagePage *new_page = buffer_pool_manager_->NewPage(&root_page_id_);UpdateRootPageId(false);InternalPage *new_root_page = reinterpret_cast<InternalPage *>(new_page->GetData());new_root_page->Init(root_page_id_, INVALID_PAGE_ID, internal_max_size_);new_root_page->PopulateNewRoot(old_node->GetPageId(), key, new_node->GetPageId());old_node->SetParentPageId(root_page_id_);new_node->SetParentPageId(root_page_id_);buffer_pool_manager_->UnpinPage(root_page_id_, true);} else {page_id_t parent_page_id = old_node->GetParentPageId();InternalPage *parent_page =reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(parent_page_id)->GetData());parent_page->InsertNodeAfter(old_node->GetPageId(), key, new_node->GetPageId());new_node->SetParentPageId(parent_page_id);// 对内部节点来说，到达max_size+1才需要进行分裂if (parent_page->GetSize() > parent_page->GetMaxSize()) {InternalPage *uncle_page = Split(parent_page);InsertIntoParent(parent_page, uncle_page->KeyAt(0), uncle_page);buffer_pool_manager_->UnpinPage(uncle_page->GetPageId(), true);}buffer_pool_manager_->UnpinPage(parent_page_id, true);}
}

PopulateNewRoot：在当前page中填充两个value（page_id）和一个key，变成一个新的root_page，再设置大小为2 。此方法只能在InsertIntoParent中调用

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_INTERNAL_PAGE_TYPE::PopulateNewRoot(const ValueType &old_value, const KeyType &new_key,const ValueType &new_value) {array[0].second = old_value;array[1].first = new_key;array[1].second = new_value;SetSize(2);
}

InsertNodeAfter：在当前page中找到old_value的位置，然后将new_key和new_value插入其中

INDEX_TEMPLATE_ARGUMENTS
int B_PLUS_TREE_INTERNAL_PAGE_TYPE::InsertNodeAfter(const ValueType &old_value, const KeyType &new_key,const ValueType &new_value) {int index = ValueIndex(old_value);for (int i = GetSize(); i > index + 1; i--) {array[i] = array[i - 1];}array[index + 1] = MappingType(new_key, new_value);IncreaseSize(1);return GetSize();
}

实现删除算法

先找到包含目标key的叶节点（如果是空树则直接返回），再调用RemoveAndDeleteRecord在叶节点上直接删除对应的key值（如果该叶中不存在该key，则直接返回）。删除后如果叶节点中key的个数小于最小值，则调用CoalesceAndRedistribute函数处理underflow。

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Remove(const KeyType &key, Transaction *transaction) {if (IsEmpty()) {return;}Page *page = FindLeafPage(key, false);  // unpinLeafPage *leaf_page = reinterpret_cast<LeafPage *>(page->GetData());leaf_page->RemoveAndDeleteRecord(key, comparator_);assert(leaf_page != nullptr);if (leaf_page->GetSize() < leaf_page->GetMinSize()) {CoalesceOrRedistribute(leaf_page, transaction);}buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), true);
}

CoalesceAndRedistribute函数：处理节点的underflow

如果是根节点则调用AdjustRoot函数，处理根节点的underflow
如果不是根节点，则先获取该page的兄弟page。（先获取左边的sibling，如果当前节点的index为0，则获取右边的sibling）
- 如果这两个page可以合并到同一个page中，则调用Coalesce函数进行合并。（在Coalesce函数中，我们统一认为参数neighbor_node为左边的节点，node为右边的节点，所以在调用Coalesce前，如果sibling不是在node的前面，我们需要转换两个指针的指向）
- 否则只能调用Redistribute函数进行借节点

INDEX_TEMPLATE_ARGUMENTS
template <typename N>
bool BPLUSTREE_TYPE::CoalesceOrRedistribute(N *node, Transaction *transaction) {assert(node != nullptr);if (node->IsRootPage()) {return AdjustRoot(node);}N *sibling;// 获取的sibling是否是node的下一个节点// Unpinbool node_prev_sibling = FindSibling(node, &sibling);// unpinInternalPage *parent_page =reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(node->GetParentPageId())->GetData());assert(node != nullptr && sibling != nullptr);if (node->GetSize() + sibling->GetSize() <= MaxSize(node)) {// 统一认为sibling在左边，node在右边if (node_prev_sibling) {N *temp = sibling;sibling = node;node = temp;}// index是右边的节点在父节点中的indexint index = parent_page->ValueIndex(node->GetPageId());Coalesce(&sibling, &node, &parent_page, index, transaction);buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), true);buffer_pool_manager_->UnpinPage(sibling->GetPageId(), true);return true;}// index为underflow的节点在父节点中的indexint index = parent_page->ValueIndex(node->GetPageId());Redistribute(sibling, node, index);buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), false);buffer_pool_manager_->UnpinPage(sibling->GetPageId(), true);return false;
}

INDEX_TEMPLATE_ARGUMENTS
template <typename N>
int BPLUSTREE_TYPE::MaxSize(N *node) {return node->IsLeafPage() ? node->GetMaxSize() - 1 : node->GetMaxSize();
}

获取sibling节点：

// 如果获取的是前一个sibling，则返回false；下一个sibling，则返回true
INDEX_TEMPLATE_ARGUMENTS
template <typename N>
bool BPLUSTREE_TYPE::FindSibling(N *node, N **sibling) {// unpinInternalPage *parent_page =reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(node->GetParentPageId())->GetData());int index = parent_page->ValueIndex(node->GetPageId());int sibling_index = index - 1;if (index == 0) {sibling_index = index + 1;}page_id_t sibling_page_id = parent_page->ValueAt(sibling_index);(*sibling) = reinterpret_cast<N *>(buffer_pool_manager_->FetchPage(sibling_page_id)->GetData());buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), false);return index == 0;
}

AdjustRoot：处理根节点的underflow，有两种情况：

根节点在underflow之前只有两个子节点，两个子节点合并后，只剩下了一个子节点，此时根节点为内部节点，且节点内只有一个指针，即根节点的大小为1。此时需要调用RemoveAndReturnOnlyChild函数把根节点删除，并返回该指针指向的唯一的子节点。将子节点更新为新的根节点。
整棵B+Tree中的所有值都被删除了，B+Tree为空，此时根节点为叶结点，且大小为0，直接将根节点删除即可

否则不需要有page被删除，则直接return flase

INDEX_TEMPLATE_ARGUMENTS
bool BPLUSTREE_TYPE::AdjustRoot(BPlusTreePage *old_root_node) {// case 2: when you delete the last element in whole b+ treeassert(old_root_node != nullptr);if (old_root_node->IsLeafPage()) {assert(old_root_node->GetSize() == 0);assert(old_root_node->GetParentPageId() == INVALID_PAGE_ID);buffer_pool_manager_->UnpinPage(old_root_node->GetPageId(), false);buffer_pool_manager_->DeletePage(root_page_id_);root_page_id_ = INVALID_PAGE_ID;// false更新UpdateRootPageId(false);// 删除了一个page，返回truereturn true;}// case 1: when you delete the last element in root page, but root page still// has one last childif (old_root_node->GetSize() == 1) {InternalPage *root_page = reinterpret_cast<InternalPage *>(old_root_node);page_id_t new_root_page_id = root_page->RemoveAndReturnOnlyChild();root_page_id_ = new_root_page_id;UpdateRootPageId(false);// 将新的根节点的父节点置为INVALIDInternalPage *new_root_page =reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(new_root_page_id)->GetData());new_root_page->SetParentPageId(INVALID_PAGE_ID);buffer_pool_manager_->UnpinPage(new_root_page_id, true);buffer_pool_manager_->UnpinPage(old_root_node->GetPageId(), false);buffer_pool_manager_->DeletePage(old_root_node->GetPageId());return true;}return false;
}

Coalesce的过程

叶节点中的合并：

内部节点中的合并：

Coalesce函数：参数为左边的节点，右边的节点，父节点，右边节点在父节点中对应的index。因为在合并时，需要删除右边节点在父节点中对应的指针和key。

调用MoveAllTo函数，将右边节点中的所有kv对移动到左边的节点。

对叶节点，直接移动即可
对内部节点，需要先将右边节点在父节点中对应的key移动到左边节点，再移动右边中的所有的kv对。还需要更新被合并的page的所有子节点的parent_page_id。所以调用内部节点的MoveAllTo函数时，还需要传入右边节点在父节点中对应的key，以及bpm

再删除右节点的page，调用remove函数移除父节点中对应的KV对。如果父节点又发生了underflow，还需要调用CoalesceAndRedistribute递归处理父节点。

INDEX_TEMPLATE_ARGUMENTS
template <typename N>
bool BPLUSTREE_TYPE::Coalesce(N **neighbor_node, N **node,BPlusTreeInternalPage<KeyType, page_id_t, KeyComparator> **parent, int index,Transaction *transaction) {if ((*node)->IsLeafPage()) {LeafPage *leaf_node = reinterpret_cast<LeafPage *>(*node);LeafPage *leaf_neighbor_node = reinterpret_cast<LeafPage *>(*neighbor_node);leaf_node->MoveAllTo(leaf_neighbor_node);leaf_neighbor_node->SetNextPageId(leaf_node->GetNextPageId());} else {InternalPage *internal_node = reinterpret_cast<InternalPage *>(*node);InternalPage *internal_neighbor_node = reinterpret_cast<InternalPage *>(*neighbor_node);internal_node->MoveAllTo(internal_neighbor_node, (*parent)->KeyAt(index), buffer_pool_manager_);}buffer_pool_manager_->UnpinPage((*node)->GetPageId(),true);buffer_pool_manager_->DeletePage((*node)->GetPageId());(*parent)->Remove(index);assert((*parent) != nullptr);if ((*parent)->GetSize() < (*parent)->GetMinSize()) {return CoalesceOrRedistribute((*parent), transaction);}return true;
}

叶节点的MoveAllTo方法：

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_LEAF_PAGE_TYPE::MoveAllTo(BPlusTreeLeafPage *recipient) {assert(recipient != nullptr);int move_num = this->GetSize();int recipient_start_index = recipient->GetSize();for (int i = 0; i < move_num; i++) {recipient->array[recipient_start_index + i] = this->array[i];}this->IncreaseSize(-1 * move_num);recipient->IncreaseSize(move_num);
}

内部节点的MoveAllTo方法：

INDEX_TEMPLATE_ARGUMENTS
void B_PLUS_TREE_INTERNAL_PAGE_TYPE::MoveAllTo(BPlusTreeInternalPage *recipient, const KeyType &middle_key,BufferPoolManager *buffer_pool_manager) {int move_num = this->GetSize();assert(recipient != nullptr);int recipient_start_index = recipient->GetSize();this->SetKeyAt(0, middle_key);for (int i = 0; i < move_num; i++) {recipient->array[recipient_start_index + i] = this->array[i];BPlusTreePage *child_page =reinterpret_cast<BPlusTreePage *>(buffer_pool_manager->FetchPage(array[i].second)->GetData());child_page->SetParentPageId(recipient->GetParentPageId());buffer_pool_manager->UnpinPage(array[i].second, true);}this->IncreaseSize(-1 * move_num);recipient->IncreaseSize(move_num);
}

Redistribute的过程

叶节点：

内部节点：

Redistribute函数：参数为sibling节点，underflow的节点，underflow的节点在父节点中对应的index。

如果underflow的节点在左边（index等于0）：
- 如果是叶节点，则调用叶节点的MoveFirstToEndOf函数，此函数将sibling的第一个kv对移动到underflow节点的后面。然后修改右边节点在父节点中对应的key
- 否则调用内部节点的MoveFirstToEndOf函数。
如果underflow的节点在右边，与上面类似

INDEX_TEMPLATE_ARGUMENTS
template <typename N>
void BPLUSTREE_TYPE::Redistribute(N *neighbor_node, N *node, int index) {InternalPage *parent_page =reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(node->GetParentPageId())->GetData());if (index == 0) {// 右边节点在父节点中对应的indexint index = parent_page->ValueIndex(neighbor_node->GetPageId());if (neighbor_node->IsLeafPage()) {LeafPage *Leaf_neighbor_node = reinterpret_cast<LeafPage *>(neighbor_node);LeafPage *Leaf_node = reinterpret_cast<LeafPage *>(node);Leaf_neighbor_node->MoveFirstToEndOf(Leaf_node);parent_page->SetKeyAt(index, Leaf_neighbor_node->KeyAt(0));} else {InternalPage *Internal_neighbor_node = reinterpret_cast<InternalPage *>(neighbor_node);InternalPage *Internal_node = reinterpret_cast<InternalPage *>(node);Internal_neighbor_node->MoveFirstToEndOf(Internal_node, parent_page->KeyAt(index), buffer_pool_manager_);parent_page->SetKeyAt(index, Internal_neighbor_node->KeyAt(0));}} else {// 右边节点在父节点中对应的indexint index = parent_page->ValueIndex(node->GetPageId());if (neighbor_node->IsLeafPage()) {LeafPage *Leaf_neighbor_node = reinterpret_cast<LeafPage *>(neighbor_node);LeafPage *Leaf_node = reinterpret_cast<LeafPage *>(node);Leaf_neighbor_node->MoveLastToFrontOf(Leaf_node);parent_page->SetKeyAt(index, Leaf_node->KeyAt(0));} else {InternalPage *Internal_neighbor_node = reinterpret_cast<InternalPage *>(neighbor_node);InternalPage *Internal_node = reinterpret_cast<InternalPage *>(node);Internal_neighbor_node->MoveLastToFrontOf(Internal_node, parent_page->KeyAt(index), buffer_pool_manager_);parent_page->SetKeyAt(index, Internal_node->KeyAt(0));}}buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), true);
}

实现迭代器

迭代器的范围为最左边的叶节点，到最右边的叶节点的最后一个kv对的下一个kv对（end）。

index_iterator.h：

  IndexIterator(B_PLUS_TREE_LEAF_PAGE_TYPE *leftmost_leaf, int index, BufferPoolManager *bpm);~IndexIterator();bool isEnd();const MappingType &operator*();IndexIterator &operator++();bool operator==(const IndexIterator &itr) const {return itr.cur_leaf_page_ == cur_leaf_page_ && itr.index_ == index_;}bool operator!=(const IndexIterator &itr) const {INDEXITERATOR_TYPE temp_it = *this;return !(itr == temp_it);}private:// add your own private member variables hereB_PLUS_TREE_LEAF_PAGE_TYPE *cur_leaf_page_;int index_;BufferPoolManager *buffer_pool_manager_;

index_iterator.cpp：

INDEX_TEMPLATE_ARGUMENTS
INDEXITERATOR_TYPE::IndexIterator(B_PLUS_TREE_LEAF_PAGE_TYPE *leftmost_leaf, int index, BufferPoolManager *bpm): cur_leaf_page_(leftmost_leaf), index_(index), buffer_pool_manager_(bpm) {}INDEX_TEMPLATE_ARGUMENTS
INDEXITERATOR_TYPE::~IndexIterator(){if(cur_leaf_page_ != nullptr){buffer_pool_manager_->UnpinPage(cur_leaf_page_->GetPageId(),false);}
}INDEX_TEMPLATE_ARGUMENTS
bool INDEXITERATOR_TYPE::isEnd() {assert(cur_leaf_page_ != nullptr);return cur_leaf_page_->GetNextPageId() == INVALID_PAGE_ID && index_ == cur_leaf_page_->GetSize();
}INDEX_TEMPLATE_ARGUMENTS
const MappingType &INDEXITERATOR_TYPE::operator*() { return cur_leaf_page_->GetItem(index_); }INDEX_TEMPLATE_ARGUMENTS
INDEXITERATOR_TYPE &INDEXITERATOR_TYPE::operator++() {index_++;assert(cur_leaf_page_ != nullptr);if (index_ >= cur_leaf_page_->GetSize() && cur_leaf_page_->GetNextPageId() != INVALID_PAGE_ID) {page_id_t next_page_id = cur_leaf_page_->GetNextPageId();buffer_pool_manager_->UnpinPage(cur_leaf_page_->GetPageId(), false);cur_leaf_page_ =reinterpret_cast<B_PLUS_TREE_LEAF_PAGE_TYPE *>(buffer_pool_manager_->FetchPage(next_page_id)->GetData());index_ = 0;}return *this;
}