SSTable 文件格式 | leveldb 篇

/*****
 * table_format :
 *              <beginning_of_file>
 *              [data block 1]
 *              [data block 2]
 *              ...
 *              [data block n]
 *              [filter block]
 *              [metaindex block]
 *              [index block]
 *              [Footer]        (fixed size; starts at file_size - sizeof(Footer))
 *              <end_of_file>
 */

/***
 * buffer    : 保存要写入到磁盘的序列化字节流
 * restarts_ : 保存位于 buffer 的每个重置点的偏移位置
 */
void BlockBuilder::Add(const Slice& key, const Slice& value) {
  Slice last_key_piece(last_key_);     // 得到上次的key

  assert(!finished_);
  assert(counter_ <= options_->block_restart_interval);
  assert(buffer_.empty()  // No value
   || options_->comparator->Compare(key, last_key_piece) > 0);

  size_t shared = 0;

  // block_restart_interval 控制着重启点之间的距离
  if (counter_ < options_->block_restart_interval) {
    // 记录相同 key 的位置
    const size_t min_length = std::min(last_key_piece.size(), key.size());
    while ((shared < min_length) && (last_key_piece[shared] == key[shared])) {
      shared++;
    }
  } else {
    // Restart compression
    // 重启过程中 key 相当于置空 -> share=0
    restarts_.push_back(buffer_.size());  // 记录重启点的 buffer 偏移
    counter_ = 0;
  }
  const size_t non_shared = key.size() - shared;  // 非共享 key 的长度

  // 存储数据长度
  PutVarint32(&buffer_, shared);       // 写入共享 key 长度   -- 变长 32 位
  PutVarint32(&buffer_, non_shared);   // 写入非共享 key 长度 -- 变长 32 位
  PutVarint32(&buffer_, value.size()); // 写入 value 长度    -- 变长 32 位

  // 存储真实数据
  buffer_.append(key.data() + shared, non_shared);   // 非共享 key 部分的存储
  buffer_.append(value.data(), value.size());           // 数据存储

  // 更新 last_key 表示为  上次插入的key
  last_key_.resize(shared);
  last_key_.append(key.data() + shared, non_shared);

  assert(Slice(last_key_) == key);
  counter_++;  // 计数加一
}

Slice BlockBuilder::Finish() {
  // Append restart array
  for (size_t i = 0; i < restarts_.size(); i++) {
    PutFixed32(&buffer_, restarts_[i]);             // 填充所有的重启点到 buffer
  }
  PutFixed32(&buffer_, restarts_.size());           // 填充重启点的数量到 buffer
  finished_ = true;
  return Slice(buffer_);                            // 返回整个 buffer
}

/***
 * Bloom 过滤器构建过程 ：
 * keys_           :  将所有的key都平铺拼接在一起
 * start_          :  每个 key 的偏移位置
 * result          :  存放得到的过滤器结果， 具有多个过滤器
 * filter_offsets_ : 每个过滤器的偏移量
 */
void FilterBlockBuilder::GenerateFilter() {

  const size_t num_keys = start_.size(); // 获取 key 的数量

  if (num_keys == 0) {
    filter_offsets_.push_back(result_.size());
    return;
  }

  // Make list of keys from flattened key structure
  start_.push_back(keys_.size());  // 因为每次都是填充上一次的key, 所以这次记录总得key的长度

  // 提取出所有的 key 使用  vector 存储
  tmp_keys_.resize(num_keys);
  for (size_t i = 0; i < num_keys; i++) {
    const char* base = keys_.data() + start_[i];
    size_t length = start_[i + 1] - start_[i];
    tmp_keys_[i] = Slice(base, length);
  }

  // 存储上次 Bloom 过滤器写入后的数据大小
  filter_offsets_.push_back(result_.size());

  // 计算 Bloom 过滤器的值， 并追加到 result 中
  policy_->CreateFilter(&tmp_keys_[0], static_cast<int>(num_keys), &result_);

  // 清空 key
  tmp_keys_.clear();
  keys_.clear();
  start_.clear();
}

Slice FilterBlockBuilder::Finish() {

  // 如果有 key 没有还没有计算完bloomFilter 则触发计算
  if (!start_.empty()) {
    GenerateFilter();
  }

  // Append array of per-filter offsets
  const uint32_t array_offset = result_.size();
  for (size_t i = 0; i < filter_offsets_.size(); i++) {
    // 填充每个bloom过滤器的偏移
    PutFixed32(&result_, filter_offsets_[i]);
  }

  // 填充总的Bloom过滤器的长度
  PutFixed32(&result_, array_offset);
  // 填充 11
  result_.push_back(kFilterBaseLg);  // Save encoding parameter in result
  return Slice(result_);
}

void FilterBlockBuilder::StartBlock(uint64_t block_offset) {
  uint64_t filter_index = (block_offset / kFilterBase);  // 现在需要的 filter_offset 数量
  assert(filter_index >= filter_offsets_.size());

  // 存在的意义是为了多写几个无效的偏移
  // 然后用于快速基于 block_offset 快速定位 filter_offsets
  while (filter_index > filter_offsets_.size()) {
    GenerateFilter();
  }
}

bool FilterBlockReader::KeyMayMatch(uint64_t block_offset, const Slice& key) {

  // 基于 block_offset 定位到 offset_[index]
  uint64_t index = block_offset >> base_lg_;  // 等同于 block_offset/kFilterBase

  if (index < num_) {
    // 截取 Bloom 过滤器
    uint32_t start = DecodeFixed32(offset_ + index * 4);
    uint32_t limit = DecodeFixed32(offset_ + index * 4 + 4);

    if (start <= limit && limit <= static_cast<size_t>(offset_ - data_)) {
      Slice filter = Slice(data_ + start, limit - start);
      // 匹配查找
      return policy_->KeyMayMatch(key, filter);
    } else if (start == limit) {
      // Empty filters do not match any keys
      return false;
    }
  }
  return true;  // Errors are treated as potential matches
}