Tair LDB基于Prefixkey的范围查找性能优化项目之如何建立prefix bloomfilter

项目是按照“”的步骤一步步完成的，上次解决了如何获取key的prefix_size问题“”。今天来继续解决第二个问题。在提案中有以下描述：

提取到prefix_size信息后，我们对所有的keys实现prefix bloomfilter，为了实现简单，我们可以在原有的filter block里加入新的prefix bloomfilter，与现有的bloomfilter一起存放，方法也很简单，就是在原有的filter block building过程中（filter_block.cc），将key的prefix也当作普通的key一起加入filter中，两者的过滤算法也是一致的。

即解决如何建立prefix bloomfilter？
这里有个关键问题：如何通过prefix_size信息得到prefix key？

有些人可能会说，直接取完整key的前prefix_size个字节作为prefix key即可。
这样就大错特错了，前面也说过从用户输入的key到最后sst文件里存储的key期间经过了好几层编码包装，如果你直接取包装后的key的前prefix_size个字节那么取得的内容并不是真实的prefix key而只是一些辅助编码信息，那么ldb层的key到底经过了哪些编码包装呢？

经过具体的源码分析，得到最终存储的key格式为：

元信息（7B）| area信息（2B） | 真实的key | valueType/sequenceNumber（8B）

其中的元信息一共7B（LDB_KEY_META_SIZE），包括下面两个部分：

expired_time(4B) | bucket_number编码(3B)

由于真实key的前后编码字节数都是固定的，因此我们想要提取prefix key是非常容易的，prefix key的格式应该如下：

元信息（7B）| area信息（2B） | prefixkey | valueType/sequenceNumber（8B）

因此在提取出prefix_size信息后重新组成prefix key时不能直接取key的前prefix_size个字节，从上面的格式可以看出，组成prefix key就是简单的字符串提取操作。

下面是具体的prefix key提取过程(这里的key是Slice类型)：

    int prefix_size = get_prefix_size(value);
    if(prefix_size != 0) {
        const char* key_data = key.data();
        char* prefix_key = new char[7 + 2 + prefix_size + 8];
        memcpy(prefix_key, key_data, 7 + 2 + prefix_size);
        memcpy(prefix_key + 7 + 2 + prefix_size, key_data + key.size() - 8, 8);
    }

提取完成后，就可以建立我们的prefix bloomfilter，与原来的bloomfilter放在一起加入filter block中，即在table_builder.cc::Add()方法中加入我们的prefix bloomfilter，如下：

void TableBuilder::Add(const Slice& key, const Slice& value) {
  Rep* r = rep_;
  assert(!r->closed);
  if (!ok()) return;
  if (r->num_entries > 0) {
    assert(r->options.comparator->Compare(key, Slice(r->last_key)) > 0); 
  }

  if (r->pending_index_entry) {
    assert(r->data_block.empty());
    r->options.comparator->FindShortestSeparator(&r->last_key, key);
    std::string handle_encoding;
    r->pending_handle.EncodeTo(&handle_encoding);
    r->index_block.Add(r->last_key, Slice(handle_encoding));
    r->pending_index_entry = false;
  }

  if (r->filter_block != NULL) {
    r->filter_block->AddKey(key);
    // add prefix key into filter block.
    int prefix_size = r->options.prefix_fetcher->get_prefix_size(value);
    if(prefix_size != 0) {
        const char* key_data = key.data();
        int base_size = r->options.kLdbKeyBaseSize;
        char* prefix_key = new char[base_size + prefix_size + kInternalKeyBaseSize];
        memcpy(prefix_key, key_data, base_size + prefix_size);
        memcpy(prefix_key + base_size + prefix_size, key_data + key.size() - kInternalKeyBaseSize, kInternalKeyBaseSize);
        r->filter_block->AddKey(Slice(prefix_key, base_size + prefix_size + kInternalKeyBaseSize));
        delete[] prefix_key;
    }
  }

  r->last_key.assign(key.data(), key.size());
  r->num_entries++;
  r->data_block.Add(key, value);

  const size_t estimated_block_size = r->data_block.CurrentSizeEstimate();
  if (estimated_block_size >= r->options.block_size) {
    Flush();
  }
}

这里考虑到整体项目的规范性，将获取prefix size信息的工具类PrefixFetcher实例对象和编码长度信息放在option.h中，比如上面的kLdbKeyBaseSize包括上面的元信息大小和area信息大小，为9B，而kInternalKeyBaseSize为后面的valueType/sequenceNumber信息大小，为8B，提取出prefixkey（如果prefix size不为0）后直接加入filter block即可。