存储管理

任务一 缓冲池管理器

1.1 磁盘存储管理器

本任务要求补全DiskManager类，要求实现读写指定页面、以及对文件进行操作。

前置知识： linux 系统下的 write ，open ，read ，unlink ，close ，leek 等函数。

注意事项： 一些系统函数比如 open 可能需要权限，所以最好在 root 用户下进行实验。还有对于什么时候抛出什么异常，文档里没细说，最后测试的时候根据测试文件的代码一个一个补充即可。

write_page 实现：

文档提示里已经告诉了我们每页的字节大小是个已定义的常数 PAGE_SIZE ，通过 page_no * PAGE_SIZE 即可确定要写入的位置的字节偏移量，再通过 leek 定位， 最后调用 write 函数进行写入即可。

void DiskManager::write_page(int fd, page_id_t page_no, const char *offset, int num_bytes) {
    off_t off = page_no * PAGE_SIZE;
    lseek(fd, off, SEEK_SET);
    int num = write(fd, offset, num_bytes);
    if (num != num_bytes) {
        throw InternalError("DiskManager::write_page Error");
    }
}

read_page 实现：

思路和上一个类似，让文件定位到要读的位置后，再调用 read 函数即可。

void DiskManager::read_page(int fd, page_id_t page_no, char *offset, int num_bytes) {
    off_t off = page_no * PAGE_SIZE;
    lseek(fd, off, SEEK_SET);
    int num = read(fd, offset, num_bytes);
    if (num != num_bytes) {
        throw InternalError("DiskManager::read_page Error");
    }
}

create_file 实现：

需要判断文件是否存在，当然我对于我的代码还有疑问，主要在于用 open 去新建一个文件的细节，虽然它确实通过了测试，但 gpt 告诉我 open(path.c_str(), O_CREAT); 没有指定打开模式的话，是不会创建文件的。

upd: 亲自去做了测试，发现参数只设置为 O_CREAT ，当 open 的文件不存在时，依旧会新建一个文件并打开，也就是返回值不是 -1 。 gpt 给的解释是这是未定义行为，libc 实现将其理解为了 O_CREAT | O_RDONLY 。

void DiskManager::create_file(const std::string &path) {
    if (!is_file(path)) {
        open(path.c_str(), O_CREAT);
        close(fd);
    } else
        throw FileExistsError("");
}

open_file 实现：

从这里开始我们就必须去维护打开文件的集合，事实上 gpt 告诉我 linux 是允许多次打开一个文件的，彼此之间相互独立，但我们的数据库似乎并不允许， 所以需要我们维护当前打开的文件集合，可以看到 disk_manager.h 文件中有如下定义：

// 文件打开列表，用于记录文件是否被打开
std::unordered_map<std::string, int> path2fd_;  //<Page文件磁盘路径,Page fd>哈希表
std::unordered_map<int, std::string> fd2path_;  //<Page fd,Page文件磁盘路径>哈希表

所以我们直接用这两个哈希表去维护就好了。

int DiskManager::open_file(const std::string &path) {
    if (!is_file(path)) {
        throw FileNotFoundError("");
    }
    if (!path2fd_.count(path)) {
        int fd = open(path.c_str(), O_RDWR);
        path2fd_[path] = fd;
        fd2path_[fd] = path;
    }
    return path2fd_[path];
}

close_file 实现：

注意维护打开文件集合即可。

void DiskManager::close_file(int fd) {
    if (fd2path_.count(fd)) {
        close(fd);
        path2fd_.erase(fd2path_[fd]);
        fd2path_.erase(fd);
    }
}

destory_file 实现：

这一关只需要注意当我们要删除的文件还在打开状态时，要先关闭再删除。

void DiskManager::destroy_file(const std::string &path) {
    if (!is_file(path)) throw FileNotFoundError("");

    if (path2fd_.count(path)) {
        close_file(path2fd_[path]);
    }

    unlink(path.c_str());
}

1.2 缓冲池替换策略

本任务要求补全Replacer类，并且实现的替换策略为最近最少使用（LRU）算法。

前置知识： 了解页(page) ，帧(frame)，pin，unpin 的概率。 注意事项： 每个函数都需要有 std::scoped_lock lock{latch_}; 这条语句，作用大概是用于避免死锁，但具体原理我也不太清楚。

事实上只要知道上述概念再结合测试就能很快完成。

victim 实现：

头文件中给我们提供了 LRUlisk_ 这个链表维护没有被固定的帧的最近访问时间顺序，其中链头是最先访问的。 然后还提供了 LRUhash_ 这个哈希表便于快速定位到某帧对应的迭代器。

根据上述内容可知：弹出尾部元素即可。

bool LRUReplacer::victim(frame_id_t* frame_id) {
    // C++17 std::scoped_lock
    // 它能够避免死锁发生，其构造函数能够自动进行上锁操作，析构函数会对互斥量进行解锁操作，保证线程安全。
    std::scoped_lock lock{latch_};  //  如果编译报错可以替换成其他lock

    if (!LRUlist_.empty()) {
        *frame_id = LRUlist_.back();
        LRUhash_.erase(LRUlist_.back());
        LRUlist_.pop_back();
    } else
        return false;
    return true;
}

pin 实现：

该函数用于固定一个帧，避免多线程操作中一边还在访问，另一边已经将其替换了。

实现上其实就是将其从链表和哈希表中删去，注意删去前要保证其存在。

void LRUReplacer::pin(frame_id_t frame_id) {
    std::scoped_lock lock{latch_};
    if (LRUhash_.count(frame_id)) {
        auto it = LRUhash_[frame_id];
        LRUlist_.erase(it);
        LRUhash_.erase(frame_id);
    }
}

unpin 实现：

用于解除对一个帧的固定，实现上很简单，但有一个让我疑惑的点就是多次 unpin 同一个帧时居然不需要重新放在链头，只是单纯的将其忽视（没办法，测试样例是这样写的）。

void LRUReplacer::unpin(frame_id_t frame_id) {
    std::scoped_lock lock{latch_};
    if (!LRUhash_.count(frame_id)) {
        LRUlist_.push_front(frame_id);
        LRUhash_[frame_id] = LRUlist_.begin();
    }
}

1.3 缓冲池管理器

本任务要求补全BufferPoolManager类，其负责管理缓冲池中的页面与磁盘文件中的页面之间的来回移动。

相比于前两个任务，这一关多了一个难点就是处理并发， Todo 指明了两个加锁的地方，虽然我目前不是特别理解其中的逻辑，并且似乎只在哪个位置加锁还是会出现并发问题。 所以选择了一个无脑的方法：就是把每个作为上层接口的函数都加上锁。

前置知识： 最好熟悉并发和加锁的逻辑，否则只能像我一样无脑加锁，然后需要了解 PageId 这个类对应于文件中具体某一页， 而 Page 这个类对应于缓冲池的一帧，所以 Page 会有 PageId 类成员用于标记该帧具体是哪个文件的哪个页。 最后是本关最后一个测试点需要跑 6s ，所以不需要着急。

find_victim_page 实现：

需要我们实现返回一个缓冲池中的可用帧的函数，思路很简单，跟着 Todo 做即可。

bool BufferPoolManager::find_victim_page(frame_id_t* frame_id) {
    if (free_list_.empty()) {
        // 缓冲池已满
        return replacer_->victim(frame_id);
    } else {
        *frame_id = free_list_.front();
        free_list_.pop_front();
    }
    return true;
}

update_page 实现：

实现一个替换可用帧的函数，通常调用了 find_victim_page 函数后就需要调用此函数。

void BufferPoolManager::update_page(Page* page, PageId new_page_id, frame_id_t new_frame_id) {
    if (page->is_dirty()) {
        disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE);
        page->is_dirty_ = false;
    }

    page_table_.erase(page->get_page_id());
    page_table_[new_page_id] = new_frame_id;

    page->reset_memory();
    page->id_ = new_page_id;
}

以上两个均属于辅助函数，减少重复的逻辑。

fetch_page 实现：

获取需要的页，类似于访问磁盘某页的数据，那我自然需要把它放到缓冲池，涉及到的逻辑有点多， 跟着 Todo 做不难。

Page* BufferPoolManager::fetch_page(PageId page_id) {
    std::scoped_lock lock{latch_};
    Page* targetPage = nullptr;

    if (page_table_.count(page_id)) {
        auto fid = page_table_[page_id];
        replacer_->pin(fid);
        targetPage = &(pages_[fid]);
        targetPage->pin_count_++;
    }
    else {
        frame_id_t fid(INVALID_FRAME_ID);
        if (!find_victim_page(&fid)) return nullptr;
        targetPage = &(pages_[fid]);
        update_page(targetPage, page_id, fid);
        disk_manager_->read_page(page_id.fd, page_id.page_no, targetPage->data_, PAGE_SIZE);
        replacer_->pin(fid);
        targetPage->pin_count_ = 1;
    }
    return targetPage;
}

unpin_page 实现：

该函数用于取消某一帧的一次固定，类似于修改完某页的操作，所以当 is_dirty == true ，也就是只要有一次修改了帧， 都需要将该帧标记为脏。

bool BufferPoolManager::unpin_page(PageId page_id, bool is_dirty) {

    std::scoped_lock lock{latch_};

    if (!page_table_.count(page_id)) return false;
    auto fid = page_table_[page_id];
    Page* targetPage = &(pages_[fid]);

    if (targetPage->pin_count_ == 0) return false;

    targetPage->pin_count_--;
    if (targetPage->pin_count_ == 0) {
        replacer_->unpin(fid);
    }

    if (is_dirty) {
        targetPage->is_dirty_ = true;
    }
    return true;
}

flush_page 实现：

该函数用于将当前帧的数据无条件地写回磁盘，注意写回磁盘后 is_dirty 需要标记为 false 。

bool BufferPoolManager::flush_page(PageId page_id) {
    std::scoped_lock lock{latch_};

    if (!page_table_.count(page_id)) {
        return false;
    }
    frame_id_t fid = page_table_[page_id];
    Page* page = &(pages_[fid]);

    disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE);
    page->is_dirty_ = false;

    return true;
}

new_page 实现：

该函数用于分配文件的一个新的页并访问，所以要调用之前实现的接口 allocate_page ，其余按照 Todo 完成问题不大。

Page* BufferPoolManager::new_page(PageId* page_id) {
    std::scoped_lock lock{latch_};
    frame_id_t fid(INVALID_FRAME_ID);

    if (!find_victim_page(&fid)) return nullptr;

    page_id->page_no = disk_manager_->allocate_page(page_id->fd);

    Page* page = &(pages_[fid]);

    update_page(page, *page_id, fid);

    replacer_->pin(fid);
    page->pin_count_ = 1;

    return page;
}

delete_page 实现：

逻辑上按照 Todo 去实现问题不大，最容易被忽视的地方在于进行删除该页时也算访问操作，所以必须固定避免并发冲突。 当然注意一种情况需要 unpin 。

bool BufferPoolManager::delete_page(PageId page_id) {
    std::scoped_lock lock{latch_};
    if (!page_table_.count(page_id)) {
        return true;
    }

    auto fid = page_table_[page_id];
    replacer_->pin(fid);

    Page* page = &pages_[fid];
    if (page->pin_count_ != 0) {
        replacer_->unpin(fid);
        return false;
    }

    disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE);
    page_table_.erase(page_id);
    free_list_.push_back(fid);

    return true;
}

flush_all_pages 实现：

直接 for 一遍即可，当然我也不确定中间判断 page_no 的条件必不必要，反正加不加都能过测试。

void BufferPoolManager::flush_all_pages(int fd) {
    for (size_t i = 0; i < pool_size_; i++) {
        Page* page = &pages_[i];
        if (page->id_.fd == fd && page->id_.page_no != INVALID_PAGE_ID) {
            disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE);
            page->is_dirty_ = false;
        }
    }
}

任务二 记录管理器

2.1 记录操作

本任务要求补全RMFileHandle类，其负责对文件中的记录进行操作。

前置知识： 完成代码前了解一点该数据库的存储结构大有帮助。首先每张表都有一个表文件（RMFileHandle），但存储的基本单位是页（RmPageHandle）， 每条记录就存储于页上，并且第一个页不存储记录，用于记录表的相关信息，未满的页彼此之间会构成一张支持头插法的链表（其实是个栈），第一个页中会记录链表头， 存储记录的页在头部会先维护链表的后继和该页具体信息（比如该页已经写了多少条记录以及哪些位置是有效的记录），剩余的位置才用于存放具体的记录。 更具体地，在此处用到了 Bitmap 这个类记录哪些位置属于有效的记录，我们可以调用它实现了静态函数完成我们想要的功能。

注意实现： Todo 的提示里似乎完全没有提示取了页后的 unpin 操作，虽然不写也能过测试，但仔细分析下还使得加上的。

get_record 实现：

给定记录的相关信息（哪页的哪条），获取该记录的具体值。直接根据 Todo 调用相关接口即可，当然这里类很多，得花一点时间才能捋清楚 彼此之间的关系。

std::unique_ptr<RmRecord> RmFileHandle::get_record(const Rid& rid, Context* context) const {
    auto page_handle = fetch_page_handle(rid.page_no);
    auto ptr = std::make_unique<RmRecord>(page_handle.file_hdr->record_size, page_handle.get_slot(rid.slot_no));
    return ptr;
}

fetch_page_handle 实现：

辅助函数，作用是获取给定页面号的页面句柄。实现上可以直接已经写好的获取一个页面到缓冲池的函数。可以这样理解： 数据本身就是以二进制存放在文件页中，这个函数帮助我们取到缓冲区，同时给了解析这个文件页的方法。

RmPageHandle RmFileHandle::fetch_page_handle(int page_no) const {
    Page* page = buffer_pool_manager_->fetch_page({fd_, page_no});
    if (page == nullptr) {
        throw PageNotExistError("", page_no); 
    }
    return RmPageHandle(&file_hdr_, page);
}

create_new_page_handle 实现：

辅助函数，用于创建一个新的页面句柄。更新信息的时候需要注意是插入到链表头部，因为它对于每条记录的存放位置其实没有严格的顺序要求，所以能插进去就行。 然后就是注意区别 num_records_per_page 和 bitmap_size ，简单来说前者是准确的一个页面最多存放的记录数，而后者因为是字节实现，所以一定是 8 的倍数。 一些琐碎信息的维护可以参考代码。

当然我还个疑问，就是 page == nullptr 时怎么办？可能需要抛出个异常啥的，不过没写也能过测试。

RmPageHandle RmFileHandle::create_new_page_handle() {
    PageId page_id = {fd_, INVALID_PAGE_ID};
    Page* page = buffer_pool_manager_->new_page(&page_id);

    RmPageHandle page_handle(&file_hdr_, page);
    Bitmap::init(page_handle.bitmap, file_hdr_.num_records_per_page);

    file_hdr_.num_pages++;
    page_handle.page_hdr->next_free_page_no = file_hdr_.first_free_page_no;  // 然后 -1 就当成终止节点了
    page_handle.page_hdr->num_records = 0;

    file_hdr_.first_free_page_no = page_id.page_no;
    return page_handle;
}

create_page_handle 实现：

辅助函数，虽然名字是 create ，但用于创建或获取一个空闲的页面句柄。实现就是判断链表中还有没有未满的，如果有就用第一个，反之就重新创建一个。

RmPageHandle RmFileHandle::create_page_handle() {
    if (file_hdr_.first_free_page_no != -1) {
        return fetch_page_handle(file_hdr_.first_free_page_no);
    } else
        return create_new_page_handle();
}

release_page_handle 实现：

辅助函数，用于当一个页面从满变为空闲时对链表的维护，其实就是插进链表头就可以。

void RmFileHandle::release_page_handle(RmPageHandle& page_handle) {
    page_handle.page_hdr->next_free_page_no = file_hdr_.first_free_page_no;
    file_hdr_.first_free_page_no = page_handle.page->get_page_id().page_no;
}

insert_record 实现：

用于向表中不指定位置插入一条数据。思路大概就是找个空闲的页，然后找个空闲的位置放进去就行，可以直接调用 Bitmap 类中已经实现好的函数，注意如果放进去后刚好满了要更新链表。 这里其实就可以看到栈的好处，满了直接弹出来。

Rid RmFileHandle::insert_record(char* buf, Context* context) {

    auto page_handle = create_page_handle();

    int slot_no = Bitmap::first_bit(0, page_handle.bitmap, file_hdr_.num_records_per_page);
    int page_no = page_handle.page->get_page_id().page_no; 

    char* target = page_handle.get_slot(slot_no);
    memcpy(target, buf, page_handle.file_hdr->record_size);
    Bitmap::set(page_handle.bitmap, slot_no);

    page_handle.page_hdr->num_records++;

    if (page_handle.page_hdr->num_records == file_hdr_.num_records_per_page) {
        file_hdr_.first_free_page_no = page_handle.page_hdr->next_free_page_no;
    }

    buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), true);
    return Rid{page_no, slot_no};
}

insert_record 实现：

用于在指定位置插入一条记录。不知道是个什么玩意，很怪，比如这是不是覆盖了一条原先已经存在的记录， 如果不是，那要不要考虑刚好满的情况。然后发现根本不在要求里面，不写也能通过测试。

void RmFileHandle::insert_record(const Rid& rid, char* buf) {

}

delete_record 实现：

用于删除指定位置的某条记录。实现上直接把标记改为 0 即可，注意刚好由满到空闲的额外处理。

void RmFileHandle::delete_record(const Rid& rid, Context* context) {
    auto page_handle = fetch_page_handle(rid.page_no);

    Bitmap::reset(page_handle.bitmap, rid.slot_no);

    if (page_handle.page_hdr->num_records == page_handle.file_hdr->num_records_per_page) {
        release_page_handle(page_handle);
    }
    page_handle.page_hdr->num_records--;

    buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), true);
}

update_record 实现：

用于更新一条记录。这个直接获取页面句柄，在指定位置去 memcpy 即可。

void RmFileHandle::update_record(const Rid& rid, char* buf, Context* context) {
    auto page_handle = fetch_page_handle(rid.page_no);
    memcpy(page_handle.get_slot(rid.slot_no), buf, page_handle.file_hdr->record_size);
    buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), true);
}

2.2 记录迭代器

本任务要求补全RmScan类，其用于遍历文件中存放的记录。

本身其实就是实现一个迭代器，我们随意给出一个遍历顺序即可，理解了记录的存放形式就很简单了。

RmScan 实现：

构造函数，用于初始化一个迭代器。直接一个个页面扫过去找到第一个 1 即可。注意第一个页面不存放记录即可。

RmScan::RmScan(const RmFileHandle *file_handle) : file_handle_(file_handle) {

    int num_pages = file_handle_->file_hdr_.num_pages;
    int max_n = file_handle_->file_hdr_.num_records_per_page;

    for (int i = 1; i < num_pages; i++) {
        auto page_handle = file_handle_->fetch_page_handle(i);
        rid_.page_no = i;
        rid_.slot_no = Bitmap::first_bit(1, page_handle.bitmap, max_n);
        file_handle_->buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), false);
        if (rid_.slot_no != max_n) return;
    }

    rid_.page_no = RM_NO_PAGE;  // 作为非法情况
}

is_end 实现：

用于判断是否已经遍历完所有记录。可以自行选择终止标志，这里选择代码本身提供的 RM_NO_PAGE 。

bool RmScan::is_end() const {
    return rid_.page_no == RM_NO_PAGE;
}

next 实现：

用于寻找当前遍历到的记录的后继。其实就是顺着继续找就行，可能需要对当前记录所在页的情况进行下讨论。

void RmScan::next() {
    if (is_end()) return;

    int num_pages = file_handle_->file_hdr_.num_pages;
    int max_n = file_handle_->file_hdr_.num_records_per_page;

    auto page_handle = file_handle_->fetch_page_handle(rid_.page_no);
    rid_.slot_no = Bitmap::next_bit(1, page_handle.bitmap, max_n, rid_.slot_no);

    if (rid_.slot_no != max_n) return;
    for (int i = rid_.page_no + 1; i < num_pages; i++) {
        auto page_handle = file_handle_->fetch_page_handle(i);
        rid_.page_no = i;
        rid_.slot_no = Bitmap::first_bit(1, page_handle.bitmap, max_n);
        file_handle_->buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), false);
        if (rid_.slot_no != max_n) return;
    }

    rid_.page_no = RM_NO_PAGE;
    file_handle_->buffer_pool_manager_->unpin_page(page_handle.page->get_page_id(), false);
}