项目是按照“Tair LDB基于Prefixkey的范围查找性能优化项目提议方案”的步骤一步步完成的，目前已经解决了前面两个问题：

• 如何获取key的prefix_size问题“Tair LDB基于Prefixkey的范围查找性能优化项目之如何提取key的prefix_size”。
• 如何建立prefix bloomfilter“Tair LDB基于Prefixkey的范围查找性能优化项目之如何建立prefix bloomfilter”

今天来继续解决最后一个关键问题。在提案中有以下描述：

1. 在get_range接口中，如果查找到了sstable这里（先查memtable和immutable memtable，两者没有磁盘IO操作），

（1）首先根据[pkey+skey,pkey+end]的范围查找可能的sstfiles。

（2）对于每一个file，对dataindex block里的信息继续进行范围查找，找到可能包含[pkey+skey,pkey+end]这个范围的blocks。

（3）在读每一个block之前，获取filter block中存储的filter，通过prefix的MayMatch方法判断该block是否包含前缀pkey，如果不包含，则直接跳过这个block，这样就通过prefix bloomfilter实现了block过滤，从而减少了不必要的磁盘IO操作。

从编码角度讲，这里有很多待解决的关键问题，下面列举几个：

如何正确定位“读每一个block之前”？
如何获取filter block中事先存储的prefix bloomfilter？
如何正确使用prefix bloomfilter？
……

首先解决的是第一个问题：get_range的过程中什么时候开始读block，我们要在读之前进行拦截，加上bloomfilter过滤代码。

由于get_range中key的查找遍历在存储层面上统一通过Iterator的方式处理。其查找过程为：memtable Iterator —> sstable Iterator —> block Iterator。我们需要在block Iterator之前使用prefix bloomfilter以实现block过滤。block Iterator的具体调用在sstable Iterator最后，由two_level_iterator.cc的Seek()函数实现，如下：

void TwoLevelIterator::Seek(const Slice& target) {
    index_iter_.Seek(target);                                                   
    InitDataBlock(target);
    if (data_iter_.iter() != NULL) {
        data_iter_.Seek(target);
    }
    SkipEmptyDataBlocksForward();
}

具体流程为：
a) index_iter_.Seek(),得到index_iter_.Value(),即key所在data的index信息data_block_handle_。
b) InitDataBlock()，根据index_block_handle_，调用hook函数，获得对应data的data_iter_。
c) data_iter_->Seek()，定位到要找的key。
d) SkipEmptyDataBlocksForward()，如果获得的data_iter是无效的，那么需要不断尝试下一个data并定位到其最开始(已经满足Seek条件)，直到找到合法的data。

一开始我的想法是：block Iterator的获得在b步骤中，我们需要在其获得之前使用prefix bloomfilter过滤一下，因此我在a和b步骤之间添加了下面过滤语句：

if (index_iter_.Valid()) {
    Slice handle_value = index_iter_.value();
    const char* key_data = target.data();
    const char *ptr = strchr(key_data + LDB_KEY_META_SIZE + 2, 0);
    int prefix_size = ptr + 1 -  key_data;
    char* prefix_key = new char[prefix_size + 8];
    memcpy(prefix_key, key_data, prefix_size);
    memcpy(prefix_key + prefix_size, key_data + target.size() - 8, 8);
    BlockHandle handle;
    if (filter_ != NULL && handle.DecodeFrom(&handle_value).ok() 
        && !filter_->KeyMayMatch(handle.offset(), Slice(prefix_key, prefix_size + 8))) {
        // Not found, prefix bloomfilter hit
    } else {
        // prefix bloomfilter miss
        InitDataBlock(); 
        ……
}

经过验证的确可以在某个时刻进行bloomfilter过滤，因为当prefix key不存在时，会发生“prefix bloomfilter hit”。但后来经过大量数据的测试发现有个问题：

• filter block时而获取到，时而获取不到，获取不到时还是会进行block reader。

这是什么原因呢？一开始这个问题我是这么想的：
TwoLevelIterator是通过NewTwoLevelIterator()构造的，而构造该Iterator的地方有两个：
（a）DBIter的构造过程里关于sstable的Iterator建立（Version::AddIterators()）
- level-0中所有sstable的iterator (TableCache::NewIterator(),作为单个sstable iterator的TwoLevelIterator)
- 每个非level-0的leve上sstable集合iterator(（VersionSet::NewConcatenatingIterator(), 作为sstable集合 iterator的TwoLevelIterator）

其中level-0上的TwoLevelIterator最终通过在Table::NewIterator()（调用NewTwoLevelIterator()返回）中构造，而Table中有filter可供使用，因此我可以在Table::NewIterator()中通过NewTwoLevelIterator()调用将filter参数传递给TwoLevelIterator，方便在Seek中使用。而对于非level-0的sstable构造出的所有TwoLevelIterator没有可用的filter 参数，因此在level-0上能够命中，在非level-0上却不能命中（因为获取不到filter）。

（b）Compaction过程：VersionSet::MakeInputIterator()
- 对level-0的每个sstable，构造出对应的iterator：TwoLevelIterator（TableCache::NewIterator()）。
- 对非level-0的sstable构造出sstable集合的iterator：TwoLevelIterator (NewTwoLevelIterator()) 与（a）同，需要解决同样的问题。

后来再仔细琢磨这个问题，发现自己理解有误，实际上level-0和非level-0上的TwoLevelIterator都调用了Table::NewIterator()，只是后一个调用的时间比较晚且是通过hook函数回调执行的，不容易发现。而这个hook函数的执行在TwoLevelIterator::InitDataBlock()方法中，从上面代码可以看出我的filter获取和过滤步骤都在InitDataBlock()方法之前进行，这时候可能还没获取到filter，当然会出现filter为NULL的情形了。那么就需要研究InitDataBlock()方法，并在hook函数执行之后根据返回值判断来获取相应的filter，该方法如下：

void TwoLevelIterator::InitDataBlock() {
  if (!index_iter_.Valid()) {
    SetDataIterator(NULL);
  } else {
    Slice handle = index_iter_.value();
    if (data_iter_.iter() != NULL && handle.compare(data_block_handle_) == 0) {
      // data_iter_ is already constructed with this iterator, so
      // no need to change anything
    } else {
      Iterator* iter = (*block_function_)(arg_, options_, handle);
      data_block_handle_.assign(handle.data(), handle.size());
      SetDataIterator(iter);
    }
  }
}

这里的block_function_就是具体要调用的hook函数，而且这里的block_function_并不是确定的某个函数，而是根据执行情况动态确定，经过调试发现block_function_主要调用了以下两个函数：

（1）table.cc中的Iterator* Table::BlockReader()方法，目的是为了根据index_block_handle_获取data的data_iter_，这一步的block reader是少不了的，因为你在用prefix bloomfilter过滤prefix key之前肯定要知道过滤的是哪个block，这个block的iterator即位置首先要找到，这就需要进行一次block reader。
（2）two_level_iterator.cc中的Iterator* NewTwoLevelIterator()方法，这是构造具体的TwoLevelIterator，block iterator就是通过TwoLevelIterator的相关函数来获得的，也是第一步要做的事情（没有TwoLevelIterator后面就不能获得data_iter_及进行查找key了）。

这里需要关心的就是当block_function_为NewTwoLevelIterator()方法时，在该方法执行过后我们就能获取filter了（从Table::NewIterator()传递过来的），而且filter的获取是在具体查找key（要进行block reader）之前，也就是可以在查找key之前进行prefix bloomfilter过滤了。这个过滤代码既可以放在InitDataBlock()方法之内也可以放在Seek()函数中InitDataBlock()方法之后。从之前写的过滤代码可以看出，如果放在Seek()中，很多操作都和InitDataBlock()方法中的操作重复了，比如判断index_iter_的有效性、获取handle等，重复执行这些操作必然影响效率。因此更好的做法是将filter的获取和过滤代码都放在InitDataBlock()方法里，如下所示：

void TwoLevelIterator::InitDataBlock(const Slice& target) {
  if (!index_iter_.Valid()) {
    SetDataIterator(NULL);
  } else {
    Slice handle = index_iter_.value();
    if (data_iter_.iter() != NULL && handle.compare(data_block_handle_) == 0) {
      // data_iter_ is already constructed with this iterator, so
      // no need to change anything
    } else {
      Iterator* iter = (*block_function_)(arg_, options_, handle);
      if(typeid(*iter) == typeid(TwoLevelIterator)) {
          filter_ = dynamic_cast<TwoLevelIterator*>(iter)->filter_;
      }
      if (options_.use_prefix_bloom) {
        BlockHandle block_handle;
        if (filter_ != NULL && block_handle.DecodeFrom(&handle).ok()) {
            const char* key_data = target.data();
            const char *ptr = strchr(key_data + 9, 0);
            int prefix_size = ptr + 1 -  key_data;
            if(prefix_size > 0) {
                char* prefix_key = new char[prefix_size + kInternalKeyBaseSize];
                memcpy(prefix_key, key_data, prefix_size);
                memcpy(prefix_key + prefix_size, key_data + target.size() - kInternalKeyBaseSize, kInternalKeyBaseSize);
                if(!filter_->KeyMayMatch(block_handle.offset(), Slice(prefix_key,   
                  prefix_size + kInternalKeyBaseSize))) {
                    // Not found. prefix bloomfilter hit...
                    SetDataIterator(NULL);
                    delete[] prefix_key;
                    return; 
                }
            }
        }
      }
      data_block_handle_.assign(handle.data(), handle.size());
      SetDataIterator(iter);
    }
  }
}

这里需要注意两个问题：
1. 在提取prefix key的过程中采取了一种取巧的方式而并不是通过PrefixFetcher类提取的。因为观察到真实key如果其中含有prefix key的话，那么prefix key和suffix key两部分不是直接连在一起的，而是通过一个‘\0‘连在一起的，因此我这里可以直接查找真实key中\0所在位置从而提取出\0之前的prefix key部分，这个方法可能不太好，但测试时没发现什么问题。

2. 这里的9之前讨论过，是元信息+area信息的长度，本类是写在option.h中，由于这里不能获取option（不能获取option也是上面为啥没有用PrefixFetcher提取prefix key的原因，因为PrefixFetcher类的工具也是写在option中），所以这里暂时直接用9代替，这也是个待改进的地方。