Rocksdb Iterator实现：从DBIter 到 TwoLevelIter 的漫长链路

ps：本文的基础迭代器设计 以及 相关代码 是基于rocksdb 6.4.6版本进行描述的

1. 迭代器简单介绍

使用Rocksdb 进行Scan的过程中 都会用到Rocksdb 的Iterator，当然在使用的过程中大家会发现格外的顺手，就像我们的STL标准库为每一个容器构造的迭代器Iterator一样，能够通过指针的地址自增去访问容器中的数据。

同样，rocksdb的迭代器也可以很方便的去访问db内部的数据。

rocksdb::Iterator* it = db->NewIterator(rocksdb::ReadOptions());
for (it -> Seek(start); it -> Valid() && it -> key().ToString() < end;it -> Next() ) {......std::cout << it -> key().ToString() << ": " << it -> Value().ToString() << std::endl;	 
}assert(it->status.ok()); // check iterator status for any errors found during scan

以上几行简单的代码，即能够实现一个[start, end)区间的数据遍历。

2. 迭代器用户态相关接口

rocksdb迭代器为用户提供了大量的便捷操作和接口访问方式

• NewIterator 创建一个迭代器，需要传入读配置项
• Seek 查找一个key
• SeekToFirst 迭代器移动到db的第一个key位置，一般用于顺序遍历整个db的所有key
• SeekToLast 迭代器移动到db的最后一个key位置， 一般用于反向遍历整个db的所有key
• SeekForPrev移动到当前key的上一个位置，一般用于遍历(limit, start]之间的key
• Next 迭代器移动到下一个key
• Prev迭代器移动到上一个key

3. 迭代器内部架构

对用户态表现的简单接口，在底层实现过程中是有代价的。
为什么简单呢？
因为rocksdb的基础组件包括Memtable, Immutable memtable, 大量的sstables， 迭代器需要在内存/磁盘 数据 之间进行移动，然而只需要使用统一的简单接口，不需要关注迭代器在查找内存数据还是磁盘数据，简单易用，C++的封装特性展现得淋漓尽致。

而底层实现的代价就是需要将用户接口到内部接口 整个链路串起来，并且这个过程中的每一个查找细节都要仔细雕琢，否则Scan的性能将会是LSM 最为明显的痛点从而丢失大量有SQL需求的用户（SQL中会有大量的范围查找）。

一个大佬想要修改之前迭代器反回状态有歧义的问题，然后提了一个PR https://github.com/facebook/rocksdb/pull/3810，由siying大佬亲自review，整个PR 对迭代器的修改过程所涉及的复杂程度让siying都震惊了。

如下图为rocksdb迭代器的架构图，其中包括迭代器之间的级联关系 以及 流程图形态的函数调用：

这里画的是主要的几个迭代器，还是能够很明显得看出来整个迭代器内部的复杂程度。
图中箭头指向的迭代器表示被包含，比如MergingIterator被DBIter包含，ArenaWrapperDBIter 属于分配内存的迭代器，所以使用虚线框起来。

剩下的一些线段上的函数调用，则是从某一个迭代器生成其他迭代器的函数逻辑。其中主体迭代器是MergingIterator，rocksdb内部一般IternalIterator 都是属于MergingIterator。

不同迭代器之间的关系可以这样做一个使用者层面的简单描述：

如上图，用户使用DBIter 查找三个用户key，iter -> Seek(key1)
这个操作在内部会交给InternalIterator类型的MergingIterator，MergingIterator会拿到已经解析好的internal_key： user_key=“key1”, seqno=10, Type=put。这样的InternalKey，后续更加底层的迭代器会拿着Internal_key和自己所查找的区域进行key的匹配，从而取到底层的value数据。

接下来将描述一下这几个迭代器在代码中是如何创建的？

4. 迭代器的入口实现

4.1 DBIter

我们通过db->NewIterator入口进入创建迭代器的逻辑，具体创建之前会拿到当前db最新的或者用户指定的一个snapshot (落到底层internal_key的话也就是上文中提到的seqno),保证后续的读取都只读取小于等于当前snapshot的目标key。

// rocksdb的快照读，读取小于等于snapshot的key
auto snapshot = read_options.snapshot != nullptr? read_options.snapshot->GetSequenceNumber(): versions_->LastSequence();
// 创建迭代器入口，这里会返回一个DBIter
result = NewIteratorImpl(read_options, cfd, snapshot, read_callback);

接下来通过DBImpl::NewIteratorImpl --> NewArenaWrappedDbIterator 来创建一个 ArenaWrappedDBIter，即一个用来进行空间分配的迭代器，后续InternalIterator相关的迭代器都需要通过arena优先分配迭代器所需空间。

ArenaWrappedDBIter* DBImpl::NewIteratorImpl(const ReadOptions& read_options,ColumnFamilyData* cfd,SequenceNumber snapshot,ReadCallback* read_callback,bool allow_blob,bool allow_refresh) {......// 构造一个arena迭代器，负责后续的 internal迭代器的空间分配// 内部会先创建一个Arena迭代器，再创建DBIter迭代器ArenaWrappedDBIter* db_iter = NewArenaWrappedDbIterator(env_, read_options, *cfd->ioptions(), sv->mutable_cf_options, snapshot,sv->mutable_cf_options.max_sequential_skip_in_iterations,sv->version_number, read_callback, this, cfd, allow_blob,((read_options.snapshot != nullptr) ? false : allow_refresh));// 构造internal 迭代器，包括一系列 MergingIterator: MemtableIter, LevelIter, TwoLevelIterInternalIterator* internal_iter =NewInternalIterator(read_options, cfd, sv, db_iter->GetArena(),db_iter->GetRangeDelAggregator(), snapshot);// 绑定db_iter和internal_iterdb_iter->SetIterUnderDBIter(internal_iter);...
}ArenaWrappedDBIter* NewArenaWrappedDbIterator(Env* env, const ReadOptions& read_options,const ImmutableCFOptions& cf_options,const MutableCFOptions& mutable_cf_options, const SequenceNumber& sequence,uint64_t max_sequential_skip_in_iterations, uint64_t version_number,ReadCallback* read_callback, DBImpl* db_impl, ColumnFamilyData* cfd,bool allow_blob, bool allow_refresh) {// 创建一个Arena迭代器ArenaWrappedDBIter* iter = new ArenaWrappedDBIter();// 为db迭代器分配空间，并创建db迭代器iter->Init(env, read_options, cf_options, mutable_cf_options, sequence,max_sequential_skip_in_iterations, version_number, read_callback,db_impl, cfd, allow_blob, allow_refresh);if (db_impl != nullptr && cfd != nullptr && allow_refresh) {iter->StoreRefreshInfo(read_options, db_impl, cfd, read_callback,allow_blob);}

4.2 MergingIterator

如上图，MergingIterator 是通过NewIternalIterator 创建的，创建的过程中主要是维护一个MergeIteratorBuilder

具体代码就是NewIteratorImpl函数中,调用的NewInternalIterator函数，同样这个函数中也会先通过Arena分配好迭代器需要的空间。
在NewInternalIterator 会先创建一个MergeIteratorBuilder，并依次创建后续的 memtable, rangetombstone,immutable memtable, LevelIterator, TwoLevelIterator等一系列迭代器。

InternalIterator* DBImpl::NewInternalIterator(const ReadOptions& read_options,ColumnFamilyData* cfd,SuperVersion* super_version,Arena* arena,RangeDelAggregator* range_del_agg,SequenceNumber sequence) {InternalIterator* internal_iter;assert(arena != nullptr);assert(range_del_agg != nullptr);// Need to create internal iterator from the arena.// 创建一个MergingIterMergeIteratorBuilder merge_iter_builder(&cfd->internal_comparator(), arena,!read_options.total_order_seek &&super_version->mutable_cf_options.prefix_extractor != nullptr);.....

MergingIterator 底层是通过最小堆 数据结构来维护的，可以通过MergeIteratorBuilder构造过程来看到：

MergingIterator(const InternalKeyComparator* comparator,InternalIterator** children, int n, bool is_arena_mode,bool prefix_seek_mode): is_arena_mode_(is_arena_mode),comparator_(comparator),current_(nullptr),direction_(kForward),minHeap_(comparator_),prefix_seek_mode_(prefix_seek_mode),pinned_iters_mgr_(nullptr) {children_.resize(n);// 将传入的元素添加到rocksdb自实现的autovector之中for (int i = 0; i < n; i++) {children_[i].Set(children[i]);}// 构建最小堆// 堆顶元素是所有堆元素中的最小值for (auto& child : children_) {if (child.Valid()) {assert(child.status().ok());minHeap_.push(&child);} else {considerStatus(child.status());}}// 取堆顶的元素，表示当前迭代器所指向的keycurrent_ = CurrentForward();
}

这里只是初始化一个空的MerginIterator,里面并没有具体的key，后续在像range查找或者compaction 这样的过程中用到iterator的时候才会进行具体key元素的添加。

4.3 Memtable系列Iterator

回到 NewInternalIterator 函数，已经构造好了一个MergingIterator的merge_iter_builder，后续的所有迭代器都会被添加到这个builder之中，也就是数据的存储形态都会按照MergingIterator 的最小堆来进行存储。

我们知道rocksdb 的memtable是一种有序内存数据结构实现的（skiplist），memtable也有几种不同类型的：

• active memtable 是接受写请求，允许插入key-value数据的一个结构
• immutable memtable 是接受读请求的，且只读。主要用在flush过程，当active memtable被写满（达到write_buffer_size的限制）会切换为immutable memtable
• rangeTombstone memtbale 是存储rangetombstone数据的memtable，当上层用户通过DeleteRange接口下发一个范围删除的请求，会将tombstone信息放在这个memtable之中。

也就是我们在实际通过Iterator进行查找遍历的时候 这是三个memtable肯定是需要进行遍历的，也就是这三种memtable都需要各自维护一个iterator， 代码如下:

// 创建Memtable Iter，并添加到Merge_iter_builder之中
merge_iter_builder.AddIterator(super_version->mem->NewIterator(read_options, arena));
std::unique_ptr<FragmentedRangeTombstoneIterator> range_del_iter;
Status s;
if (!read_options.ignore_range_deletions) {// 创建range tombstone 迭代器 in memrange_del_iter.reset(super_version->mem->NewRangeTombstoneIterator(read_options, sequence));range_del_agg->AddTombstones(std::move(range_del_iter));
}
// Collect all needed child iterators for immutable memtables
if (s.ok()) {// 创建imm 迭代器super_version->imm->AddIterators(read_options, &merge_iter_builder);if (!read_options.ignore_range_deletions) {// 创建range tombstone 迭代器 in imms = super_version->imm->AddRangeTombstoneIterators(read_options, arena,range_del_agg);}
}

在memtable系列迭代器的底层移动是通过GetIterator函数访问 用户传入的memtable工厂对应的数据结构的迭代器：

GetIterator过程中会根据用户传入的lookahead(预读数据的大小) 来创建对应的SkipListRep 的迭代器，如果上层调用的next或者prev，到更加底层的数据结构中就是sikplist的next和prev了。

MemTableRep::Iterator* GetIterator(Arena* arena = nullptr) override {if (lookahead_ > 0) {void *mem =arena ? arena->AllocateAligned(sizeof(SkipListRep::LookaheadIterator)): operator new(sizeof(SkipListRep::LookaheadIterator));return new (mem) SkipListRep::LookaheadIterator(*this);} else {void *mem =arena ? arena->AllocateAligned(sizeof(SkipListRep::Iterator)): operator new(sizeof(SkipListRep::Iterator));return new (mem) SkipListRep::Iterator(&skip_list_);}
}

4.4 LevelIterator 和 TwoLevelIterator

创建完memtable系列的迭代器 就需要创建一系列sst上移动的迭代器。

如上图，可以看到rocksdb中on sst系列的迭代器主要维护了两种，第一种是Level1-Level N 的迭代器，第二种是Level0迭代器。

这里通过层来划分的原因主要是L0的sst文件之间会有重叠key，即sst之间不是有序的，所以查找的过程中对于L0，其所有的SST文件都需要被遍历到。
L1–LN 层的SST文件之间都是严格有序的，所以这一些层的迭代器只需要一种。

还是在NewInternalIterator函数中，创建完memtable系列的迭代器之后会通过current->AddIterators 中的AddIteratorsForLevel 函数，创建两种不同层的迭代器。

void Version::AddIteratorsForLevel(const ReadOptions& read_options,const EnvOptions& soptions,MergeIteratorBuilder* merge_iter_builder,int level,RangeDelAggregator* range_del_agg) {...// 为level0 创建其迭代器if (level == 0) {// Merge all level zero files together since they may overlapfor (size_t i = 0; i < storage_info_.LevelFilesBrief(0).num_files; i++) {const auto& file = storage_info_.LevelFilesBrief(0).files[i];merge_iter_builder->AddIterator(cfd_->table_cache()->NewIterator(read_options, soptions, cfd_->internal_comparator(),*file.file_metadata, range_del_agg,mutable_cf_options_.prefix_extractor.get(), nullptr,cfd_->internal_stats()->GetFileReadHist(0),TableReaderCaller::kUserIterator, arena,/*skip_filters=*/false, /*level=*/0,/*smallest_compaction_key=*/nullptr,/*largest_compaction_key=*/nullptr));}...}else if(storage_info_.LevelFilesBrief(level).num_files > 0) {// 创建大于level0 层的迭代器auto* mem = arena->AllocateAligned(sizeof(LevelIterator));merge_iter_builder->AddIterator(new (mem) LevelIterator(cfd_->table_cache(), read_options, soptions,cfd_->internal_comparator(), &storage_info_.LevelFilesBrief(level),mutable_cf_options_.prefix_extractor.get(), should_sample_file_read(),cfd_->internal_stats()->GetFileReadHist(level),TableReaderCaller::kUserIterator, IsFilterSkipped(level), level,range_del_agg, /*largest_compaction_key=*/nullptr));  }
}  

大于L0 的迭代器就是LevelIterator， 创建好之后在大于L0层的迭代器检索过程中会通过LevelIterator::Prev或者相关的其他接口进行查找。
当然其中有一个file_iter_数据成员 是实际的sst文件的iter，这个数据成员是在seek过程进行初始化的，将其绑定到具体的sst文件进行查找。

关于Level0 中sst文件的迭代器是通过TableCache::NewIterator函数中的table_reader->NewIterator创建的。之后会进入到我们默认配置的BlockBased::NewIterator 函数，当然如果这里不是使用sst，而是使用PlainTable或者CuckooTable这样的数据格式，那就是这一些table的迭代器了。

InternalIterator* BlockBasedTable::NewIterator(...) {...if (arena == nullptr) {return new BlockBasedTableIterator<DataBlockIter>(this, read_options, rep_->internal_comparator,NewIndexIterator(read_options,need_upper_bound_check &&rep_->index_type == BlockBasedTableOptions::kHashSearch,/*input_iter=*/nullptr, /*get_context=*/nullptr, &lookup_context),!skip_filters && !read_options.total_order_seek &&prefix_extractor != nullptr,need_upper_bound_check, prefix_extractor, BlockType::kData, caller,compaction_readahead_size);} ...
}

其中NewIndexIterator函数是我们要关注的，用来创建blockbased table的iterator，这个函数内部会使用index_reader->NewIterator函数来具体创建，创建的类型默认是BinarySearchIndexReader，我们这里使用比较典型有趣的设计ParttitionIndexReader，当然其底层也是BinarySearch的。

我们这里可以看PartitionIdexReader的NewIterator实现。

InternalIteratorBase<IndexValue>* NewIterator(const ReadOptions& read_options, bool /* disable_prefix_seek */,IndexBlockIter* iter, GetContext* get_context,BlockCacheLookupContext* lookup_context) override {const bool no_io = (read_options.read_tier == kBlockCacheTier);...if (!partition_map_.empty()) {// We don't return pinned data from index blocks, so no need// to set `block_contents_pinned`.// Two level iteratorit = NewTwoLevelIterator(new BlockBasedTable::PartitionedIndexIteratorState(table(),&partition_map_),index_block.GetValue()->NewIndexIterator(internal_comparator(), internal_comparator()->user_comparator(),nullptr, kNullStats, true, index_has_first_key(),index_key_includes_seq(), index_value_is_full()));...
}

这里TwoLevelIterator 就是在L0中，一个SST文件维护两个迭代器，一个迭代器用来构造IndexBlock的所以遍历，另一个迭代器用来实际的访问value数据即datablock。
NewTwoLevelIterator函数将构造好的NewIndexIterator作为参数传入之后并作为first_level_iter迭代器。
关于second_level_iter 迭代器是通过其seek 函数进行设置的:

void TwoLevelIndexIterator::Seek(const Slice& target) {first_level_iter_.Seek(target);InitDataBlock();if (second_level_iter_.iter() != nullptr) {second_level_iter_.Seek(target);}SkipEmptyDataBlocksForward();
}

其中InitDataBlock 中进行second_level_iter的创建

void TwoLevelIndexIterator::InitDataBlock() {if (!first_level_iter_.Valid()) {SetSecondLevelIterator(nullptr);} else {// index block中存放的是每隔restart bytes 的data block的起始地址BlockHandle handle = first_level_iter_.value().handle;if (second_level_iter_.iter() != nullptr &&!second_level_iter_.status().IsIncomplete() &&handle.offset() == data_block_handle_.offset()) {// second_level_iter is already constructed with this iterator, so// no need to change anything} else {// 通过从first_level_iter中取到的data 的起始位置作为hanle, 创建second_level_iterInternalIteratorBase<IndexValue>* iter =state_->NewSecondaryIterator(handle);data_block_handle_ = handle;SetSecondLevelIterator(iter);}}
}

到此，整个迭代器的创建过程基本说完了，在使用迭代器进行Seek/Next/Prev…etc 等操作的时候同样也是由dbiter开始，各个迭代器进行各自维护的组件中进行移动，最终将结果拿到MergingIterator的最小堆做完排序返回。

更具体的细节设计 后续会逐渐补充，毕竟迭代器组件事关引擎Scan性能，一点也不能马虎。

复杂的调用链中核心是简单的C++封装和动态绑定的特性，将内存到磁盘的数据结构穿起来，让整个Scan在不同组件之间并行去完成。