record iterator

deps/common/lang/bitmap.cpp

@@ -36,9 +36,17 @@ int find_first_setted(char byte, int start)
   return -1;
 }
 
+Bitmap::Bitmap() : bitmap_(nullptr), size_(0)
+{}
 Bitmap::Bitmap(char *bitmap, int size) : bitmap_(bitmap), size_(size)
 {}
 
+void Bitmap::init(char *bitmap, int size)
+{
+  bitmap_ = bitmap;
+  size_ = size;
+}
+
 bool Bitmap::get_bit(int index)
 {
   char bits = bitmap_[index / 8];

deps/common/lang/bitmap.h

@@ -19,12 +19,17 @@ namespace common {
 
 class Bitmap {
 public:
+  Bitmap();
   Bitmap(char *bitmap, int size);
 
+  void init(char *bitmap, int size);
   bool get_bit(int index);
   void set_bit(int index);
   void clear_bit(int index);
 
+  /**
+   * @param start 从哪个位开始查找，start是包含在内的
+   */
   int next_unsetted_bit(int start);
   int next_setted_bit(int start);
 

src/observer/storage/common/record_manager.cpp

@@ -46,6 +46,37 @@ int page_header_size(int record_capacity)
   const int bitmap_size = page_bitmap_size(record_capacity);
   return align8(page_fix_size() + bitmap_size);
 }
+////////////////////////////////////////////////////////////////////////////////
+RecordPageIterator::RecordPageIterator()
+{}
+RecordPageIterator::~RecordPageIterator()
+{}
+
+void RecordPageIterator::init(RecordPageHandler &record_page_handler)
+{
+  record_page_handler_ = &record_page_handler;
+  page_num_ = record_page_handler.get_page_num();
+  bitmap_.init(record_page_handler.bitmap_, record_page_handler.page_header_->record_capacity);
+  next_slot_num_ = bitmap_.next_setted_bit(0);
+}
+
+bool RecordPageIterator::has_next()
+{
+  return -1 != next_slot_num_;
+}
+
+RC RecordPageIterator::next(Record &record)
+{
+  record.rid.page_num = page_num_;
+  record.rid.slot_num = next_slot_num_;
+  record.data = record_page_handler_->get_record_data(record.rid.slot_num);
+
+  if (next_slot_num_ >= 0) {
+    next_slot_num_ = bitmap_.next_setted_bit(next_slot_num_ + 1);
+  }
+  return record.rid.slot_num != -1 ? RC::SUCCESS : RC::RECORD_EOF;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 RecordPageHandler::~RecordPageHandler()
@@ -208,34 +239,6 @@ RC RecordPageHandler::get_record(const RID *rid, Record *rec)
   return RC::SUCCESS;
 }
 
-RC RecordPageHandler::get_first_record(Record *rec)
-{
-  rec->rid.slot_num = -1;
-  return get_next_record(rec);
-}
-
-RC RecordPageHandler::get_next_record(Record *rec)
-{
-  if (rec->rid.slot_num >= page_header_->record_capacity - 1) {
-    LOG_ERROR("Invalid slot_num:%d, exceed page's record capacity, page_num %d.",
-	      rec->rid.slot_num, frame_->page_num());
-    return RC::RECORD_EOF;
-  }
-
-  Bitmap bitmap(bitmap_, page_header_->record_capacity);
-  int index = bitmap.next_setted_bit(rec->rid.slot_num + 1);
-
-  if (index < 0) {
-    LOG_WARN("There is no empty slot on page -- page_num:%d.", frame_->page_num());
-    return RC::RECORD_EOF;
-  }
-
-  rec->rid.page_num = get_page_num();
-  rec->rid.slot_num = index;
-  rec->data = get_record_data(index);
-  return RC::SUCCESS;
-}
-
 PageNum RecordPageHandler::get_page_num() const
 {
   if (nullptr == page_header_) {
@@ -275,47 +278,25 @@ RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
 {
   RC ret = RC::SUCCESS;
   // 找到没有填满的页面
-  int page_count = 0;
-  if ((ret = disk_buffer_pool_->get_page_count(&page_count)) != RC::SUCCESS) {
-    LOG_ERROR("Failed to get page count while inserting record");
-    return ret;
-  }
-
-  PageNum current_page_num = record_page_handler_.get_page_num();
-  if (current_page_num < 0) {
-    if (page_count >= 2) {
-      // 当前buffer pool 有页面时才尝试加载第一页
-      // 参考diskBufferPool，有效page的pageNum从1开始
-      if ((ret = record_page_handler_.init(*disk_buffer_pool_, 1)) != RC::SUCCESS) {
-        LOG_ERROR("Failed to init record page handler, ret=%d", ret);
-        return ret;
-      }
-      current_page_num = record_page_handler_.get_page_num();
-    } else {
-      current_page_num = 0;
-    }
-  }
 
+  BufferPoolIterator bp_iterator;
+  bp_iterator.init(*disk_buffer_pool_);
+  RecordPageHandler record_page_handler;
   bool page_found = false;
-  for (int i = 0; i < page_count; i++) {
-    current_page_num = (current_page_num + i) % page_count;  // 从当前打开的页面开始查找
-    if (current_page_num == 0) {
-      continue;
-    }
-    if (current_page_num != record_page_handler_.get_page_num()) {
-      record_page_handler_.cleanup();
-      ret = record_page_handler_.init(*disk_buffer_pool_, current_page_num);
-      if (ret != RC::SUCCESS && ret != RC::BUFFERPOOL_INVALID_PAGE_NUM) {
-        LOG_ERROR("Failed to init record page handler. page number is %d. ret=%d:%s",
-		  current_page_num, ret, strrc(ret));
+  PageNum current_page_num = 0;
+  while (bp_iterator.has_next()) {
+    current_page_num = bp_iterator.next();
+    ret = record_page_handler.init(*disk_buffer_pool_, current_page_num);
+    if (ret != RC::SUCCESS) {
+      LOG_WARN("failed to init record page handler. page num=%d, rc=%d:%s", current_page_num, ret, strrc(ret));
       return ret;
     }
-    }
 
-    if (!record_page_handler_.is_full()) {
+    if (!record_page_handler.is_full()) {
       page_found = true;
       break;
     }
+    record_page_handler.cleanup();
   }
 
   // 找不到就分配一个新的页面
@@ -327,8 +308,7 @@ RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
     }
 
     current_page_num = frame->page_num();
-    record_page_handler_.cleanup();
-    ret = record_page_handler_.init_empty_page(*disk_buffer_pool_, current_page_num, record_size);
+    ret = record_page_handler.init_empty_page(*disk_buffer_pool_, current_page_num, record_size);
     if (ret != RC::SUCCESS) {
       LOG_ERROR("Failed to init empty page. ret:%d", ret);
       if (RC::SUCCESS != disk_buffer_pool_->unpin_page(frame)) {
@@ -341,15 +321,11 @@ RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
   }
 
   // 找到空闲位置
-  return record_page_handler_.insert_record(data, rid);
+  return record_page_handler.insert_record(data, rid);
 }
 
 RC RecordFileHandler::update_record(const Record *rec)
 {
-
-  if (record_page_handler_.get_page_num() == rec->rid.page_num) {
-    return record_page_handler_.update_record(rec);
-  }
   RC ret;
   RecordPageHandler page_handler;
   if ((ret = page_handler.init(*disk_buffer_pool_, rec->rid.page_num)) != RC::SUCCESS) {
@@ -362,10 +338,6 @@ RC RecordFileHandler::update_record(const Record *rec)
 
 RC RecordFileHandler::delete_record(const RID *rid)
 {
-  if (record_page_handler_.get_page_num() == rid->page_num) {
-    return record_page_handler_.delete_record(rid);
-  }
-
   RC ret = RC::SUCCESS;
   RecordPageHandler page_handler;
   if ((ret != page_handler.init(*disk_buffer_pool_, rid->page_num)) != RC::SUCCESS) {
@@ -384,10 +356,6 @@ RC RecordFileHandler::get_record(const RID *rid, Record *rec)
     return RC::INVALID_ARGUMENT;
   }
 
-  if (record_page_handler_.get_page_num() == rid->page_num) {
-    return record_page_handler_.get_record(rid, rec);
-  }
-
   RecordPageHandler page_handler;
   if ((ret != page_handler.init(*disk_buffer_pool_, rid->page_num)) != RC::SUCCESS) {
     LOG_ERROR("Failed to init record page handler.page number=%d", rid->page_num);
@@ -405,83 +373,85 @@ RC RecordFileScanner::open_scan(DiskBufferPool &buffer_pool, ConditionFilter *co
 
   disk_buffer_pool_ = &buffer_pool;
 
+  RC rc = bp_iterator_.init(buffer_pool);
+  if (rc != RC::SUCCESS) {
+    LOG_WARN("failed to init bp iterator. rc=%d:%s", rc, strrc(rc));
+    return rc;
+  }
   condition_filter_ = condition_filter;
-  return RC::SUCCESS;
+
+  rc = fetch_next_record();
+  if (rc == RC::RECORD_EOF) {
+    rc = RC::SUCCESS;
+  }
+  return rc;
 }
 
-RC RecordFileScanner::close_scan()
+RC RecordFileScanner::fetch_next_record()
 {
-  if (disk_buffer_pool_ != nullptr) {
-    disk_buffer_pool_ = nullptr;
+  RC rc = RC::SUCCESS;
+  while (bp_iterator_.has_next()) {
+    PageNum page_num = bp_iterator_.next();
+    record_page_handler_.cleanup();
+    rc = record_page_handler_.init(*disk_buffer_pool_, page_num);
+    if (rc != RC::SUCCESS) {
+      LOG_WARN("failed to init record page handler. rc=%d:%s", rc, strrc(rc));
+      return rc;
     }
 
-  if (condition_filter_ != nullptr) {
-    condition_filter_ = nullptr;
+    record_page_iterator_.init(record_page_handler_);
+    rc = fetch_record_in_page();
+    if (rc == RC::SUCCESS || rc != RC::RECORD_EOF) {
+      return rc;
     }
-
-  return RC::SUCCESS;
-}
-
-RC RecordFileScanner::get_first_record(Record *rec)
-{
-  rec->rid.page_num = 1;  // from 1 参考DiskBufferPool
-  rec->rid.slot_num = -1;
-  // rec->valid = false;
-  return get_next_record(rec);
+  }
+  next_record_.rid.slot_num = -1;
+  return RC::RECORD_EOF;
 }
 
-RC RecordFileScanner::get_next_record(Record *rec)
+RC RecordFileScanner::fetch_record_in_page()
 {
-  if (nullptr == disk_buffer_pool_) {
-    LOG_ERROR("Scanner has been closed.");
-    return RC::RECORD_CLOSED;
+  RC rc = RC::SUCCESS;
+  while (record_page_iterator_.has_next()) {
+    rc = record_page_iterator_.next(next_record_);
+    if (rc != RC::SUCCESS) {
+      return rc;
     }
 
-  RC ret = RC::SUCCESS;
-  Record current_record = *rec;
-
-  int page_count = 0;
-  if ((ret = disk_buffer_pool_->get_page_count(&page_count)) != RC::SUCCESS) {
-    LOG_ERROR("Failed to get page count while getting next record.");
-    return RC::RECORD_EOF;
+    if (condition_filter_ == nullptr || condition_filter_->filter(next_record_)) {
+      return rc;
     }
-
-  if (1 == page_count) {
-    return RC::RECORD_EOF;
   }
 
-  while (current_record.rid.page_num < page_count) {
+  next_record_.rid.slot_num = -1;
+  return RC::RECORD_EOF;
+}
 
-    if (current_record.rid.page_num != record_page_handler_.get_page_num()) {
-      record_page_handler_.cleanup();
-      ret = record_page_handler_.init(*disk_buffer_pool_, current_record.rid.page_num);
-      if (ret != RC::SUCCESS && ret != RC::BUFFERPOOL_INVALID_PAGE_NUM) {
-        LOG_ERROR("Failed to init record page handler. page num=%d", current_record.rid.page_num);
-        return ret;
+RC RecordFileScanner::close_scan()
+{
+  if (disk_buffer_pool_ != nullptr) {
+    disk_buffer_pool_ = nullptr;
   }
 
-      if (RC::BUFFERPOOL_INVALID_PAGE_NUM == ret) {
-        current_record.rid.page_num++;
-        current_record.rid.slot_num = -1;
-        continue;
-      }
+  if (condition_filter_ != nullptr) {
+    condition_filter_ = nullptr;
   }
 
-    ret = record_page_handler_.get_next_record(&current_record);
-    if (RC::SUCCESS == ret) {
-      if (condition_filter_ == nullptr || condition_filter_->filter(current_record)) {
-        break;  // got one
-      }
-    } else if (RC::RECORD_EOF == ret) {
-      current_record.rid.page_num++;
-      current_record.rid.slot_num = -1;
-    } else {
-      break;  // ERROR
-    }
-  }
+  return RC::SUCCESS;
+}
 
-  if (RC::SUCCESS == ret) {
-    *rec = current_record;
+bool RecordFileScanner::has_next()
+{
+  return next_record_.rid.slot_num != -1;
+}
+
+RC RecordFileScanner::next(Record &record)
+{
+  record = next_record_;
+  
+  RC rc = fetch_next_record();
+  if (rc == RC::RECORD_EOF) {
+    rc = RC::SUCCESS;
   }
-  return ret;
+  return rc;
 }

src/observer/storage/common/record_manager.h

@@ -17,6 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include <sstream>
 #include <limits>
 #include "storage/default/disk_buffer_pool.h"
+#include "common/lang/bitmap.h"
 
 typedef int32_t SlotNum;
 
@@ -95,6 +96,25 @@ struct Record {
   char *data;  // record's data
 };
 
+class RecordPageHandler;
+class RecordPageIterator
+{
+public:
+  RecordPageIterator();
+  ~RecordPageIterator();
+
+  void init(RecordPageHandler &record_page_handler);
+
+  bool has_next();
+  RC   next(Record &record);
+
+private:
+  RecordPageHandler *record_page_handler_ = nullptr;
+  PageNum page_num_ = BP_INVALID_PAGE_NUM;
+  common::Bitmap  bitmap_;
+  SlotNum next_slot_num_ = 0;
+};
+
 class RecordPageHandler {
 public:
   RecordPageHandler() = default;
@@ -122,8 +142,6 @@ public:
   RC delete_record(const RID *rid);
 
   RC get_record(const RID *rid, Record *rec);
-  RC get_first_record(Record *rec);
-  RC get_next_record(Record *rec);
 
   PageNum get_page_num() const;
 
@@ -140,6 +158,9 @@ protected:
   Frame *frame_ = nullptr;
   PageHeader *page_header_ = nullptr;
   char *bitmap_ = nullptr;
+
+private:
+  friend class RecordPageIterator;
 };
 
 class RecordFileHandler {
@@ -184,7 +205,6 @@ public:
 
 private:
   DiskBufferPool *disk_buffer_pool_ = nullptr;
-  RecordPageHandler record_page_handler_;  // 目前只有insert record使用
 };
 
 class RecordFileScanner {
@@ -193,36 +213,29 @@ public:
 
   /**
    * 打开一个文件扫描。
-   * 本函数利用从第二个参数开始的所有输入参数初始化一个由参数rmFileScan指向的文件扫描结构，
-   * 在使用中，用户应先调用此函数初始化文件扫描结构，
-   * 然后再调用GetNextRec函数来逐个返回文件中满足条件的记录。
-   * 如果条件数量conNum为0，则意味着检索文件中的所有记录。
    * 如果条件不为空，则要对每条记录进行条件比较，只有满足所有条件的记录才被返回
    */
   RC open_scan(DiskBufferPool &buffer_pool, ConditionFilter *condition_filter);
 
   /**
    * 关闭一个文件扫描，释放相应的资源
-   * @return
    */
   RC close_scan();
 
-  RC get_first_record(Record *rec);
-
-  /**
-   * 获取下一个符合扫描条件的记录。
-   * 如果该方法成功，返回值rec应包含记录副本及记录标识符。
-   * 如果没有发现满足扫描条件的记录，则返回RM_EOF
-   * @param rec 上一条记录。如果为NULL，就返回第一条记录
-   * @return
-   */
-  RC get_next_record(Record *rec);
+  bool has_next();
+  RC   next(Record &record);
 
+private:
+  RC fetch_next_record();
+  RC fetch_record_in_page();
 private:
   DiskBufferPool *disk_buffer_pool_ = nullptr;
 
+  BufferPoolIterator bp_iterator_;
   ConditionFilter *condition_filter_ = nullptr;
   RecordPageHandler record_page_handler_;
+  RecordPageIterator record_page_iterator_;
+  Record next_record_;
 };
 
 #endif  //__OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_

src/observer/storage/common/table.cpp

@@ -390,8 +390,8 @@ RC Table::scan_record(
   return scan_record(trx, filter, limit, (void *)&adapter, scan_record_reader_adapter);
 }
 
-RC Table::scan_record(
-    Trx *trx, ConditionFilter *filter, int limit, void *context, RC (*record_reader)(Record *record, void *context))
+RC Table::scan_record(Trx *trx, ConditionFilter *filter, int limit, void *context,
+		      RC (*record_reader)(Record *record, void *context))
 {
   if (nullptr == record_reader) {
     return RC::INVALID_ARGUMENT;
@@ -420,8 +420,12 @@ RC Table::scan_record(
 
   int record_count = 0;
   Record record;
-  rc = scanner.get_first_record(&record);
-  for (; RC::SUCCESS == rc && record_count < limit; rc = scanner.get_next_record(&record)) {
+  while (scanner.has_next()) {
+    rc = scanner.next(record);
+    if (rc != RC::SUCCESS) {
+      LOG_WARN("failed to fetch next record. rc=%d:%s", rc, strrc(rc));
+      return rc;
+    }
     if (trx == nullptr || trx->is_visible(this, &record)) {
       rc = record_reader(&record, context);
       if (rc != RC::SUCCESS) {
@@ -827,7 +831,7 @@ IndexScanner *Table::find_index_for_scan(const ConditionFilter *filter)
 
 RC Table::sync()
 {
-  RC rc = data_buffer_pool_->purge_all_pages();
+  RC rc = data_buffer_pool_->flush_all_pages();
   if (rc != RC::SUCCESS) {
     LOG_ERROR("Failed to flush table's data pages. table=%s, rc=%d:%s", name(), rc, strrc(rc));
     return rc;

src/observer/storage/default/disk_buffer_pool.cpp

@@ -88,6 +88,43 @@ std::list<Frame *> BPFrameManager::find_list(int file_desc)
   return find_all(match_file, &file_desc);
 }
 
+////////////////////////////////////////////////////////////////////////////////
+BufferPoolIterator::BufferPoolIterator()
+{}
+BufferPoolIterator::~BufferPoolIterator()
+{}
+RC BufferPoolIterator::init(DiskBufferPool &bp, PageNum start_page /* = 0 */)
+{
+  bitmap_.init(bp.file_header_->bitmap, bp.file_header_->page_count);
+  if (start_page <= 0) {
+    current_page_num_ = 0;
+  } else {
+    current_page_num_ = start_page;
+  }
+  return RC::SUCCESS;
+}
+
+bool BufferPoolIterator::has_next()
+{
+  return bitmap_.next_setted_bit(current_page_num_) != -1;
+}
+
+PageNum BufferPoolIterator::next()
+{
+  PageNum next_page = bitmap_.next_setted_bit(current_page_num_);
+  if (next_page != -1) {
+    current_page_num_ = next_page;
+  }
+  return next_page;
+}
+
+RC BufferPoolIterator::reset()
+{
+  current_page_num_ = 0;
+  return RC::SUCCESS;
+}
+
+////////////////////////////////////////////////////////////////////////////////
 DiskBufferPool::DiskBufferPool(BufferPoolManager &bp_manager, BPFrameManager &frame_manager)
   : bp_manager_(bp_manager), frame_manager_(frame_manager)
 {
@@ -390,6 +427,19 @@ RC DiskBufferPool::flush_page(Frame &frame)
   return RC::SUCCESS;
 }
 
+RC DiskBufferPool::flush_all_pages()
+{
+  std::list<Frame *> used = frame_manager_.find_list(file_desc_);
+  for (Frame *frame : used) {
+    RC rc = flush_page(*frame);
+    if (rc != RC::SUCCESS) {
+      LOG_WARN("failed to flush all pages");
+      return rc;
+    }
+  }
+  return RC::SUCCESS;
+}
+
 RC DiskBufferPool::allocate_frame(Frame **buffer)
 {
   Frame *frame = frame_manager_.alloc();

src/observer/storage/default/disk_buffer_pool.h

@@ -26,6 +26,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "rc.h"
 #include "common/mm/mem_pool.h"
+#include "common/lang/bitmap.h"
 
 typedef int32_t PageNum;
 
@@ -56,6 +57,11 @@ struct BPFileHeader {
   int32_t page_count;        //! 当前文件一共有多少个页面
   int32_t allocated_pages;   //! 已经分配了多少个页面
   char    bitmap[0];         //! 页面分配位图, 第0个页面(就是当前页面)，总是1
+
+  /**
+   * 能够分配的最大的页面个数，即bitmap的字节数 乘以8
+   */
+  static const int MAX_PAGE_NUM = (BP_PAGE_DATA_SIZE - sizeof(page_count) - sizeof(allocated_pages)) * 8;
 };
 
 class Frame
@@ -111,7 +117,8 @@ private:
   Page          page_;
 };
 
-class BPFrameManager : public common::MemPoolSimple<Frame> {
+class BPFrameManager : public common::MemPoolSimple<Frame>
+{
 public:
   BPFrameManager(const char *tag);
 
@@ -119,10 +126,30 @@ public:
 
   std::list<Frame *> find_list(int file_desc);
 
+  /**
+   * 如果不能从空闲链表中分配新的页面，就使用这个接口，
+   * 尝试从pin count=0的页面中淘汰一个
+   */
   Frame *begin_purge();
 };
 
-class DiskBufferPool {
+class BufferPoolIterator
+{
+public:
+  BufferPoolIterator();
+  ~BufferPoolIterator();
+
+  RC init(DiskBufferPool &bp, PageNum start_page = 0);
+  bool has_next();
+  PageNum next();
+  RC reset();
+private:
+  common::Bitmap   bitmap_;
+  PageNum  current_page_num_ = -1;
+};
+
+class DiskBufferPool
+{
 public:
   DiskBufferPool(BufferPoolManager &bp_manager, BPFrameManager &frame_manager);
   ~DiskBufferPool();
@@ -133,8 +160,7 @@ public:
   RC create_file(const char *file_name);
 
   /**
-   * 根据文件名打开一个分页文件，返回文件ID
-   * @return
+   * 根据文件名打开一个分页文件
    */
   RC open_file(const char *file_name);
 
@@ -145,7 +171,6 @@ public:
 
   /**
    * 根据文件ID和页号获取指定页面到缓冲区，返回页面句柄指针。
-   * @return
    */
   RC get_this_page(PageNum page_num, Frame **frame);
 
@@ -180,6 +205,10 @@ public:
    */
   RC get_page_count(int *page_count);
 
+  /**
+   * 检查是否所有页面都是pin count == 0状态(除了第1个页面)
+   * 调试使用
+   */
   RC check_all_pages_unpinned();
 
   int file_desc() const;
@@ -189,6 +218,11 @@ public:
    */
   RC flush_page(Frame &frame);
 
+  /**
+   * 刷新所有页面到磁盘，即使pin count不是0
+   */
+  RC flush_all_pages();
+
 protected:
   RC allocate_frame(Frame **buf);
 
@@ -211,6 +245,9 @@ private:
   Frame *            hdr_frame_ = nullptr;
   BPFileHeader *     file_header_ = nullptr;
   std::set<PageNum>  disposed_pages;
+
+private:
+  friend class BufferPoolIterator;
 };
 
 class BufferPoolManager

src/observer/storage/index/bplus_tree.cpp

@@ -751,7 +751,7 @@ bool InternalIndexNodeHandler::validate(const KeyComparator &comparator, DiskBuf
 
 RC BplusTreeHandler::sync()
 {
-  return disk_buffer_pool_->purge_all_pages();
+  return disk_buffer_pool_->flush_all_pages();
 }
 
 RC BplusTreeHandler::create(const char *file_name, AttrType attr_type, int attr_length,

unitest/record_manager_test.cpp

+/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+//
+// Created by wangyunlai.wyl on 2022
+//
+
+#include <string.h>
+#include <sstream>
+
+#include "gtest/gtest.h"
+#include "storage/default/disk_buffer_pool.h"
+#include "storage/common/record_manager.h"
+
+using namespace common;
+
+TEST(test_record_page_handler, test_record_page_handler)
+{
+  const char *record_manager_file = "record_manager.bp";
+  ::remove(record_manager_file);
+
+  BufferPoolManager *bpm = new BufferPoolManager();
+  DiskBufferPool *bp = nullptr;
+  RC rc = bpm->create_file(record_manager_file);
+  ASSERT_EQ(rc, RC::SUCCESS);
+  
+  rc = bpm->open_file(record_manager_file, bp);
+  ASSERT_EQ(rc, RC::SUCCESS);
+
+  Frame *frame = nullptr;
+  rc = bp->allocate_page(&frame);
+  ASSERT_EQ(rc, RC::SUCCESS);
+
+  const int record_size = 8;
+  RecordPageHandler record_page_handle;
+  rc = record_page_handle.init_empty_page(*bp, frame->page_num(), record_size);
+  ASSERT_EQ(rc, RC::SUCCESS);
+
+  RecordPageIterator iterator;
+  iterator.init(record_page_handle);
+
+  int count = 0;
+  Record record;
+  while (iterator.has_next()) {
+    count++;
+    rc = iterator.next(record);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+  ASSERT_EQ(count, 0);
+
+  char buf[record_size];
+  RID rid;
+  rid.page_num = 100;
+  rid.slot_num = 100;
+  rc = record_page_handle.insert_record(buf, &rid);
+  ASSERT_EQ(rc, RC::SUCCESS);
+
+  count = 0;
+  iterator.init(record_page_handle);
+  while (iterator.has_next()) {
+    count++;
+    rc = iterator.next(record);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+  ASSERT_EQ(count, 1);
+
+  for (int i = 0; i < 10; i++) {
+    rid.page_num = i;
+    rid.slot_num = i;
+    rc = record_page_handle.insert_record(buf, &rid);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+
+  count = 0;
+  iterator.init(record_page_handle);
+  while (iterator.has_next()) {
+    count++;
+    rc = iterator.next(record);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+  ASSERT_EQ(count, 11);
+
+  for (int i = 0; i < 5; i++) {
+    rid.page_num = i * 2;
+    rid.slot_num = i * 2;
+    rc = record_page_handle.delete_record(&rid);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+
+  count = 0;
+  iterator.init(record_page_handle);
+  while (iterator.has_next()) {
+    count++;
+    rc = iterator.next(record);
+    ASSERT_EQ(rc, RC::SUCCESS);
+  }
+  ASSERT_EQ(count, 6);
+}
+
+int main(int argc, char **argv)
+{
+  // 分析gtest程序的命令行参数
+  testing::InitGoogleTest(&argc, argv);
+
+  // 调用RUN_ALL_TESTS()运行所有测试用例
+  // main函数返回RUN_ALL_TESTS()的运行结果
+  return RUN_ALL_TESTS();
+}