rewrite btree twice

src/observer/storage/common/bplus_tree_index.cpp

@@ -100,17 +100,16 @@ RC BplusTreeIndex::delete_entry(const char *record, const RID *rid)
   return index_handler_.delete_entry(record + field_meta_.offset(), rid);
 }
 
-IndexScanner *BplusTreeIndex::create_scanner(CompOp comp_op, const char *value)
+IndexScanner *BplusTreeIndex::create_scanner(const char *left_key, bool left_inclusive,
+					     const char *right_key, bool right_inclusive)
 {
-  BplusTreeScanner *bplus_tree_scanner = new BplusTreeScanner(index_handler_);
-  RC rc = bplus_tree_scanner->open(comp_op, value);
+  BplusTreeIndexScanner *index_scanner = new BplusTreeIndexScanner(index_handler_);
+  RC rc = index_scanner->open(left_key, left_inclusive, right_key, right_inclusive);
   if (rc != RC::SUCCESS) {
-    LOG_WARN("Failed to open index scanner. file_id:%d, rc=%d:%s", index_handler_.get_file_id(), rc, strrc(rc));
-    delete bplus_tree_scanner;
+    LOG_WARN("failed to open index scanner. rc=%d:%s", rc, strrc(rc));
+    delete index_scanner;
     return nullptr;
   }
-
-  BplusTreeIndexScanner *index_scanner = new BplusTreeIndexScanner(bplus_tree_scanner);
   return index_scanner;
 }
 
@@ -120,22 +119,26 @@ RC BplusTreeIndex::sync()
 }
 
 ////////////////////////////////////////////////////////////////////////////////
-BplusTreeIndexScanner::BplusTreeIndexScanner(BplusTreeScanner *tree_scanner) : tree_scanner_(tree_scanner)
+BplusTreeIndexScanner::BplusTreeIndexScanner(BplusTreeHandler &tree_handler) : tree_scanner_(tree_handler)
 {}
 
 BplusTreeIndexScanner::~BplusTreeIndexScanner() noexcept
 {
-  tree_scanner_->close();
-  delete tree_scanner_;
+  tree_scanner_.close();
+}
+
+RC BplusTreeIndexScanner::open(const char *left_key, bool left_inclusive, const char *right_key, bool right_inclusive)
+{
+  return tree_scanner_.open(left_key, left_inclusive, right_key, right_inclusive);
 }
 
 RC BplusTreeIndexScanner::next_entry(RID *rid)
 {
-  return tree_scanner_->next_entry(rid);
+  return tree_scanner_.next_entry(rid);
 }
 
 RC BplusTreeIndexScanner::destroy()
 {
   delete this;
   return RC::SUCCESS;
-}
\ No newline at end of file
+}

src/observer/storage/common/bplus_tree_index.h

@@ -30,7 +30,11 @@ public:
   RC insert_entry(const char *record, const RID *rid) override;
   RC delete_entry(const char *record, const RID *rid) override;
 
-  IndexScanner *create_scanner(CompOp comp_op, const char *value) override;
+  /**
+   * 扫描指定范围的数据
+   */
+  IndexScanner *create_scanner(const char *left_key, bool left_inclusive,
+			       const char *right_key, bool right_inclusive) override;
 
   RC sync() override;
 
@@ -41,14 +45,15 @@ private:
 
 class BplusTreeIndexScanner : public IndexScanner {
 public:
-  BplusTreeIndexScanner(BplusTreeScanner *tree_scanner);
+  BplusTreeIndexScanner(BplusTreeHandler &tree_handle);
   ~BplusTreeIndexScanner() noexcept override;
 
   RC next_entry(RID *rid) override;
   RC destroy() override;
 
+  RC open(const char *left_key, bool left_inclusive, const char *right_key, bool right_inclusive);
 private:
-  BplusTreeScanner *tree_scanner_;
+  BplusTreeScanner tree_scanner_;
 };
 
 #endif  //__OBSERVER_STORAGE_COMMON_BPLUS_TREE_INDEX_H_

src/observer/storage/common/index.h

@@ -46,7 +46,8 @@ public:
   virtual RC insert_entry(const char *record, const RID *rid) = 0;
   virtual RC delete_entry(const char *record, const RID *rid) = 0;
 
-  virtual IndexScanner *create_scanner(CompOp comp_op, const char *value) = 0;
+  virtual IndexScanner *create_scanner(const char *left_key, bool left_inclusive,
+				       const char *right_key, bool right_inclusive) = 0;
 
   virtual RC sync() = 0;
 
@@ -63,8 +64,12 @@ public:
   IndexScanner() = default;
   virtual ~IndexScanner() = default;
 
+  /**
+   * 遍历元素数据
+   * 如果没有更多的元素，返回RECORD_EOF
+   */
   virtual RC next_entry(RID *rid) = 0;
   virtual RC destroy() = 0;
 };
 
-#endif  // __OBSERVER_STORAGE_COMMON_INDEX_H_
\ No newline at end of file
+#endif  // __OBSERVER_STORAGE_COMMON_INDEX_H_

src/observer/storage/common/record_manager.h

@@ -17,7 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include <sstream>
 #include "storage/default/disk_buffer_pool.h"
 
-typedef int SlotNum;
+typedef int32_t SlotNum;
 
 class ConditionFilter;
 
@@ -48,6 +48,11 @@ struct RID {
     return page_num == other.page_num && slot_num == other.slot_num;
   }
 
+  bool operator!=(const RID &other) const
+  {
+    return !(*this == other);
+  }
+
   static int compare(const RID *rid1, const RID *rid2)
   {
     int page_diff = rid1->page_num - rid2->page_num;
@@ -57,6 +62,22 @@ struct RID {
       return rid1->slot_num - rid2->slot_num;
     }
   }
+
+  /**
+   * 返回一个不可能出现的最小的RID
+   * 虽然page num 0和slot num 0都是合法的，但是page num 0通常用于存放meta数据，所以对数据部分来说都是
+   * 不合法的. 这里在bplus tree中查找时会用到。
+   */
+  static RID *min()
+  {
+    static RID rid{0, 0};
+    return &rid;
+  }
+  static RID *max()
+  {
+    static RID rid{std::numeric_limits<PageNum>::max(), std::numeric_limits<SlotNum>::max()};
+    return &rid;
+  }
 };
 
 class RidDigest {
@@ -222,4 +243,4 @@ private:
   RecordPageHandler record_page_handler_;
 };
 
-#endif  //__OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_
\ No newline at end of file
+#endif  //__OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_

src/observer/storage/common/table.cpp

@@ -768,7 +768,43 @@ IndexScanner *Table::find_index_for_scan(const DefaultConditionFilter &filter)
     return nullptr;
   }
 
-  return index->create_scanner(filter.comp_op(), (const char *)value_cond_desc->value);
+  const char *left_key = nullptr;
+  const char *right_key = nullptr;
+  bool left_inclusive = false;
+  bool right_inclusive = false;
+  switch (filter.comp_op()) {
+  case EQUAL_TO: {
+    left_key = (const char *)value_cond_desc->value;
+    right_key = (const char *)value_cond_desc->value;
+    left_inclusive = true;
+    right_inclusive = true;
+  }
+    break;
+  case LESS_EQUAL: {
+    right_key = (const char *)value_cond_desc->value;
+    right_inclusive = true;
+  }
+    break;
+  case GREAT_EQUAL: {
+    left_key = (const char *)value_cond_desc->value;
+    left_inclusive = true;
+  }
+    break;
+  case LESS_THAN: {
+    right_key = (const char *)value_cond_desc->value;
+    right_inclusive = false;
+  }
+    break;
+  case GREAT_THAN: {
+    left_key = (const char *)value_cond_desc->value;
+    left_inclusive = false;
+  }
+    break;
+  default: {
+    return nullptr;
+  }
+  }
+  return index->create_scanner(left_key, left_inclusive, right_key, right_inclusive);
 }
 
 IndexScanner *Table::find_index_for_scan(const ConditionFilter *filter)

src/observer/storage/default/disk_buffer_pool.cpp

@@ -478,7 +478,7 @@ RC DiskBufferPool::purge_page(int file_id, PageNum page_num)
 RC DiskBufferPool::purge_page(Frame *buf)
 {
   if (buf->pin_count > 0) {
-    LOG_INFO("Begin to free page %d of %d, but it's pinned, pin_count:%d.",
+    LOG_INFO("Begin to free page %d of %d(file id), but it's pinned, pin_count:%d.",
         buf->page.page_num,
         buf->file_desc,
         buf->pin_count);
@@ -488,12 +488,12 @@ RC DiskBufferPool::purge_page(Frame *buf)
   if (buf->dirty) {
     RC rc = flush_page(buf);
     if (rc != RC::SUCCESS) {
-      LOG_WARN("Failed to flush page %d of %d during purge page.", buf->page.page_num, buf->file_desc);
+      LOG_WARN("Failed to flush page %d of %d(file desc) during purge page.", buf->page.page_num, buf->file_desc);
       return rc;
     }
   }
 
-  LOG_DEBUG("Successfully purge frame =%p, page %d of %d", buf, buf->page.page_num, buf->file_desc);
+  LOG_DEBUG("Successfully purge frame =%p, page %d of %d(file desc)", buf, buf->page.page_num, buf->file_desc);
   bp_manager_.free(buf);
   return RC::SUCCESS;
 }
@@ -533,7 +533,8 @@ RC DiskBufferPool::purge_all_pages(BPFileHandle *file_handle)
   for (std::list<Frame *>::iterator it = used.begin(); it != used.end(); ++it) {
     Frame *frame = *it;
     if (frame->pin_count > 0) {
-      LOG_WARN("The page has been pinned, file_id:%d, pagenum:%d", frame->file_desc, frame->page.page_num);
+      LOG_WARN("The page has been pinned, file_desc:%d, pagenum:%d, pin_count=%d",
+	       frame->file_desc, frame->page.page_num, frame->pin_count);
       continue;
     }
     if (frame->dirty) {
@@ -548,6 +549,29 @@ RC DiskBufferPool::purge_all_pages(BPFileHandle *file_handle)
   return RC::SUCCESS;
 }
 
+RC DiskBufferPool::check_all_pages_unpinned(int file_id)
+{
+  RC rc = check_file_id(file_id);
+  if (rc != RC::SUCCESS) {
+    LOG_ERROR("Failed to flush pages due to invalid file_id %d", file_id);
+    return rc;
+  }
+
+  BPFileHandle *file_handle = open_list_[file_id];
+  std::list<Frame *> frames = bp_manager_.find_list(file_handle->file_desc);
+  for (auto & frame : frames) {
+    if (frame->page.page_num == 0 && frame->pin_count > 1) {
+      LOG_WARN("This page has been pinned. file id=%d, page num:%d, pin count=%d",
+	       file_id, frame->page.page_num, frame->pin_count);
+    } else if (frame->page.page_num != 0 && frame->pin_count > 0) {
+      LOG_WARN("This page has been pinned. file id=%d, page num:%d, pin count=%d",
+	       file_id, frame->page.page_num, frame->pin_count);
+    }
+  }
+  LOG_INFO("all pages have been checked of file id %d", file_id);
+  return RC::SUCCESS;
+}
+
 RC DiskBufferPool::flush_page(Frame *frame)
 {
   // The better way is use mmap the block into memory,

src/observer/storage/default/disk_buffer_pool.h

@@ -69,6 +69,16 @@ typedef struct BPPageHandle {
   BPPageHandle() : open(false), frame(nullptr)
   {}
 
+  PageNum page_num() const {
+    return frame->page.page_num;
+  }
+  void mark_dirty() {
+    this->frame->dirty = true;
+  }
+
+  char *data() {
+    return this->frame->page.data;
+  }
   bool open;
   Frame *frame;
 } BPPageHandle;
@@ -195,6 +205,8 @@ public:
 
   RC purge_all_pages(int file_id);
 
+  RC check_all_pages_unpinned(int file_id);
+
 protected:
   RC allocate_page(Frame **buf);