template <typename KeyType,
          typename ValueType,
          typename KeyComparator>
class BPlusTree{
   // ---
};

这些类别已经为你实现了

• KeyType: The type of each key in the index. This will only be GenericKey, the actual size of GenericKey is specified and instantiated with a template argument and depends on the data type of indexed attribute.

• ValueType: The type of each value in the index. This will only be 64-bit RID.

• KeyComparator: The class used to compare whether two KeyType instances are less/greater-than each other. These will be included in the KeyType implementation files.

TASK #2.B - B+TREE DATA STRUCTURE (DELETION)

您的B+树索引需要支持删除。如果删除导致某些页面低于占用阈值，那么您的B+树索引应该正确执行合并或重新分配。同样，您的B+树索引只能支持唯一键

TASK #3 - INDEX ITERATOR

您将构建一个通用索引迭代器，以有效地检索所有叶子页面。 基本思想是将它们组织到一个链接列表中，然后按照B + Tree叶子页中存储的特定方向遍历每个键/值对。 您的索引迭代器应遵循C ++ 17中定义的迭代器功能，包括使用一组运算符对一系列元素进行迭代的能力，以及for-each循环（至少具有++，==，!=和解引用运算符）。 请注意为了支持索引的每个循环功能，您的BPlusTree应该正确实现begin（）和end()。

您必须在指定的文件中实现索引迭代器。 仅允许您修改头文件（src / include / storage / index / index_iterator.h）及其相应的源文件（src / index / storage / index_iterator.cpp）。 您不需要修改任何其他文件。 您必须在这些文件中的IndexIterator类中实现以下功能。 在索引迭代器的实现中，只要您具有以下三种方法，就可以添加任何帮助程序方法。

• isEnd(): Return whether this iterator is pointing at the last key/value pair.
• operator++(): Move to the next key/value pair.
• operator*(): Return the key/value pair this iterator is currently pointing at.
• operator==(): Return whether two iterators are equal
• operator!=(): Return whether two iterators are not equal.

TASK #4 - CONCURRENT INDEX

在这一部分中，您需要更新原始的单线程B + Tree索引，以便它可以支持并发操作。 我们将使用课堂和教科书中介绍的Latch捕捉技术。 遍历索引的线程将获取然后释放B + Tree页上的Latch锁。 如果线程的子页面被认为是“安全的”，则该线程只能释放其父页面上的Latch锁。 请注意，“安全”的定义可能会根据线程执行的操作类型而有所不同：

• Search: Starting with root page, grab read (R) latch on child Then release latch on parent as soon as you land on the child page.
• Insert: Starting with root page, grab write (W) latch on child. Once child is locked, check if it is safe, in this case, not full. If child is safe, release all locks on ancestors.
• Delete: Starting with root page, grab write (W) latch on child. Once child is locked, check if it is safe, in this case, at least half-full. (NOTE: for root page, we need to check with different standards) If child is safe, release all locks on ancestors.

Hints

1. 你必须使用传入的transaction，把已经加锁的页面保存起来。
2. 我们提供了读写锁存器的实现（src / include / common / rwlatch.h）。 并且已经在页面头文件下添加了辅助函数来获取和释放Latch锁（src / include / storage / page / page.h）。

2. Insert实现

首先附上书上的b+树插入算法

对上面几种情况的分析

1. 如果当前为空树则创建一个叶子结点并且也是根节点

   -- 这里是leaf结点所以这里需要用到leaf page内的函数

   -- 注意这里需要用lab1实现的buffer池管理器来获得page。 这里记得创建完新的结点之后要unpin

   -- 进行插入的时候用二分插入来进行优化

   • 创建新结点

   INDEX_TEMPLATE_ARGUMENTS
   void BPLUSTREE_TYPE::StartNewTree(const KeyType &key, const ValueType &value) {
     auto page = buffer_pool_manager_->NewPage(&root_page_id_);
     if (page == nullptr) {
       throw "all page are pinned";
     }
     auto root =reinterpret_cast<BPlusTreeLeafPage<KeyType, ValueType,KeyComparator> *>(page->GetData());
     UpdateRootPageId(true);
     root->Init(root_page_id_, INVALID_PAGE_ID ,leaf_max_size_);
     root->Insert(key, value, comparator_);
     // unpin
     buffer_pool_manager_->UnpinPage(root->GetPageId(), true);
   }
   

   • insert函数

   /*
   in b_plus_leaf_page.h
   */
   INDEX_TEMPLATE_ARGUMENTS
   int B_PLUS_TREE_LEAF_PAGE_TYPE::Insert(const KeyType &key, const ValueType &value, const KeyComparator &comparator) {
     if(!GetSize()||comparator(key, KeyAt(GetSize() - 1)) > 0) array[GetSize()] = {key, value};
       else{
         int l=0,r=GetSize()-1;
         while(l<r){
           int mid=(l+r)>>1;
           if(comparator(key,array[mid].first)<0)r=mid;
           else if(comparator(key,array[mid].first)>0)l=mid+1;
           else assert(0);
         }
         memmove(array + r + 1, array + r,static_cast<size_t>((GetSize() - r)*sizeof(MappingType)));
         array[r] = {key, value};
       }
     IncreaseSize(1);
     return GetSize();
   }
   

2. 否则寻找应该包含插入元素的叶子结点
   -- 首先找到叶子结点
   -- 如果叶子结点内的元素个数小于最大值则直接插入
   -- 否则需要进行分裂。产生两个新的结点。把元素上提
   -- 如果提到父亲结点，父结点仍需要分裂。则递归进行分裂否则结束

   a . 如果叶子结点内的关键字小于m-1,则直接插入到叶子结点insert_into_leaf

   • 利用findLeafPage函数找到叶子结点

   要考虑无论是读或者写从根节点。到叶子结点都需要加锁。然后注意释放锁否则会锁死。（这个地方测试的时候卡死了我好久）

   这里对原来的函数定义做了一些修改。加入了操作类型的判断。

   /*
   定义在b_plus_tree.h中
   定义方法和定义page类型保持一致
   */
   enum class Operation { READ = 0, INSERT, DELETE };
   

   INDEX_TEMPLATE_ARGUMENTS
   Page *BPlusTree<KeyType, ValueType, KeyComparator>::FindLeafPage(const KeyType &key, bool leftMost, Operation op, Transaction *transaction) {
   
     if (IsEmpty()) {
       return nullptr;
     }
     auto root = buffer_pool_manager_->FetchPage(root_page_id_);
     if (root == nullptr) {
       throw "no page can find";
     }
   
     if (op == Operation::READ) {
       root->RLatch();
     } else {
       root->WLatch();
     }
     if (transaction != nullptr) {
       transaction->AddIntoPageSet(root);
     }
   
     auto node = reinterpret_cast<BPlusTreePage *>(root->GetData());
     while (!node->IsLeafPage()) {
       auto internal =reinterpret_cast<BPlusTreeInternalPage<KeyType, page_id_t,KeyComparator> *>(node);
       page_id_t parent_page_id = node->GetPageId(), child_page_id;
       if (leftMost) {
         child_page_id = internal->ValueAt(0);
       } else {
         child_page_id = internal->Lookup(key, comparator_);
       }
       auto child = buffer_pool_manager_->FetchPage(child_page_id);
       if (child == nullptr) {
         throw "not find child in findLeaf";
       }
   
       if (op == Operation::READ) {
         child->RLatch();
         UnlockUnpinPages(op, transaction);
       } else {
         child->WLatch();
       }
       node = reinterpret_cast<BPlusTreePage *>(child->GetData());
       assert(node->GetParentPageId() == parent_page_id);
       // child is locked, If child is safe, release all locks on ancestors.
       if (op != Operation::READ && isSafe(node, op)) {
         UnlockUnpinPages(op, transaction);
       }
       if (transaction != nullptr) {
         transaction->AddIntoPageSet(child);
       } else {
         root->RUnlatch();
         buffer_pool_manager_->UnpinPage(root->GetPageId(), false);
         root = child;
       }
     }
     return reinterpret_cast<Page*>(node);
   }
   

   • Lookup函数

   找到key值所在的page---二分查找

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

   • 找到Leaf page之后

   判断该元素是否已经在树中。如果该元素当前不在树中则调用insert函数直接插入

    // if already in the tree, return false
    ValueType v;
    if (leaf->Lookup(key, &v, comparator_)) {
      UnlockUnpinPages(Operation::INSERT, transaction);
      return false;
    }
   //case 1  keys in leaf page <m-1
   if (leaf->GetSize() < leaf->GetMaxSize()) {
   leaf->Insert(key, value, comparator_);
   }
   

b. 进行分裂

3. 分裂的步骤

4. 调用split函数对叶子结点进行分割

   --- split的时候会产生一个含有m-m/2个关键字的新结点。注意把两个叶子结点连接起来。

   --- 然后调用InsertIntoParent

    // case 2 need to split
     else {
       leaf->Insert(key, value, comparator_);
       auto new_leaf = Split<BPlusTreeLeafPage<KeyType, ValueType, KeyComparator>>(leaf);
       new_leaf->SetNextPageId(leaf->GetNextPageId());
       leaf->SetNextPageId(new_leaf->GetPageId());
   
       // insert the split key into parent
       InsertIntoParent(leaf, new_leaf->KeyAt(0), new_leaf, transaction);
     }
     UnlockUnpinPages(Operation::INSERT, transaction);
   
     return true;
   
   }
   

5. 在InsertIntoParent中

   case1-- 如果当前结点为根节点。则创建一个新的根节点。新根节点的子结点为分裂所得(经过split操作后)得到的两个结点

   INDEX_TEMPLATE_ARGUMENTS
   void BPLUSTREE_TYPE::InsertIntoParent(BPlusTreePage *old_node, const KeyType &key, BPlusTreePage *new_node,Transaction *transaction) {
     //case 1 create new root
     if (old_node->IsRootPage()) {
       auto page = buffer_pool_manager_->NewPage(&root_page_id_);
       if (page == nullptr) {
         throw "not page can used in  InsertIntoParent";
       }
       assert(page->GetPinCount() == 1);
       auto root =reinterpret_cast<BPlusTreeInternalPage<KeyType, page_id_t,KeyComparator> *>(page->GetData());
       root->Init(root_page_id_,INVALID_PAGE_ID,internal_max_size_);
       root->PopulateNewRoot(old_node->GetPageId(), key, new_node->GetPageId());
   
       old_node->SetParentPageId(root_page_id_);
       new_node->SetParentPageId(root_page_id_);
   
       //TODO update to new root_page_id
       UpdateRootPageId(false);
       //TODO unpin
       buffer_pool_manager_->UnpinPage(new_node->GetPageId(), true);
       buffer_pool_manager_->UnpinPage(root->GetPageId(), true);
   
     }
   

   case2 -- 否则要递归上述的过程

   a. 先找分裂产生结点的父亲结点。如果可以直接插入则直接插入

   b. 否则需要对父结点在进行分裂，即递归调用。

  //case2  insert into parent
  else {
    auto parent_page = buffer_pool_manager_->FetchPage(old_node->GetParentPageId());
    if (parent_page == nullptr) {
      throw "no old_node parent page can used";
    }
    auto internal =reinterpret_cast<BPlusTreeInternalPage<KeyType, page_id_t,KeyComparator> *>(parent_page->GetData());
    // case 2.a insert directly
    if (internal->GetSize() < internal->GetMaxSize()) {
      internal->InsertNodeAfter(old_node->GetPageId(), key, new_node->GetPageId());
      new_node->SetParentPageId(internal->GetPageId());
      buffer_pool_manager_->UnpinPage(new_node->GetPageId(), true);
    }
    //case 2.b the parent node need to split
    else {
        internal->InsertNodeAfter(old_node->GetPageId(), key, new_node->GetPageId());
       auto new_internal =Split<BPlusTreeInternalPage<KeyType, page_id_t, KeyComparator>>(internal);
       new_internal->SetMaxSize(internal_max_size_);

       // TODO unpin and delete virtual page
       buffer_pool_manager_->UnpinPage(new_node->GetPageId(), true);
       buffer_pool_manager_->UnpinPage(internal->GetPageId(), false);
       InsertIntoParent(internal, new_internal->KeyAt(0), new_internal);
    }

    buffer_pool_manager_->UnpinPage(internal->GetPageId(), true);
  }
}

好了实验2的第一部分就到这里了。整个实验都已经写完啦。剩下就是优化代码，写博客记录了，所以实验2的第二部分也会很快更新的。这里面的代码不是很详细。等到第二部分写完之后，会一整个完全上传到GitHub上的。

附上一个pass的截图完成第一部分?
如果我们插入10、15、16、17、19、20那么我们用程序得到的结果如下

可以发现是完全正确的 ??

GitHub代码地址 https://github.com/JayL-zxl/CMU15-445Lab