存储管理

这一个实验做得格外难受，一是实验文档和注释特别混乱，而且似乎有所残缺，总给人一种没写完的感觉，二是我以前并没有了解过 B+ 树的结构。但总归还是靠着 github 上前人的代码坚持下来了。

前置知识： 这份代码实现上为 B+ 树的每个节点（IxNodeHandle）都分配了一个页，用于存储相关信息。其中最核心的信息就是键值对，除此之外就是一些比较杂的比如标记是否为叶节点，是否为根节点等放在页首处的信息。其中键值对的键本质是一段定长的字符串，当然实际上是需要解析一个包含各种类的元组的，而值是类 Rid ，对于非叶子节点，键对应子树中所有键的最小值，值用于指向相应的儿子，因此只有 page_no 成员有用，而对于叶子节点，这就是实际上对应的记录。叶节点还需要额外维护前驱和后继，构成一个链表。对于这颗 B+ 树整体（IxIndexHandle），对应于一个索引文件，其一页用来维护 B+ 树本身的一些信息，比如索引各字段类型，第二页用来维护叶节点构成的链表的链表头。所以根节点的页面号初始化为 2 。对于非根节点的页面，其键值对总是需要在半满和全满之间（已经实现的函数可以获取这两者对应的具体值），这是维护的一个重心。

实验指导文档的最后说：进行测试前，学生还需自行完成 src/system/sm_manager.cpp 中的SmManager::create_index()函数，方可进行测试。但是这玩意其实并不是特别好写，所以我直接从 github 借用了别人的实现。

void SmManager::create_index(const std::string& tab_name, const std::vector<std::string>& col_names, Context* context) {
    TabMeta& tab = db_.get_table(tab_name);

    // Check if the index already exists
    if (ix_manager_->exists(tab_name, col_names)) {
        throw IndexExistsError(tab_name, col_names);
    }

    // Prepare column metadata for index creation
    std::vector<ColMeta> cols;
    for (const auto& col_name : col_names) {
        cols.push_back(*tab.get_col(col_name));
    }

    // Create the index
    ix_manager_->create_index(tab_name, cols);

    // Open the newly created index file
    auto ih = ix_manager_->open_index(tab_name, cols);

    // Calculate total length of index columns
    int col_total_length = 0;
    for (const auto& col : cols) {
        col_total_length += col.len;
    }

    // Insert all existing records into the new index
    auto file_handle = fhs_.at(tab_name).get();
    std::vector<char> key(col_total_length);  // Use vector to handle variable length safely
    for (RmScan scan(file_handle); !scan.is_end(); scan.next()) {
        auto record = file_handle->get_record(scan.rid(), context);
        int offset = 0;
        for (size_t i = 0; i < cols.size(); ++i) {
            std::memcpy(&key[offset], record.get()->data + cols[i].offset, cols[i].len);
            offset += cols[i].len;
        }
        ih->insert_entry(key.data(), scan.rid(), context->txn_);
    }
    // Update table metadata with new index information
    tab.indexes.push_back(IndexMeta{tab_name, col_total_length, static_cast<int>(cols.size()), cols});

    // Store index handler in manager
    ihs_.emplace(ix_manager_->get_index_name(tab_name, col_names), std::move(ih));

    // Persist metadata changes to disk
    flush_meta();
}

任务一 B+树的查找

1.1 结点内的查找

为了实现整个B+树的查找，首先需要实现B+树单个结点内部的查找。

lower_bound 实现：

用于在当前 node 中查找第一个大于等于 target 的 key_idx 。二分也好，直接遍历也好，都可以。因为大概 B+ 树的键值对数量不会很多。值得一提的是这个 key_idx 的取值范围是 [0,num_key) ，返回 num_key 的情况表示没有找到。

注意比较可以调用已经实现的函数 ix_compare 。

int IxNodeHandle::lower_bound(const char *target) const {
    int n = page_hdr->num_key, i = 0;
    for (; i < n; i++) {
        if (ix_compare(target, get_key(i), file_hdr->col_types_, file_hdr->col_lens_) <= 0) break;
    }
    return i;
}

upper_bound 实现：

用于在当前 node 中查找第一个大于 target 的 key_idx 。基本和前面类似，唯一需要注意的是最小值设置为 1 而非 0 ，这样可以减少后面的一些特判。

int IxNodeHandle::upper_bound(const char *target) const {
    int n = page_hdr->num_key, i = 1;
    for (; i < n; i++) {
        if (ix_compare(target, get_key(i), file_hdr->col_types_, file_hdr->col_lens_) < 0) break;
    }
    return i;
}

leaf_lookup 实现：

用于寻找叶子节点中 key 对应的记录，通过 value 传回上层，然后返回值表示是否存在该 key 。实现上看下代码就懂了。

bool IxNodeHandle::leaf_lookup(const char *key, Rid **value) {
    int i = lower_bound(key);
    if (i != page_hdr->num_key && ix_compare(key, get_key(i), file_hdr->col_types_, file_hdr->col_lens_) == 0) {
        *value = get_rid(i);
        return true;
    } else
        return false;
}

internal_lookup 实现：

用于为非叶子节点 key 所在的子树，这里就可以看出来 upper_bound 下界设为 -1 的好处。

page_id_t IxNodeHandle::internal_lookup(const char *key) {
    int i = upper_bound(key);
    return value_at(i - 1);
}

1.2 B+ 树的查找

注意此时的代码还没考虑任务三的并发，所以没有加锁。

find_leaf_page 实现：

用于查找指定键所在的叶子结点。

这里的第二个返回值是用于处理并发的，目前没啥用，思路就是调用之前写好的节点内部的寻找函数，需要注意的是我们取出节点对应的页后不要忘记 unpin 。当然函数中好像少了没找到的处理逻辑，估计是在上层进行判断。

std::pair<IxNodeHandle *, bool> IxIndexHandle::find_leaf_page(const char *key, Operation operation,
                                                              Transaction *transaction, bool find_first) {
    page_id_t idx = file_hdr_->root_page_;
    auto it = fetch_node(idx);

    while (!it->page_hdr->is_leaf) {
        idx = it->internal_lookup(key);
        buffer_pool_manager_->unpin_page(it->get_page_id(), false);
        it = fetch_node(idx);
    }

    return std::make_pair(it, false);
}

get_value 实现：

用于查找指定键在叶子结点中的对应的值。

这里就需要有判断是否找到的逻辑了，找得到的话就存入 *result 中，注意 unpin 就没啥大问题。

bool IxIndexHandle::get_value(const char *key, std::vector<Rid> *result, Transaction *transaction) {

    auto [it, _] = find_leaf_page(key, Operation::FIND, transaction, false);
    Rid *rid;
    bool find = it->leaf_lookup(key, &rid);
    if (find) (*result).push_back(*rid);
    buffer_pool_manager_->unpin_page(it->get_page_id(), false);
    return find;
}